Merge pull request #44 from imonlinux/gpio-button-suport

Summary

Add support for the ReSpeaker 2-Mic HAT momentary button as a first-class control surface for the Linux Voice Assistant. The button now behaves like:

Short press

If TTS or music is playing → stop playback (equivalent to the Stop wake word)

Otherwise → start a new conversation (equivalent to a wake word trigger)

Long press

Toggle microphone mute (wired through the existing set_mic_mute event, so MQTT state and LEDs stay in sync)

The implementation uses a polling-based GPIO loop (RPi.GPIO) instead of kernel edge-detection to avoid “Failed to add edge detection” issues on some HAT/overlay setups. Button behavior is fully configurable via config.json.

README.md

@@ -16,6 +16,7 @@ Want to run the satellite on a Linux desktop using a simple tray client? See [th
 - This fork is from https://github.com/OHF-Voice/linux-voice-assistant Release v1.0.0 which introduces the ability to use both MicroWakeWord and OpenWakeWord detections models.
 - Refactor: Sync with Upstream architectural changes, includes the 7 commits since the release of v1.0.0.
 - **Core Refactor (Nov 2025):** Major architectural cleanup moving audio logic to a dedicated engine.
+- **Added ReSpeaker 2-Mic Button Support:** The GPIO button (17) can be enabled via the config.json file. See [the tutorial](docs/linux-voice-assistant-2mic-install.md).
 - **Changed Alarm to include a Duration Setting:** Added a MQTT control to make the alarm duration adjustable.
 - **Performance Optimization:** Optimized audio threading for lower CPU usage when muted and improved responsiveness (reduced latency) when unmuted.
 - **Non-Blocking Operations:** Wake word model downloads are now handled in background threads, preventing the device (LEDs/MQTT) from freezing during configuration updates.

docs/linux-voice-assistant-2mic-install.md

@@ -498,6 +498,82 @@ systemctl --user status linux-voice-assistant --no-pager -l
 ```
 </details>
 
+<details>
+<summary><strong>Optional (GPIO Button)</strong></summary>
+
+Add support for the ReSpeaker 2-Mic HAT momentary button as a first-class control surface for the Linux Voice Assistant. The button now behaves like:
+
+***Short press***
+
+If TTS or music is playing → stop playback (equivalent to the Stop wake word)
+
+Otherwise → start a new conversation (equivalent to a wake word trigger)
+
+***Long press***
+
+Toggle microphone mute (wired through the existing set_mic_mute event, so MQTT state and LEDs stay in sync)
+
+The implementation uses a polling-based GPIO loop (RPi.GPIO) instead of kernel edge-detection to avoid “Failed to add edge detection” issues on some HAT/overlay setups. Button behavior is fully configurable via config.json.
+
+**Requires a compatible GPIO board** This has been tested on the ReSpeaker 2-Mic Pi Hat:
+
+**Edit the LVA config.json file:**
+
+```bash
+nano ~/linux-voice-assistant/linux_voice_assistant/config.json
+```
+
+**Enable the GPIO button support:**
+
+```bash
+"button": {
+  "enabled": true,
+  "pin": 17,
+  "long_press_seconds": 1.0
+  }
+```
+
+**Example (LVA config.json file with MQTT, Grove Port, and GPIO Button enabled** ***Note: The GPIO button can be changed from the default (17) on the 2-Mic hat.***
+
+```bash
+{
+  "app": {
+    "name": "Linux Voice Assistant"
+  },
+  "mqtt": {
+    "host": "192.168.1.2",
+    "port": 1883,
+    "username": "mqtt_username",
+    "password": "mqtt_password"
+  },
+  "led": {
+    "led_type": "dotstar",
+    "interface": "gpio",
+    "clock_pin": 13,
+    "data_pin": 12,
+    "num_leds": 10
+  },
+  "button": {
+  "enabled": true,
+  "pin": 17,
+  "long_press_seconds": 1.0
+  }
+}
+```
+
+**Enable & start:**
+
+```bash
+systemctl --user restart linux-voice-assistant.service
+```
+
+**Verify:**
+
+```bash
+systemctl --user status linux-voice-assistant --no-pager -l
+```
+</details>
+
 ## 6. Connect to Home Assistant
 
 ### If HA does not discover the new LVA:

docs/lva-desktop.md

@@ -54,6 +54,10 @@ git clone https://github.com/imonlinux/linux-voice-assistant.git
 cd linux-voice-assistant/
 chmod +x script/tray
 script/setup
+
+source .venv/bin/activate
+pip install -e .[tray]
+deactivate
 ```
 
 ### 5. Modify config.json File:

linux_voice_assistant/__main__.py

@@ -15,6 +15,7 @@ from pymicro_wakeword import MicroWakeWord, MicroWakeWordFeatures
 from pyopen_wakeword import OpenWakeWord
 
 from .audio_engine import AudioEngine
+from .button_controller import ButtonController
 from .config import Config, load_config_from_json
 from .event_bus import EventBus, EventHandler, subscribe
 from .led_controller import LedController
@@ -41,6 +42,7 @@ class WakeWordData:
     active: Set[str]
     stop_model: MicroWakeWord
 
+
 @dataclass
 class MediaPlayers:
     """Dataclass to hold media player instances."""
@@ -107,7 +109,7 @@ class MicMuteHandler(EventHandler):
             { "alarm_duration_seconds": <int> }
 
         Semantics:
-            0  -> infinite alarm (only stop via Stop wake word / wake word logic)
+            0  -> infinite alarm (only Stop/wake word stops it)
             >0 -> auto-stop alarm after N seconds (plus Stop wake word support)
         """
         duration = data.get("alarm_duration_seconds")
@@ -424,6 +426,22 @@ def _init_controllers(
         mqtt_controller=mqtt_controller,
     )
 
+    # Hardware button controller (e.g. ReSpeaker 2-Mic HAT)
+    try:
+        if getattr(config, "button", None) is not None and config.button.enabled:
+            button_controller = ButtonController(
+                loop=loop,
+                event_bus=event_bus,
+                state=state,
+                config=config.button,
+            )
+            # Keep a reference on state so it is not garbage-collected.
+            setattr(state, "button_controller", button_controller)
+        else:
+            _LOGGER.debug("ButtonController not enabled in config; skipping")
+    except Exception:
+        _LOGGER.exception("Failed to initialize ButtonController")
+
 async def _run_server(state: ServerState, config: Config):
     """Starts the ESPHome server and ZeroConf discovery."""
     server = await state.loop.create_server(

linux_voice_assistant/button_controller.py

@@ -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},
+        )
+

linux_voice_assistant/config.py

@@ -5,6 +5,9 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Dict, List, Optional
 
+import logging
+_LOGGER = logging.getLogger(__name__)
+
 # -----------------------------------------------------------------------------
 # Configuration Dataclasses
 # -----------------------------------------------------------------------------
@@ -16,6 +19,7 @@ class AudioConfig:
     input_block_size: int = 1024
     output_device: Optional[str] = None
 
+
 @dataclass
 class WakeWordConfig:
     """Settings for wake word detection."""
@@ -25,12 +29,14 @@ class WakeWordConfig:
     refractory_seconds: float = 2.0
     download_dir: str = "local"
 
+
 @dataclass
 class ESPHomeConfig:
     """Settings for the ESPHome API server."""
     host: str = "0.0.0.0"
     port: int = 6053
 
+
 @dataclass
 class LedConfig:
     """Settings for the LED strip."""
@@ -38,7 +44,8 @@ class LedConfig:
     interface: str = "spi"
     clock_pin: int = 13
     data_pin: int = 12
-    num_leds: int = 3  # Note: This will be overridden by 'preferences.json' if it exists
+    num_leds: int = 3  # Note: overridden by 'preferences.json' if it exists
+
 
 @dataclass
 class MqttConfig:
@@ -49,6 +56,17 @@ class MqttConfig:
     username: Optional[str] = None
     password: Optional[str] = None
 
+
+@dataclass
+class ButtonConfig:
+    """Settings for a hardware momentary button (e.g. ReSpeaker 2-Mic HAT)."""
+    enabled: bool = False
+    # BCM GPIO pin number for the button input.
+    pin: int = 17
+    # Press duration (in seconds) to be considered a "long press".
+    long_press_seconds: float = 1.0
+
+
 @dataclass
 class AppConfig:
     """General application settings."""
@@ -58,6 +76,7 @@ class AppConfig:
     preferences_file: str = "preferences.json"
     debug: bool = False
 
+
 @dataclass
 class Config:
     """Main configuration object."""
@@ -67,6 +86,7 @@ class Config:
     esphome: ESPHomeConfig = field(default_factory=ESPHomeConfig)
     led: LedConfig = field(default_factory=LedConfig)
     mqtt: MqttConfig = field(default_factory=MqttConfig)
+    button: ButtonConfig = field(default_factory=ButtonConfig)
 
 # -----------------------------------------------------------------------------
 # Helper Function
@@ -88,7 +108,9 @@ def load_config_from_json(config_path: Path) -> Config:
 
     # --- Step 2: Create config objects from raw data ---
     if "app" not in raw_data:
-        raise ValueError("Configuration file must contain an 'app' section with a 'name'.")
+        raise ValueError(
+            "Configuration file must contain an 'app' section with a 'name'."
+        )
 
     app_config = AppConfig(**raw_data.get("app", {}))
     audio_config = AudioConfig(**raw_data.get("audio", {}))
@@ -96,6 +118,7 @@ def load_config_from_json(config_path: Path) -> Config:
     esphome_config = ESPHomeConfig(**raw_data.get("esphome", {}))
     led_config = LedConfig(**raw_data.get("led", {}))
     mqtt_config = MqttConfig(**raw_data.get("mqtt", {}))
+    button_config = ButtonConfig(**raw_data.get("button", {}))
 
     # --- Step 3: Set MQTT 'enabled' flag ---
     if mqtt_config.host:
@@ -109,7 +132,5 @@ def load_config_from_json(config_path: Path) -> Config:
         esphome=esphome_config,
         led=led_config,
         mqtt=mqtt_config,
+        button=button_config,
     )
-
-import logging
-_LOGGER = logging.getLogger(__name__)

linux_voice_assistant/satellite.py

@@ -478,6 +478,16 @@ class VoiceSatelliteProtocol(APIServer):
         )
         self._stop_timer_alarm("auto_timeout")
 
+    def _start_conversation(self, wake_word_phrase: str) -> None:
+        """Shared helper to start a new conversation run."""
+        _LOGGER.debug("Starting conversation: %s", wake_word_phrase)
+        self.send_messages(
+            [VoiceAssistantRequest(start=True, wake_word_phrase=wake_word_phrase)]
+        )
+        self._set_state(SatelliteState.LISTENING)
+        self._is_streaming_audio = True
+        self.state.tts_player.play(self.state.wakeup_sound)
+
     def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
         if self._state not in (SatelliteState.IDLE, SatelliteState.STARTING):
             # Existing behavior: ignore wakeup in other states.
@@ -491,12 +501,22 @@ class VoiceSatelliteProtocol(APIServer):
 
         wake_word_phrase = getattr(wake_word, "wake_word", "wake word")
         _LOGGER.debug("Detected wake word: %s", wake_word_phrase)
-        self.send_messages(
-            [VoiceAssistantRequest(start=True, wake_word_phrase=wake_word_phrase)]
-        )
-        self._set_state(SatelliteState.LISTENING)
-        self._is_streaming_audio = True
-        self.state.tts_player.play(self.state.wakeup_sound)
+        self._start_conversation(wake_word_phrase)
+
+    def manual_wakeup(self, phrase: str = "button") -> None:
+        """
+        Manual wakeup entrypoint (e.g. hardware button) that behaves like a
+        wake word, but without requiring a wake-word model instance.
+        """
+        if self._state not in (SatelliteState.IDLE, SatelliteState.STARTING):
+            return
+
+        if self._timer_finished:
+            self._stop_timer_alarm("button")
+            return
+
+        _LOGGER.debug("Manual wakeup triggered: %s", phrase)
+        self._start_conversation(phrase)
 
     def stop(self) -> None:
         """

pyproject.toml

@@ -22,6 +22,7 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
 ]
 requires-python = ">=3.9.0"
 dependencies = [
@@ -39,8 +40,6 @@ dependencies = [
     "RPi.GPIO",
     "gpiozero",
     "spidev",
-    "PyQt5",
-    "PyQt6",
 ]
 
 [project.optional-dependencies]
@@ -52,6 +51,11 @@ dev = [
     "pytest",
 ]
 
+# Optional GUI/tray dependencies. Only needed on desktop installs that use lva_tray_client.
+tray = [
+    "PyQt5>=5.15",
+]
+
 [project.urls]
 "Source Code" = "https://github.com/imonlinux/linux-voice-assistant"