radar_system/python_app/hardware_full/switch_drivers/gpio_uapi.py

"""Minimal Linux GPIO v2 UAPI wrapper for output lines and input edge events."""

from __future__ import annotations

import ctypes
import errno
import fcntl
import os
from typing import Sequence


GPIO_MAX_NAME_SIZE = 32
GPIO_V2_LINES_MAX = 64
GPIO_V2_LINE_NUM_ATTRS_MAX = 10
GPIO_V2_LINE_FLAG_INPUT = 1 << 2
GPIO_V2_LINE_FLAG_OUTPUT = 1 << 3
GPIO_V2_LINE_FLAG_EDGE_RISING = 1 << 4
GPIO_V2_LINE_FLAG_EDGE_FALLING = 1 << 5
GPIO_V2_LINE_FLAG_BIAS_PULL_UP = 1 << 8
GPIO_V2_LINE_FLAG_BIAS_PULL_DOWN = 1 << 9
GPIO_V2_LINE_FLAG_BIAS_DISABLED = 1 << 10

GPIO_V2_LINE_ATTR_ID_DEBOUNCE = 3

GPIO_V2_LINE_EVENT_RISING_EDGE = 1
GPIO_V2_LINE_EVENT_FALLING_EDGE = 2

_IOC_NRBITS = 8
_IOC_TYPEBITS = 8
_IOC_SIZEBITS = 14
_IOC_DIRBITS = 2

_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS

_IOC_WRITE = 1
_IOC_READ = 2


def _ioc(direction: int, ioc_type: int, number: int, size: int) -> int:
    """Build raw ioctl command number."""
    return (
        (direction << _IOC_DIRSHIFT)
        | (ioc_type << _IOC_TYPESHIFT)
        | (number << _IOC_NRSHIFT)
        | (size << _IOC_SIZESHIFT)
    )


def _iowr(ioc_type: int, number: int, struct_type: type[ctypes.Structure]) -> int:
    """Build read-write ioctl number for provided structure."""
    return _ioc(_IOC_READ | _IOC_WRITE, ioc_type, number, ctypes.sizeof(struct_type))


class GpioV2LineAttribute(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_attribute`."""

    _fields_ = [
        ("id", ctypes.c_uint32),
        ("padding", ctypes.c_uint32),
        ("value", ctypes.c_uint64),
    ]


class GpioV2LineConfigAttribute(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_config_attribute`."""

    _fields_ = [
        ("attr", GpioV2LineAttribute),
        ("mask", ctypes.c_uint64),
    ]


class GpioV2LineConfig(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_config`."""

    _fields_ = [
        ("flags", ctypes.c_uint64),
        ("num_attrs", ctypes.c_uint32),
        ("padding", ctypes.c_uint32 * 5),
        ("attrs", GpioV2LineConfigAttribute * GPIO_V2_LINE_NUM_ATTRS_MAX),
    ]


class GpioV2LineRequest(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_request`."""

    _fields_ = [
        ("offsets", ctypes.c_uint32 * GPIO_V2_LINES_MAX),
        ("consumer", ctypes.c_char * GPIO_MAX_NAME_SIZE),
        ("config", GpioV2LineConfig),
        ("num_lines", ctypes.c_uint32),
        ("event_buffer_size", ctypes.c_uint32),
        ("padding", ctypes.c_uint32 * 5),
        ("fd", ctypes.c_int32),
    ]


class GpioV2LineValues(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_values`."""

    _fields_ = [
        ("bits", ctypes.c_uint64),
        ("mask", ctypes.c_uint64),
    ]


class GpioV2LineEvent(ctypes.Structure):
    """ctypes mapping of `gpio_v2_line_event`."""

    _fields_ = [
        ("timestamp_ns", ctypes.c_uint64),
        ("id", ctypes.c_uint32),
        ("offset", ctypes.c_uint32),
        ("seqno", ctypes.c_uint32),
        ("line_seqno", ctypes.c_uint32),
        ("padding", ctypes.c_uint32 * 6),
    ]


GPIO_V2_GET_LINE_IOCTL = _iowr(0xB4, 0x07, GpioV2LineRequest)
GPIO_V2_LINE_SET_VALUES_IOCTL = _iowr(0xB4, 0x0F, GpioV2LineValues)


class GpioOutputLines:
    """Open and control one or more GPIO output lines as a single request."""

    def __init__(self, chip: str, offsets: Sequence[int], consumer: str) -> None:
        """Build GPIO line request descriptor."""
        if not chip:
            raise ValueError("gpio chip path must not be empty")
        if not offsets:
            raise ValueError("at least one GPIO line offset is required")
        if len(offsets) > GPIO_V2_LINES_MAX:
            raise ValueError(f"too many GPIO offsets requested: {len(offsets)}")

        normalized_offsets = [int(offset) for offset in offsets]
        if any(offset < 0 for offset in normalized_offsets):
            raise ValueError("GPIO offsets must be non-negative")
        if len(set(normalized_offsets)) != len(normalized_offsets):
            raise ValueError("GPIO offsets must be unique")

        self._chip = chip
        self._offsets = normalized_offsets
        self._consumer = (consumer or "radar_switch").encode("ascii", errors="ignore")[: GPIO_MAX_NAME_SIZE - 1]

        self._chip_fd = -1
        self._line_fd = -1

    def open(self) -> None:
        """Open GPIO chip and request configured output lines."""
        if self._line_fd >= 0:
            return

        try:
            self._chip_fd = os.open(self._chip, os.O_RDONLY | os.O_CLOEXEC)
        except OSError as exc:
            raise RuntimeError(f"Failed to open GPIO chip '{self._chip}': {exc}") from exc

        request = GpioV2LineRequest()
        for index, offset in enumerate(self._offsets):
            request.offsets[index] = ctypes.c_uint32(offset).value

        request.num_lines = ctypes.c_uint32(len(self._offsets)).value
        request.config.flags = ctypes.c_uint64(GPIO_V2_LINE_FLAG_OUTPUT).value
        request.consumer = self._consumer

        try:
            fcntl.ioctl(self._chip_fd, GPIO_V2_GET_LINE_IOCTL, request)
        except OSError as exc:
            self._close_chip_fd()
            raise RuntimeError(f"Failed to request GPIO lines on '{self._chip}': {exc}") from exc

        if request.fd < 0:
            self._close_chip_fd()
            raise RuntimeError(f"GPIO line request returned invalid fd for '{self._chip}'")

        self._line_fd = int(request.fd)

    def close(self) -> None:
        """Close line request and chip file descriptors (best-effort, idempotent)."""
        try:
            self._close_line_fd()
        finally:
            # Ensure the chip fd is always closed even if closing the line fd raised.
            self._close_chip_fd()

    def set_values(self, values: Sequence[int]) -> None:
        """Apply output values for all requested lines."""
        if self._line_fd < 0:
            raise RuntimeError("GPIO line request is not open")
        if len(values) != len(self._offsets):
            raise ValueError(
                f"GPIO values length mismatch: expected {len(self._offsets)}, got {len(values)}"
            )

        bits = 0
        for index, value in enumerate(values):
            normalized = int(value)
            if normalized not in (0, 1):
                raise ValueError(f"GPIO output value must be 0 or 1, got {value}")
            if normalized == 1:
                bits |= (1 << index)

        mask = (1 << len(self._offsets)) - 1
        line_values = GpioV2LineValues(bits=ctypes.c_uint64(bits).value, mask=ctypes.c_uint64(mask).value)

        try:
            fcntl.ioctl(self._line_fd, GPIO_V2_LINE_SET_VALUES_IOCTL, line_values)
        except OSError as exc:
            if exc.errno == errno.ENODEV:
                raise RuntimeError("GPIO device disconnected") from exc
            raise RuntimeError(f"Failed to set GPIO output values: {exc}") from exc

    def _close_line_fd(self) -> None:
        """Close line file descriptor if currently open (best-effort, idempotent)."""
        if self._line_fd >= 0:
            try:
                os.close(self._line_fd)
            finally:
                # Always clear the fd so close() stays idempotent even if os.close raised.
                self._line_fd = -1

    def _close_chip_fd(self) -> None:
        """Close chip file descriptor if currently open (best-effort, idempotent)."""
        if self._chip_fd >= 0:
            try:
                os.close(self._chip_fd)
            finally:
                # Always clear the fd so close() stays idempotent even if os.close raised.
                self._chip_fd = -1


class GpioLineEventWatcher:
    """Watch a single GPIO input line for edge events via Linux GPIO v2 UAPI.

    The line file descriptor returned by the kernel becomes readable whenever a
    requested edge occurs; each read yields exactly one ``gpio_v2_line_event``.
    Callers drive the wait loop themselves (e.g. with :func:`select.select`)
    using :meth:`fileno`, which keeps this wrapper free of any threading or
    polling policy.
    """

    def __init__(
        self,
        chip: str,
        offset: int,
        *,
        edge_flags: int,
        bias_flags: int = 0,
        debounce_us: int = 0,
        consumer: str = "radar_input",
    ) -> None:
        """Build an input line request descriptor for edge detection."""
        if not chip:
            raise ValueError("gpio chip path must not be empty")
        if offset < 0:
            raise ValueError("GPIO offset must be non-negative")
        if not edge_flags:
            raise ValueError("at least one edge flag is required")

        self._chip = chip
        self._offset = int(offset)
        self._flags = GPIO_V2_LINE_FLAG_INPUT | int(edge_flags) | int(bias_flags)
        self._debounce_us = max(0, int(debounce_us))
        self._consumer = (consumer or "radar_input").encode("ascii", errors="ignore")[: GPIO_MAX_NAME_SIZE - 1]

        self._chip_fd = -1
        self._line_fd = -1

    def open(self) -> None:
        """Open GPIO chip and request the configured input line with edge events."""
        if self._line_fd >= 0:
            return

        try:
            self._chip_fd = os.open(self._chip, os.O_RDONLY | os.O_CLOEXEC)
        except OSError as exc:
            raise RuntimeError(f"Failed to open GPIO chip '{self._chip}': {exc}") from exc

        request = GpioV2LineRequest()
        request.offsets[0] = ctypes.c_uint32(self._offset).value
        request.num_lines = ctypes.c_uint32(1).value
        request.config.flags = ctypes.c_uint64(self._flags).value
        request.consumer = self._consumer

        if self._debounce_us > 0:
            config_attr = request.config.attrs[0]
            config_attr.attr.id = ctypes.c_uint32(GPIO_V2_LINE_ATTR_ID_DEBOUNCE).value
            config_attr.attr.value = ctypes.c_uint64(self._debounce_us).value
            config_attr.mask = ctypes.c_uint64(1).value  # applies to line index 0
            request.config.num_attrs = ctypes.c_uint32(1).value

        try:
            fcntl.ioctl(self._chip_fd, GPIO_V2_GET_LINE_IOCTL, request)
        except OSError as exc:
            self._close_chip_fd()
            raise RuntimeError(f"Failed to request GPIO line on '{self._chip}': {exc}") from exc

        if request.fd < 0:
            self._close_chip_fd()
            raise RuntimeError(f"GPIO line request returned invalid fd for '{self._chip}'")

        self._line_fd = int(request.fd)

    def fileno(self) -> int:
        """Return the line file descriptor for use with poll/select."""
        if self._line_fd < 0:
            raise RuntimeError("GPIO line request is not open")
        return self._line_fd

    def read_event(self) -> int:
        """Read one queued edge event and return its event id (rising/falling)."""
        if self._line_fd < 0:
            raise RuntimeError("GPIO line request is not open")

        size = ctypes.sizeof(GpioV2LineEvent)
        data = os.read(self._line_fd, size)
        if len(data) < size:
            raise RuntimeError(f"Short GPIO event read: expected {size} bytes, got {len(data)}")

        event = GpioV2LineEvent.from_buffer_copy(data)
        return int(event.id)

    def close(self) -> None:
        """Close line request and chip file descriptors (best-effort, idempotent)."""
        try:
            self._close_line_fd()
        finally:
            # Ensure the chip fd is always closed even if closing the line fd raised.
            self._close_chip_fd()

    def _close_line_fd(self) -> None:
        """Close line file descriptor if currently open (best-effort, idempotent)."""
        if self._line_fd >= 0:
            try:
                os.close(self._line_fd)
            finally:
                # Always clear the fd so close() stays idempotent even if os.close raised.
                self._line_fd = -1

    def _close_chip_fd(self) -> None:
        """Close chip file descriptor if currently open (best-effort, idempotent)."""
        if self._chip_fd >= 0:
            try:
                os.close(self._chip_fd)
            finally:
                # Always clear the fd so close() stays idempotent even if os.close raised.
                self._chip_fd = -1