sirmrak/linux-voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add button_controller.py

Browse Source 
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.
This commit is contained in:
imonlinux
2025-11-29 23:41:24 -08:00
committed by GitHub
parent edc66183c3
commit dc9049e899
1 changed files with 223 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/button_controller.py
+223
View File
@@ -0,0 +1,223 @@
"""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},
)