143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
"""Background watcher that turns a physical GPIO button press into a Qt signal.
|
|
|
|
A single :class:`GpioLineEventWatcher` is polled on a dedicated thread via
|
|
:func:`select.select`, woken either by a GPIO edge or by a self-pipe used for
|
|
clean shutdown. Because the watcher is a ``QObject``, its ``pressed`` signal is
|
|
delivered through the event loop on the thread that owns it (the GUI thread),
|
|
so connected slots may touch widgets exactly as a button click would.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import select
|
|
import threading
|
|
|
|
from PyQt6.QtCore import QObject, pyqtSignal
|
|
|
|
from python_app.hardware_full.switch_drivers.gpio_uapi import (
|
|
GPIO_V2_LINE_EVENT_FALLING_EDGE,
|
|
GPIO_V2_LINE_EVENT_RISING_EDGE,
|
|
GPIO_V2_LINE_FLAG_BIAS_DISABLED,
|
|
GPIO_V2_LINE_FLAG_BIAS_PULL_DOWN,
|
|
GPIO_V2_LINE_FLAG_BIAS_PULL_UP,
|
|
GPIO_V2_LINE_FLAG_EDGE_FALLING,
|
|
GPIO_V2_LINE_FLAG_EDGE_RISING,
|
|
GpioLineEventWatcher,
|
|
)
|
|
|
|
_BIAS_FLAGS = {
|
|
"pull_up": GPIO_V2_LINE_FLAG_BIAS_PULL_UP,
|
|
"pull_down": GPIO_V2_LINE_FLAG_BIAS_PULL_DOWN,
|
|
"disabled": GPIO_V2_LINE_FLAG_BIAS_DISABLED,
|
|
}
|
|
|
|
|
|
class ControlButtonWatcher(QObject):
|
|
"""Monitor a GPIO push-button on a background thread and emit ``pressed``.
|
|
|
|
The press is detected on a single edge (falling for active-low wiring,
|
|
rising otherwise), so a normal push produces exactly one ``pressed`` signal.
|
|
``failed`` reports an unrecoverable watcher error as a human-readable string.
|
|
"""
|
|
|
|
pressed = pyqtSignal()
|
|
failed = pyqtSignal(str)
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
chip: str,
|
|
pin: int,
|
|
active_low: bool = True,
|
|
bias: str = "",
|
|
debounce_ms: int = 50,
|
|
parent: QObject | None = None,
|
|
) -> None:
|
|
"""Configure the watcher; the GPIO line stays closed until :meth:`start`."""
|
|
super().__init__(parent)
|
|
active_low = bool(active_low)
|
|
# Active-low wiring idles high and falls on press; active-high is the mirror.
|
|
self._press_edge = (
|
|
GPIO_V2_LINE_EVENT_FALLING_EDGE if active_low else GPIO_V2_LINE_EVENT_RISING_EDGE
|
|
)
|
|
edge_flag = (
|
|
GPIO_V2_LINE_FLAG_EDGE_FALLING if active_low else GPIO_V2_LINE_FLAG_EDGE_RISING
|
|
)
|
|
self._line = GpioLineEventWatcher(
|
|
chip,
|
|
int(pin),
|
|
edge_flags=edge_flag,
|
|
bias_flags=self._resolve_bias_flags(bias, active_low),
|
|
debounce_us=max(0, int(debounce_ms)) * 1000,
|
|
consumer="radar_control_button",
|
|
)
|
|
self._thread: threading.Thread | None = None
|
|
self._stop_read_fd = -1
|
|
self._stop_write_fd = -1
|
|
|
|
@staticmethod
|
|
def _resolve_bias_flags(bias: str, active_low: bool) -> int:
|
|
"""Return GPIO bias flags, defaulting to the bias that matches the wiring."""
|
|
name = (bias or "").strip().lower()
|
|
if name in _BIAS_FLAGS:
|
|
return _BIAS_FLAGS[name]
|
|
return GPIO_V2_LINE_FLAG_BIAS_PULL_UP if active_low else GPIO_V2_LINE_FLAG_BIAS_PULL_DOWN
|
|
|
|
def start(self) -> None:
|
|
"""Open the GPIO line and begin watching for presses on a background thread."""
|
|
# Guard against double-start: a second start would leak the first line/pipe/thread.
|
|
if self._thread is not None:
|
|
return
|
|
try:
|
|
self._line.open()
|
|
self._stop_read_fd, self._stop_write_fd = os.pipe()
|
|
self._thread = threading.Thread(
|
|
target=self._run, name="control-button-watcher", daemon=True
|
|
)
|
|
self._thread.start()
|
|
except Exception:
|
|
# Release any line/pipe fds opened before the failure so nothing leaks
|
|
# and no orphaned thread survives a partial start.
|
|
self._thread = None
|
|
self._close_stop_pipe()
|
|
self._line.close()
|
|
raise
|
|
|
|
def stop(self) -> None:
|
|
"""Signal the watcher thread to exit and release the GPIO line and pipe."""
|
|
if self._stop_write_fd >= 0:
|
|
try:
|
|
os.write(self._stop_write_fd, b"\x00")
|
|
except OSError:
|
|
pass
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=2.0)
|
|
self._thread = None
|
|
self._close_stop_pipe()
|
|
self._line.close()
|
|
|
|
def _run(self) -> None:
|
|
"""Block on the GPIO line until an edge fires or shutdown is requested."""
|
|
line_fd = self._line.fileno()
|
|
try:
|
|
while True:
|
|
readable, _, _ = select.select([line_fd, self._stop_read_fd], [], [])
|
|
if self._stop_read_fd in readable:
|
|
return
|
|
if line_fd in readable and self._line.read_event() == self._press_edge:
|
|
self.pressed.emit()
|
|
except Exception as exc: # noqa: BLE001
|
|
self.failed.emit(str(exc))
|
|
|
|
def _close_stop_pipe(self) -> None:
|
|
"""Close both ends of the self-pipe used to wake the watcher thread."""
|
|
for attr in ("_stop_read_fd", "_stop_write_fd"):
|
|
fd = getattr(self, attr)
|
|
if fd >= 0:
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
setattr(self, attr, -1)
|