some fixes

python_app/hardware_full/switch_drivers/gpio_uapi.py

@@ -1,4 +1,4 @@
-"""Minimal Linux GPIO v2 UAPI wrapper for output-only line control."""
+"""Minimal Linux GPIO v2 UAPI wrapper for output lines and input edge events."""
 
 from __future__ import annotations
 
@@ -12,7 +12,18 @@ 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
@@ -96,6 +107,19 @@ class GpioV2LineValues(ctypes.Structure):
     ]
 
 
+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)
 
@@ -198,3 +222,112 @@ class GpioOutputLines:
         if self._chip_fd >= 0:
             os.close(self._chip_fd)
             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."""
+        self._close_line_fd()
+        self._close_chip_fd()
+
+    def _close_line_fd(self) -> None:
+        """Close line file descriptor if currently open."""
+        if self._line_fd >= 0:
+            os.close(self._line_fd)
+            self._line_fd = -1
+
+    def _close_chip_fd(self) -> None:
+        """Close chip file descriptor if currently open."""
+        if self._chip_fd >= 0:
+            os.close(self._chip_fd)
+            self._chip_fd = -1