imonlinux
dc9049e899
Introduced a new ButtonController responsible for handling the ReSpeaker (or generic GPIO) button:
Uses RPi.GPIO with BCM numbering and internal pull-up (PUD_UP), assuming an active-low button.
Starts a dedicated polling thread that:
Reads the GPIO level at a configurable interval (default 10 ms / 100 Hz),
Detects press and release transitions,
Computes press duration to classify short vs long press.
Short press behavior:
If state.tts_player.is_playing or state.music_player.is_playing:
Schedules state.satellite.stop() on the asyncio loop,
Also stops music via state.music_player.stop().
Otherwise:
Schedules state.satellite.manual_wakeup("button") to start a new voice interaction (wake-like behavior without needing a model instance).
Long press behavior:
Toggles mic mute using the existing event path:
Publishes set_mic_mute on the EventBus with {"state": not state.mic_muted}.
This ensures MicMuteHandler remains the single writer for mic state, updates mic_muted_event, and syncs MQTT mute state/LEDs.
Designed to fail gracefully:
If ButtonConfig.enabled is False or RPi.GPIO is unavailable, logs a message and disables button support without affecting the rest of the system.
224 lines
7.5 KiB
Python
224 lines
7.5 KiB
Python
"""Hardware button controller (e.g. ReSpeaker 2-Mic HAT momentary switch).
|
|
|
|
This controller:
|
|
- Short press:
|
|
* If audio is playing (TTS or music): stop playback (like Stop wake word).
|
|
* Otherwise: start a new conversation (like a wake word).
|
|
- Long press:
|
|
* Toggle microphone mute (via the existing set_mic_mute event).
|
|
|
|
Implementation note:
|
|
- Uses a simple polling loop with RPi.GPIO instead of kernel edge detection.
|
|
- This avoids "Failed to add edge detection" errors on some HATs/overlays.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from .event_bus import EventBus
|
|
from .models import ServerState
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
try:
|
|
import RPi.GPIO as GPIO
|
|
except ImportError: # pragma: no cover - non-RPi hosts
|
|
GPIO = None
|
|
|
|
|
|
@dataclass
|
|
class ButtonRuntimeConfig:
|
|
"""Runtime-safe config wrapper for the hardware button."""
|
|
enabled: bool
|
|
pin: int
|
|
long_press_seconds: float
|
|
poll_interval_seconds: float = 0.01 # 100 Hz polling
|
|
|
|
|
|
class ButtonController:
|
|
"""
|
|
Hardware button handler for the ReSpeaker 2-Mic HAT (or any GPIO button).
|
|
|
|
This class:
|
|
- Uses a background polling thread to watch the GPIO level.
|
|
- Interprets level transitions as press/release.
|
|
- Schedules actions on the asyncio loop via loop.call_soon_threadsafe().
|
|
- Uses the EventBus to toggle mic mute.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
loop,
|
|
event_bus: EventBus,
|
|
state: ServerState,
|
|
config,
|
|
) -> None:
|
|
"""
|
|
:param loop: asyncio event loop (state.loop).
|
|
:param event_bus: Global EventBus.
|
|
:param state: Global ServerState.
|
|
:param config: Config.button (ButtonConfig-like object).
|
|
"""
|
|
self.loop = loop
|
|
self.event_bus = event_bus
|
|
self.state = state
|
|
|
|
self._cfg = ButtonRuntimeConfig(
|
|
enabled=getattr(config, "enabled", False),
|
|
pin=getattr(config, "pin", 17),
|
|
long_press_seconds=float(
|
|
getattr(config, "long_press_seconds", 1.0)
|
|
),
|
|
poll_interval_seconds=float(
|
|
getattr(config, "poll_interval_seconds", 0.01)
|
|
if hasattr(config, "poll_interval_seconds")
|
|
else 0.01
|
|
),
|
|
)
|
|
|
|
self._thread: threading.Thread | None = None
|
|
self._press_time: float | None = None
|
|
self._last_level: int | None = None
|
|
|
|
if not self._cfg.enabled:
|
|
_LOGGER.info("ButtonController disabled in config; not initializing GPIO")
|
|
return
|
|
|
|
if GPIO is None:
|
|
_LOGGER.warning(
|
|
"RPi.GPIO not available; hardware button support disabled"
|
|
)
|
|
return
|
|
|
|
try:
|
|
GPIO.setmode(GPIO.BCM)
|
|
# Assume button wired as active-low with internal pull-up (common for HATs)
|
|
GPIO.setup(self._cfg.pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)
|
|
self._last_level = GPIO.input(self._cfg.pin)
|
|
_LOGGER.info(
|
|
"ButtonController initialized on GPIO pin %s "
|
|
"(long_press_seconds=%.2f, poll_interval=%.3fs)",
|
|
self._cfg.pin,
|
|
self._cfg.long_press_seconds,
|
|
self._cfg.poll_interval_seconds,
|
|
)
|
|
except Exception:
|
|
_LOGGER.exception("Failed to configure GPIO pin %s for button", self._cfg.pin)
|
|
return
|
|
|
|
# Start polling thread
|
|
self._thread = threading.Thread(
|
|
target=self._poll_loop,
|
|
name="ButtonControllerThread",
|
|
daemon=True,
|
|
)
|
|
self._thread.start()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Polling loop
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _poll_loop(self) -> None:
|
|
"""Background loop to poll the GPIO level and detect presses."""
|
|
_LOGGER.debug("ButtonController polling thread started")
|
|
|
|
while not getattr(self.state, "shutdown", False):
|
|
try:
|
|
level = GPIO.input(self._cfg.pin)
|
|
except Exception:
|
|
_LOGGER.exception("Error reading GPIO pin %s", self._cfg.pin)
|
|
break
|
|
|
|
if self._last_level is None:
|
|
self._last_level = level
|
|
|
|
# Active-low: 0 = pressed, 1 = released
|
|
if level != self._last_level:
|
|
if level == 0:
|
|
self._on_pressed()
|
|
else:
|
|
self._on_released()
|
|
self._last_level = level
|
|
|
|
time.sleep(self._cfg.poll_interval_seconds)
|
|
|
|
_LOGGER.debug("ButtonController polling thread exiting")
|
|
# Do NOT call GPIO.cleanup() here; other components may use GPIO as well.
|
|
|
|
# -------------------------------------------------------------------------
|
|
# "Interrupt-like" handlers called from polling thread
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _on_pressed(self) -> None:
|
|
self._press_time = time.monotonic()
|
|
_LOGGER.debug("Button pressed (GPIO %s low)", self._cfg.pin)
|
|
|
|
def _on_released(self) -> None:
|
|
if self._press_time is None:
|
|
return
|
|
|
|
duration = time.monotonic() - self._press_time
|
|
self._press_time = None
|
|
_LOGGER.debug("Button released after %.3f seconds", duration)
|
|
|
|
if duration >= self._cfg.long_press_seconds:
|
|
self._handle_long_press()
|
|
else:
|
|
self._handle_short_press()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# High-level behaviors
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _handle_short_press(self) -> None:
|
|
"""Short press: wake / stop behavior."""
|
|
# Snapshot playing state quickly
|
|
tts_playing = bool(getattr(self.state.tts_player, "is_playing", False))
|
|
music_playing = bool(getattr(self.state.music_player, "is_playing", False))
|
|
|
|
if tts_playing or music_playing:
|
|
_LOGGER.debug(
|
|
"Button short press while audio playing "
|
|
"(tts=%s, music=%s) -> stopping playback",
|
|
tts_playing,
|
|
music_playing,
|
|
)
|
|
|
|
# Stop TTS/timer via the same logic as the Stop wake word.
|
|
if self.state.satellite is not None:
|
|
self.loop.call_soon_threadsafe(self.state.satellite.stop)
|
|
|
|
# Also stop any music playback controlled by the LVA.
|
|
self.loop.call_soon_threadsafe(self.state.music_player.stop)
|
|
else:
|
|
_LOGGER.debug(
|
|
"Button short press with no audio playing -> manual wakeup"
|
|
)
|
|
if self.state.satellite is not None:
|
|
# Use a dedicated manual wakeup to avoid faking a wake-word object.
|
|
self.loop.call_soon_threadsafe(
|
|
self.state.satellite.manual_wakeup, "button"
|
|
)
|
|
|
|
def _handle_long_press(self) -> None:
|
|
"""Long press: toggle mic mute."""
|
|
new_state = not self.state.mic_muted
|
|
_LOGGER.debug(
|
|
"Button long press: toggling mic mute -> %s", new_state
|
|
)
|
|
|
|
# Use the standard set_mic_mute event so MicMuteHandler can:
|
|
# - update ServerState.mic_muted
|
|
# - control the mic_muted_event
|
|
# - publish MQTT mute state
|
|
self.loop.call_soon_threadsafe(
|
|
self.event_bus.publish,
|
|
"set_mic_mute",
|
|
{"state": new_state},
|
|
)
|
|
|