Omar Amin
2d460b6728
Use events and commands to control custom onboard leds and buttons. Can be used to communicate with an external Python script via a WebSocket API running in a container alongside LVA.
1201 lines
43 KiB
Python
1201 lines
43 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Satellite 1 HAT Board – Linux Voice Assistant peripheral controller.
|
||
|
||
Mirrors the Home Assistant Voice PE LED ring animations and maps the four
|
||
hardware buttons to LVA peripheral API commands.
|
||
|
||
Hardware layout (Satellite 1 HAT on Raspberry Pi)
|
||
--------------------------------------------------
|
||
LED ring : 12 × SK6812 RGBW NeoPixels → GPIO 12 (PWM0)
|
||
Right btn : Volume Up → GPIO 17
|
||
Left btn : Volume Down → GPIO 27
|
||
Top btn : Mute / Unmute mic → GPIO 22
|
||
Bottom btn : Context action → GPIO 23
|
||
Single press:
|
||
idle → start_listening
|
||
listening → stop_pipeline
|
||
thinking → stop_pipeline
|
||
tts_speaking → stop_pipeline
|
||
timer_ringing → stop_timer_ringing
|
||
media_playing → stop_media_player
|
||
Multi-press (detected via timing):
|
||
double press (< 250ms between releases) → button_double_press
|
||
triple press (< 250ms between releases) → button_triple_press
|
||
long press (held > 1000ms) → button_long_press
|
||
|
||
Install dependencies
|
||
---------------------
|
||
pip install websockets rpi-ws281x gpiozero lgpio
|
||
|
||
Run
|
||
---
|
||
python3 Satellite1_HAT_Board.py
|
||
python3 Satellite1_HAT_Board.py --host 127.0.0.1 --port 6055 --debug
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import math
|
||
import random
|
||
import signal
|
||
import sys
|
||
import threading
|
||
import time
|
||
from enum import Enum
|
||
from typing import List, Optional, Tuple
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hardware dependencies (gracefully stubbed when not on real Pi hardware so
|
||
# the script can be imported / syntax-checked on a dev machine)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
try:
|
||
from gpiozero import Button # type: ignore[import]
|
||
_HAS_GPIO = True
|
||
except ImportError:
|
||
_HAS_GPIO = False
|
||
logging.warning("gpiozero not found – button input disabled")
|
||
|
||
try:
|
||
from rpi_ws281x import PixelStrip, Color as _NeoColor # type: ignore[import]
|
||
_HAS_NEOPIXEL = True
|
||
except ImportError:
|
||
_HAS_NEOPIXEL = False
|
||
logging.warning("rpi_ws281x not found – LED output disabled")
|
||
|
||
try:
|
||
import websockets # type: ignore[import]
|
||
except ImportError:
|
||
sys.exit("websockets not installed. Run: pip install websockets")
|
||
|
||
|
||
# ===========================================================================
|
||
# Configuration – edit these to match your wiring
|
||
# ===========================================================================
|
||
|
||
DEFAULT_LVA_HOST = "localhost"
|
||
DEFAULT_LVA_PORT = 6055
|
||
|
||
# GPIO (BCM numbering)
|
||
LED_GPIO_PIN = 12
|
||
BTN_VOLUME_UP = 17 # Right button
|
||
BTN_VOLUME_DOWN = 27 # Left button
|
||
BTN_MUTE = 22 # Top button
|
||
BTN_ACTION = 23 # Bottom button
|
||
|
||
BTN_DEBOUNCE_MS = 30 # Milliseconds
|
||
|
||
# LED ring
|
||
LED_COUNT = 12
|
||
LED_FREQ_HZ = 800_000
|
||
LED_DMA = 10
|
||
LED_BRIGHTNESS = 168 # 66 % of 255 – matches ESPHome default
|
||
LED_INVERT = False
|
||
LED_CHANNEL = 0
|
||
|
||
# Default ring color (ESPHome default: 9.4 % R, 73.3 % G, 94.9 % B)
|
||
DEFAULT_R, DEFAULT_G, DEFAULT_B = 24, 187, 242
|
||
|
||
# HA Light entity registered with LVA via register_light. The same object_id
|
||
# is used to route incoming light_command events back to this script.
|
||
LIGHT_OBJECT_ID = "led_ring"
|
||
LIGHT_NAME = "LED Ring"
|
||
LIGHT_ICON = "mdi:circle-outline"
|
||
EFFECT_VOICE_ASSISTANT = "Voice Assistant"
|
||
|
||
# Reconnect delay when WebSocket connection to LVA is lost
|
||
RECONNECT_DELAY_S = 3.0
|
||
|
||
# Color wheel parameters (matches Home Assistant Voice PE)
|
||
# HSV hue rotation: ±10° per volume button press
|
||
HUE_INCREMENT = 10 # Degrees
|
||
DEFAULT_HUE = 195 # Cyan (matches default RGB 24, 187, 242)
|
||
DEFAULT_SATURATION = 1.0 # 100%
|
||
DEFAULT_VALUE = 1.0 # 100%
|
||
|
||
|
||
# ===========================================================================
|
||
# Color conversion utilities (HSV ↔ RGB)
|
||
# ===========================================================================
|
||
|
||
def hsv_to_rgb(hue: int, saturation: float, value: float) -> Tuple[int, int, int]:
|
||
"""
|
||
Convert HSV (Hue: 0-360, Saturation: 0-1, Value: 0-1) to RGB (0-255).
|
||
Mirrors the Home Assistant Voice PE color wheel.
|
||
"""
|
||
# Normalize hue to 0-360 range
|
||
hue = hue % 360
|
||
|
||
# Clamp saturation and value to 0-1
|
||
sat = max(0.0, min(1.0, saturation))
|
||
val = max(0.0, min(1.0, value))
|
||
|
||
# Ensure saturation is at least 5% to avoid desaturated colors
|
||
if sat < 0.05:
|
||
sat = 1.0
|
||
|
||
c = val * sat # Chroma
|
||
h_prime = hue / 60.0
|
||
x = c * (1.0 - abs(h_prime % 2.0 - 1.0))
|
||
|
||
if h_prime < 1:
|
||
r_prime, g_prime, b_prime = c, x, 0.0
|
||
elif h_prime < 2:
|
||
r_prime, g_prime, b_prime = x, c, 0.0
|
||
elif h_prime < 3:
|
||
r_prime, g_prime, b_prime = 0.0, c, x
|
||
elif h_prime < 4:
|
||
r_prime, g_prime, b_prime = 0.0, x, c
|
||
elif h_prime < 5:
|
||
r_prime, g_prime, b_prime = x, 0.0, c
|
||
else:
|
||
r_prime, g_prime, b_prime = c, 0.0, x
|
||
|
||
m = val - c
|
||
r = int((r_prime + m) * 255)
|
||
g = int((g_prime + m) * 255)
|
||
b = int((b_prime + m) * 255)
|
||
|
||
return (r, g, b)
|
||
|
||
|
||
def rgb_to_hsv(r: int, g: int, b: int) -> Tuple[int, float, float]:
|
||
"""
|
||
Convert RGB (0-255) to HSV (Hue: 0-360, Saturation: 0-1, Value: 0-1).
|
||
"""
|
||
r_norm = r / 255.0
|
||
g_norm = g / 255.0
|
||
b_norm = b / 255.0
|
||
|
||
max_c = max(r_norm, g_norm, b_norm)
|
||
min_c = min(r_norm, g_norm, b_norm)
|
||
delta = max_c - min_c
|
||
|
||
# Value
|
||
value = max_c
|
||
|
||
# Saturation
|
||
if max_c == 0:
|
||
saturation = 0.0
|
||
else:
|
||
saturation = delta / max_c
|
||
|
||
# Hue
|
||
if delta == 0:
|
||
hue = 0
|
||
elif max_c == r_norm:
|
||
hue = (60 * ((g_norm - b_norm) / delta) + 360) % 360
|
||
elif max_c == g_norm:
|
||
hue = (60 * ((b_norm - r_norm) / delta) + 120) % 360
|
||
else: # max_c == b_norm
|
||
hue = (60 * ((r_norm - g_norm) / delta) + 240) % 360
|
||
|
||
return (int(hue), saturation, value)
|
||
|
||
|
||
# ===========================================================================
|
||
# Logging
|
||
# ===========================================================================
|
||
|
||
|
||
_LOGGER = logging.getLogger("satellite1")
|
||
|
||
|
||
# ===========================================================================
|
||
# Assistant state
|
||
# ===========================================================================
|
||
|
||
class AssistState(str, Enum):
|
||
NOT_READY = "not_ready"
|
||
IDLE = "idle"
|
||
WAKE_WORD = "wake_word_detected"
|
||
LISTENING = "listening"
|
||
THINKING = "thinking"
|
||
SPEAKING = "tts_speaking"
|
||
TTS_FINISHED = "tts_finished"
|
||
ERROR = "error"
|
||
MUTED = "muted"
|
||
TIMER_TICKING = "timer_ticking"
|
||
TIMER_RINGING = "timer_ringing"
|
||
MEDIA_PLAYING = "media_player_playing"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared state (written by WS thread, read by LED thread + button callbacks)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class SharedState:
|
||
def __init__(self) -> None:
|
||
self._lock = threading.Lock()
|
||
self.assist_state: AssistState = AssistState.NOT_READY
|
||
self.ha_connected: bool = False
|
||
self.muted: bool = False
|
||
self.volume: float = 1.0
|
||
self.timer_total_seconds: int = 0
|
||
self.timer_seconds_left: int = 0
|
||
# Light entity state, driven by HA via light_command events.
|
||
# Defaults match the LEDLightEntity in LVA core so the script
|
||
# behaves sensibly before the first light_command arrives: off by
|
||
# default, so idle stays dark until the user turns the light on.
|
||
self.light_is_on: bool = False
|
||
self.light_brightness: float = 0.66
|
||
self.light_red: float = 0.094
|
||
self.light_green: float = 0.733
|
||
self.light_blue: float = 0.949
|
||
# HSV color tracking for hue rotation via volume buttons
|
||
self.hsv_hue: int = DEFAULT_HUE
|
||
self.hsv_saturation: float = DEFAULT_SATURATION
|
||
self.hsv_value: float = DEFAULT_VALUE
|
||
|
||
def update(self, **kwargs) -> None:
|
||
with self._lock:
|
||
for key, val in kwargs.items():
|
||
setattr(self, key, val)
|
||
|
||
@property
|
||
def snapshot(self) -> dict:
|
||
with self._lock:
|
||
return {
|
||
"assist_state": self.assist_state,
|
||
"ha_connected": self.ha_connected,
|
||
"muted": self.muted,
|
||
"volume": self.volume,
|
||
"timer_total_seconds": self.timer_total_seconds,
|
||
"timer_seconds_left": self.timer_seconds_left,
|
||
"light_is_on": self.light_is_on,
|
||
"light_brightness": self.light_brightness,
|
||
"light_red": self.light_red,
|
||
"light_green": self.light_green,
|
||
"light_blue": self.light_blue,
|
||
"hsv_hue": self.hsv_hue,
|
||
"hsv_saturation": self.hsv_saturation,
|
||
"hsv_value": self.hsv_value,
|
||
}
|
||
|
||
|
||
# ===========================================================================
|
||
# LED ring controller
|
||
# Mirrors every animation from the ESPHome home-assistant-voice.yaml exactly.
|
||
# ===========================================================================
|
||
|
||
RGB = Tuple[int, int, int]
|
||
|
||
BLACK: RGB = (0, 0, 0)
|
||
RED: RGB = (255, 0, 0)
|
||
|
||
|
||
def _scale(color: RGB, factor: float) -> RGB:
|
||
"""Scale an RGB colour by 0.0–1.0."""
|
||
f = max(0.0, min(1.0, factor))
|
||
return (int(color[0] * f), int(color[1] * f), int(color[2] * f))
|
||
|
||
|
||
def _neo(color: RGB) -> int:
|
||
"""Convert (r, g, b) to a rpi_ws281x 24-bit integer."""
|
||
if _HAS_NEOPIXEL:
|
||
return _NeoColor(color[0], color[1], color[2])
|
||
return 0
|
||
|
||
|
||
class LEDRing:
|
||
"""
|
||
Manages the 12-LED NeoPixel ring.
|
||
|
||
All animations run in a dedicated daemon thread. Calling
|
||
``set_animation()`` switches cleanly to the new pattern.
|
||
"""
|
||
|
||
# Animation names mirror ESPHome effect names for easy cross-reference
|
||
ANIM_IDLE = "idle" # Idle with light color when on
|
||
ANIM_OFF = "off"
|
||
ANIM_TWINKLE = "twinkle" # Not-ready / no HA
|
||
ANIM_TWINKLE_BLUE = "twinkle_blue" # Init / connecting
|
||
ANIM_WAIT_CMD = "waiting" # Wake word detected (slow spin)
|
||
ANIM_LISTENING = "listening" # STT active (fast spin)
|
||
ANIM_THINKING = "thinking" # Intent processing (pulse pair)
|
||
ANIM_REPLYING = "replying" # TTS speaking (reverse spin)
|
||
ANIM_MUTED = "muted" # Muted indicators
|
||
ANIM_ERROR = "error" # Red pulse
|
||
ANIM_TIMER_TICK = "timer_tick" # Countdown arc
|
||
ANIM_TIMER_RING = "timer_ring" # Ringing pulse
|
||
|
||
def __init__(self, state: SharedState) -> None:
|
||
self._state = state
|
||
self._pixels: List[RGB] = [BLACK] * LED_COUNT
|
||
self._strip = None
|
||
self._animation = self.ANIM_OFF
|
||
self._anim_lock = threading.Lock()
|
||
self._stop_event = threading.Event()
|
||
self._thread = threading.Thread(
|
||
target=self._loop, daemon=True, name="led-ring"
|
||
)
|
||
|
||
# Per-animation internal counters (reset on animation change)
|
||
self._index: int = 0
|
||
self._brightness_step: int = 0
|
||
self._brightness_dec: bool = True
|
||
self._twinkle_state: List[float] = [0.0] * LED_COUNT
|
||
|
||
if _HAS_NEOPIXEL:
|
||
self._strip = PixelStrip(
|
||
LED_COUNT, LED_GPIO_PIN, LED_FREQ_HZ,
|
||
LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL,
|
||
)
|
||
self._strip.begin()
|
||
_LOGGER.info("LED ring initialised (%d LEDs on GPIO %d)", LED_COUNT, LED_GPIO_PIN)
|
||
|
||
def start(self) -> None:
|
||
self._thread.start()
|
||
|
||
def stop(self) -> None:
|
||
self._stop_event.set()
|
||
self._thread.join(timeout=2)
|
||
self._all_off()
|
||
|
||
def set_animation(self, name: str) -> None:
|
||
with self._anim_lock:
|
||
if self._animation != name:
|
||
_LOGGER.debug("LED animation → %s", name)
|
||
self._animation = name
|
||
# Reset counters on animation change (mirrors ESPHome initial_run)
|
||
self._index = 0
|
||
self._brightness_step = 0
|
||
self._brightness_dec = True
|
||
self._twinkle_state = [random.random() for _ in range(LED_COUNT)]
|
||
|
||
# ------------------------------------------------------------------
|
||
# Internal helpers
|
||
# ------------------------------------------------------------------
|
||
|
||
def _write(self) -> None:
|
||
if not _HAS_NEOPIXEL or self._strip is None:
|
||
return
|
||
for i, color in enumerate(self._pixels):
|
||
self._strip.setPixelColor(i, _neo(color))
|
||
self._strip.show()
|
||
|
||
def _all_off(self) -> None:
|
||
self._pixels = [BLACK] * LED_COUNT
|
||
self._write()
|
||
|
||
def _set(self, i: int, color: RGB) -> None:
|
||
self._pixels[i % LED_COUNT] = color
|
||
|
||
def _color(self) -> RGB:
|
||
"""Return the current user-configured ring colour from light entity state."""
|
||
snap = self._state.snapshot
|
||
return (
|
||
int(max(0.0, min(1.0, snap["light_red"])) * 255),
|
||
int(max(0.0, min(1.0, snap["light_green"])) * 255),
|
||
int(max(0.0, min(1.0, snap["light_blue"])) * 255),
|
||
)
|
||
|
||
def _pulse_step(self, steps: int = 10) -> float:
|
||
"""Advance pulse counter and return brightness 0.0–1.0."""
|
||
factor = (steps - self._brightness_step) / steps
|
||
if self._brightness_dec:
|
||
self._brightness_step += 1
|
||
else:
|
||
self._brightness_step -= 1
|
||
if self._brightness_step <= 0 or self._brightness_step >= steps:
|
||
self._brightness_dec = not self._brightness_dec
|
||
return factor
|
||
|
||
# ------------------------------------------------------------------
|
||
# Animation implementations (each returns sleep time in seconds)
|
||
# All logic mirrors the ESPHome addressable_lambda effects verbatim.
|
||
# ------------------------------------------------------------------
|
||
|
||
def _anim_idle(self) -> float:
|
||
"""Resting state with user color when light is on, else dark."""
|
||
snap = self._state.snapshot
|
||
if snap["light_is_on"]:
|
||
# Show user color at configured brightness
|
||
color = self._color()
|
||
brightness = snap["light_brightness"]
|
||
for i in range(LED_COUNT):
|
||
self._set(i, _scale(color, brightness))
|
||
self._write()
|
||
else:
|
||
self._all_off()
|
||
return 0.1
|
||
|
||
def _anim_off(self) -> float:
|
||
self._all_off()
|
||
return 0.1
|
||
|
||
def _anim_twinkle(self, color: RGB) -> float:
|
||
"""Random sparkle – used for Not Ready and No HA connection states."""
|
||
FADE = 0.85
|
||
SPARK_PROB = 0.15
|
||
for i in range(LED_COUNT):
|
||
if random.random() < SPARK_PROB:
|
||
self._twinkle_state[i] = 1.0
|
||
else:
|
||
self._twinkle_state[i] *= FADE
|
||
self._set(i, _scale(color, self._twinkle_state[i]))
|
||
self._write()
|
||
return 0.05
|
||
|
||
def _anim_spin(self, color: RGB, interval: float, reverse: bool = False) -> float:
|
||
"""
|
||
Shared spin pattern for Waiting / Listening / Replying.
|
||
|
||
Mirrors the ESPHome "Waiting for Command", "Listening For Command"
|
||
and "Replying" addressable_lambda effects.
|
||
"""
|
||
if reverse:
|
||
# Replying goes anticlockwise: index decrements
|
||
self._index = (LED_COUNT + self._index - 1) % LED_COUNT
|
||
offsets = [(0, 1.0), (1, 192/255), (2, 128/255),
|
||
(6, 1.0), (7, 192/255), (8, 128/255)]
|
||
else:
|
||
offsets = [(0, 1.0), (11, 192/255), (10, 128/255),
|
||
(6, 1.0), (5, 192/255), (4, 128/255)]
|
||
|
||
for i in range(LED_COUNT):
|
||
self._set(i, BLACK)
|
||
|
||
for offset, brightness in offsets:
|
||
self._set((self._index + offset) % LED_COUNT, _scale(color, brightness))
|
||
|
||
if not reverse:
|
||
self._index = (self._index + 1) % LED_COUNT
|
||
|
||
self._write()
|
||
return interval
|
||
|
||
def _anim_thinking(self, color: RGB) -> float:
|
||
"""
|
||
Two opposing LEDs pulsing in brightness.
|
||
Mirrors ESPHome "Thinking" effect. Index does NOT advance.
|
||
"""
|
||
factor = self._pulse_step(10)
|
||
for i in range(LED_COUNT):
|
||
if i == self._index % LED_COUNT or i == (self._index + 6) % LED_COUNT:
|
||
self._set(i, _scale(color, factor))
|
||
else:
|
||
self._set(i, BLACK)
|
||
self._write()
|
||
return 0.01
|
||
|
||
def _anim_muted(self, color: RGB, muted: bool) -> float:
|
||
"""
|
||
Solid ring with red indicators at all 4 mic positions when muted.
|
||
Satellite 1 HAT has mics at 12, 3, 6 and 9 o'clock → LEDs 0, 3, 6, 9.
|
||
"""
|
||
for i in range(LED_COUNT):
|
||
self._set(i, color)
|
||
|
||
if muted:
|
||
# 4 mic positions: top (0), right (3), bottom (6), left (9)
|
||
# Blank the immediate neighbours so the red indicators stand out
|
||
self._set(11, BLACK); self._set(0, RED); self._set(1, BLACK)
|
||
self._set(2, BLACK); self._set(3, RED); self._set(4, BLACK)
|
||
self._set(5, BLACK); self._set(6, RED); self._set(7, BLACK)
|
||
self._set(8, BLACK); self._set(9, RED); self._set(10, BLACK)
|
||
|
||
self._write()
|
||
return 0.016
|
||
|
||
def _anim_error(self) -> float:
|
||
"""
|
||
All LEDs red, pulsing. Mirrors ESPHome "Error" effect.
|
||
"""
|
||
factor = self._pulse_step(10)
|
||
for i in range(LED_COUNT):
|
||
self._set(i, _scale(RED, factor))
|
||
self._write()
|
||
return 0.01
|
||
|
||
def _anim_timer_ring(self, color: RGB, muted: bool) -> float:
|
||
"""
|
||
All LEDs pulse with ring colour; red at all 4 mic positions if muted.
|
||
Mirrors ESPHome "Timer Ring" effect.
|
||
"""
|
||
factor = self._pulse_step(10)
|
||
for i in range(LED_COUNT):
|
||
self._set(i, _scale(color, factor))
|
||
if muted:
|
||
# 4 mic positions: LEDs 0, 3, 6, 9
|
||
self._set(0, _scale(RED, factor))
|
||
self._set(3, _scale(RED, factor))
|
||
self._set(6, _scale(RED, factor))
|
||
self._set(9, _scale(RED, factor))
|
||
self._write()
|
||
return 0.01
|
||
|
||
def _anim_timer_tick(
|
||
self, color: RGB, muted: bool,
|
||
seconds_left: int, total_seconds: int,
|
||
) -> float:
|
||
"""
|
||
Arc of LEDs proportional to remaining time.
|
||
Mirrors ESPHome "Timer Tick" effect exactly, including the
|
||
brightness-dip on the sweep marker LED.
|
||
"""
|
||
total = max(total_seconds, 1)
|
||
timer_ratio = LED_COUNT * seconds_left / total
|
||
last_led_on = max(0, math.ceil(timer_ratio) - 1)
|
||
|
||
# Index sweeps anticlockwise (matches ESPHome anticlockwise decrement)
|
||
for i in range(LED_COUNT):
|
||
brightness_dip = (
|
||
0.9 if (i == self._index % LED_COUNT and i != last_led_on) else 1.0
|
||
)
|
||
if i <= timer_ratio:
|
||
brightness = min(brightness_dip * (timer_ratio - i), brightness_dip)
|
||
self._set(i, _scale(color, brightness))
|
||
else:
|
||
self._set(i, BLACK)
|
||
|
||
if muted:
|
||
# 4 mic positions: LEDs 0, 3, 6, 9
|
||
self._set(11, BLACK); self._set(0, RED); self._set(1, BLACK)
|
||
self._set(2, BLACK); self._set(3, RED); self._set(4, BLACK)
|
||
self._set(5, BLACK); self._set(6, RED); self._set(7, BLACK)
|
||
self._set(8, BLACK); self._set(9, RED); self._set(10, BLACK)
|
||
|
||
self._index = (LED_COUNT + self._index - 1) % LED_COUNT
|
||
self._write()
|
||
return 0.1
|
||
|
||
# ------------------------------------------------------------------
|
||
# Animation loop
|
||
# ------------------------------------------------------------------
|
||
|
||
def _loop(self) -> None:
|
||
while not self._stop_event.is_set():
|
||
with self._anim_lock:
|
||
anim = self._animation
|
||
|
||
snap = self._state.snapshot
|
||
color = self._color()
|
||
muted = snap["muted"]
|
||
t_total = snap["timer_total_seconds"]
|
||
t_left = snap["timer_seconds_left"]
|
||
ha_connected = snap["ha_connected"]
|
||
|
||
if anim == self.ANIM_IDLE:
|
||
sleep = self._anim_idle()
|
||
|
||
elif anim == self.ANIM_OFF:
|
||
sleep = self._anim_off()
|
||
|
||
elif anim == self.ANIM_TWINKLE:
|
||
sleep = self._anim_twinkle(RED)
|
||
|
||
elif anim == self.ANIM_TWINKLE_BLUE:
|
||
sleep = self._anim_twinkle(color)
|
||
|
||
elif anim == self.ANIM_WAIT_CMD:
|
||
sleep = self._anim_spin(color, interval=0.1, reverse=False)
|
||
|
||
elif anim == self.ANIM_LISTENING:
|
||
sleep = self._anim_spin(color, interval=0.05, reverse=False)
|
||
|
||
elif anim == self.ANIM_THINKING:
|
||
sleep = self._anim_thinking(color)
|
||
|
||
elif anim == self.ANIM_REPLYING:
|
||
sleep = self._anim_spin(color, interval=0.05, reverse=True)
|
||
|
||
elif anim == self.ANIM_MUTED:
|
||
sleep = self._anim_muted(color, muted=True)
|
||
|
||
elif anim == self.ANIM_ERROR:
|
||
sleep = self._anim_error()
|
||
|
||
elif anim == self.ANIM_TIMER_RING:
|
||
sleep = self._anim_timer_ring(color, muted)
|
||
|
||
elif anim == self.ANIM_TIMER_TICK:
|
||
sleep = self._anim_timer_tick(color, muted, t_left, t_total)
|
||
|
||
else:
|
||
sleep = self._anim_off()
|
||
|
||
time.sleep(sleep)
|
||
|
||
|
||
# ===========================================================================
|
||
# State → animation mapping
|
||
# Mirrors the control_leds script priority logic from ESPHome
|
||
# ===========================================================================
|
||
|
||
def state_to_animation(state: AssistState, ha_connected: bool, muted: bool) -> str:
|
||
if not ha_connected:
|
||
return LEDRing.ANIM_TWINKLE # Red twinkle = no HA connection
|
||
|
||
if state == AssistState.NOT_READY:
|
||
return LEDRing.ANIM_TWINKLE # Red twinkle
|
||
|
||
if state == AssistState.TIMER_RINGING:
|
||
return LEDRing.ANIM_TIMER_RING
|
||
|
||
if state == AssistState.WAKE_WORD:
|
||
return LEDRing.ANIM_WAIT_CMD # Slow clockwise spin
|
||
|
||
if state == AssistState.LISTENING:
|
||
return LEDRing.ANIM_LISTENING # Fast clockwise spin
|
||
|
||
if state == AssistState.THINKING:
|
||
return LEDRing.ANIM_THINKING # Pulsing pair
|
||
|
||
if state == AssistState.SPEAKING:
|
||
return LEDRing.ANIM_REPLYING # Anticlockwise spin
|
||
|
||
if state == AssistState.ERROR:
|
||
return LEDRing.ANIM_ERROR # Red pulse
|
||
|
||
if state == AssistState.MUTED or muted:
|
||
return LEDRing.ANIM_MUTED
|
||
|
||
if state == AssistState.TIMER_TICKING:
|
||
return LEDRing.ANIM_TIMER_TICK # Countdown arc
|
||
|
||
if state == AssistState.IDLE:
|
||
return LEDRing.ANIM_IDLE # User color when light on, else off
|
||
|
||
# MEDIA_PLAYING, TTS_FINISHED → off
|
||
return LEDRing.ANIM_OFF
|
||
|
||
|
||
# ===========================================================================
|
||
# Button handler — uses gpiozero (works on Pi 3/4/5 without RPi.GPIO)
|
||
# ===========================================================================
|
||
|
||
class ButtonHandler:
|
||
"""
|
||
Manages the four hardware buttons using gpiozero.
|
||
|
||
All callbacks run in the GPIO event-detection thread and schedule
|
||
coroutines onto the asyncio event loop thread-safely.
|
||
|
||
Supports:
|
||
- Volume +/- for volume control
|
||
- Mute button for microphone toggle
|
||
- Action button for context-aware commands or color changing:
|
||
* Hold + Volume +: Increase hue by 10°
|
||
* Hold + Volume -: Decrease hue by 10°
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
state: SharedState,
|
||
loop: asyncio.AbstractEventLoop,
|
||
command_queue: asyncio.Queue,
|
||
) -> None:
|
||
self._state = state
|
||
self._loop = loop
|
||
self._queue = command_queue
|
||
self._buttons: List = []
|
||
# Track action button hold state for color changing
|
||
self._action_button_held = False
|
||
self._action_button_obj = None
|
||
|
||
def setup(self) -> None:
|
||
if not _HAS_GPIO:
|
||
_LOGGER.warning("GPIO unavailable – buttons not configured")
|
||
return
|
||
|
||
debounce = BTN_DEBOUNCE_MS / 1000.0
|
||
|
||
btn_up = Button(BTN_VOLUME_UP, pull_up=True, bounce_time=debounce)
|
||
btn_down = Button(BTN_VOLUME_DOWN, pull_up=True, bounce_time=debounce)
|
||
btn_mute = Button(BTN_MUTE, pull_up=True, bounce_time=debounce)
|
||
btn_act = Button(BTN_ACTION, pull_up=True, bounce_time=debounce)
|
||
|
||
btn_up.when_pressed = self._on_volume_up
|
||
btn_down.when_pressed = self._on_volume_down
|
||
btn_mute.when_pressed = self._on_mute
|
||
btn_act.when_pressed = self._on_action
|
||
btn_act.when_held = self._on_action_held
|
||
btn_act.when_released = self._on_action_released
|
||
btn_act.hold_time = 0.1 # 100ms to detect hold
|
||
|
||
# Keep references — gpiozero Buttons are released when garbage-collected
|
||
self._buttons = [btn_up, btn_down, btn_mute, btn_act]
|
||
self._action_button_obj = btn_act
|
||
_LOGGER.info(
|
||
"Buttons configured (gpiozero BCM %d/%d/%d/%d)",
|
||
BTN_VOLUME_UP, BTN_VOLUME_DOWN, BTN_MUTE, BTN_ACTION,
|
||
)
|
||
|
||
def cleanup(self) -> None:
|
||
for btn in self._buttons:
|
||
btn.close()
|
||
self._buttons.clear()
|
||
|
||
def _send(self, command: str) -> None:
|
||
_LOGGER.info("Button → %s", command)
|
||
asyncio.run_coroutine_threadsafe(
|
||
self._queue.put(command), self._loop
|
||
)
|
||
|
||
def _on_volume_up(self) -> None:
|
||
if self._action_button_held:
|
||
self._on_color_hue_increase()
|
||
else:
|
||
self._send("volume_up")
|
||
|
||
def _on_volume_down(self) -> None:
|
||
if self._action_button_held:
|
||
self._on_color_hue_decrease()
|
||
else:
|
||
self._send("volume_down")
|
||
|
||
def _on_action_held(self) -> None:
|
||
"""Action button held – enter color change mode."""
|
||
self._action_button_held = True
|
||
_LOGGER.info("Color change mode active (hold action + use volume buttons)")
|
||
|
||
def _on_action_released(self) -> None:
|
||
"""Action button released – exit color change mode."""
|
||
self._action_button_held = False
|
||
|
||
def _on_color_hue_increase(self) -> None:
|
||
"""Increase hue by 10° (volume up while action held)."""
|
||
snap = self._state.snapshot
|
||
new_hue = (snap["hsv_hue"] + HUE_INCREMENT) % 360
|
||
self._send_color_command(new_hue, snap["hsv_saturation"], snap["hsv_value"])
|
||
|
||
def _on_color_hue_decrease(self) -> None:
|
||
"""Decrease hue by 10° (volume down while action held)."""
|
||
snap = self._state.snapshot
|
||
new_hue = (snap["hsv_hue"] - HUE_INCREMENT) % 360
|
||
self._send_color_command(new_hue, snap["hsv_saturation"], snap["hsv_value"])
|
||
|
||
def _send_color_command(self, hue: int, saturation: float, value: float) -> None:
|
||
"""Convert HSV to RGB and send light_command to LVA."""
|
||
# Update local HSV state
|
||
self._state.update(hsv_hue=hue, hsv_saturation=saturation, hsv_value=value)
|
||
|
||
# Convert HSV to RGB for light command
|
||
r, g, b = hsv_to_rgb(hue, saturation, value)
|
||
|
||
# Normalize RGB to 0-1 range
|
||
light_cmd = {
|
||
"command": "light_command",
|
||
"data": {
|
||
"object_id": LIGHT_OBJECT_ID,
|
||
"state": True,
|
||
"brightness": self._state.snapshot["light_brightness"],
|
||
"red": r / 255.0,
|
||
"green": g / 255.0,
|
||
"blue": b / 255.0,
|
||
},
|
||
}
|
||
|
||
_LOGGER.info("Color change: Hue=%d° → RGB(%d, %d, %d)", hue, r, g, b)
|
||
|
||
# Send command to LVA via command queue
|
||
asyncio.run_coroutine_threadsafe(
|
||
self._queue.put(json.dumps(light_cmd)), self._loop
|
||
)
|
||
|
||
def _on_mute(self) -> None:
|
||
"""Toggle mute: sends mute_mic or unmute_mic based on current state."""
|
||
if self._state.muted:
|
||
self._send("unmute_mic")
|
||
else:
|
||
self._send("mute_mic")
|
||
|
||
def _on_action(self) -> None:
|
||
"""
|
||
Context-sensitive bottom button — mirrors HA Voice PE priority:
|
||
1. Timer ringing → stop_timer_ringing
|
||
2. Pipeline active → stop_pipeline
|
||
3. Media playing → stop_media_player
|
||
4. Idle / anything else → start_listening
|
||
"""
|
||
assist = self._state.assist_state
|
||
|
||
if assist == AssistState.TIMER_RINGING:
|
||
self._send("stop_timer_ringing")
|
||
elif assist in (AssistState.WAKE_WORD, AssistState.LISTENING, AssistState.THINKING, AssistState.SPEAKING):
|
||
# stop_pipeline aborts the voice pipeline at any of these phases
|
||
self._send("stop_pipeline")
|
||
elif assist == AssistState.MEDIA_PLAYING:
|
||
self._send("stop_media_player")
|
||
else:
|
||
self._send("start_listening")
|
||
|
||
class ButtonMultipressHandler:
|
||
"""Handles multiple presses of the action button (double, triple, and long press) to trigger different commands. """
|
||
def __init__(self):
|
||
self.button_pin = BTN_ACTION
|
||
self.last_press_time = 0
|
||
self.press_count = 0
|
||
|
||
def _send(self, command: str) -> None:
|
||
_LOGGER.info("Button → %s", command)
|
||
asyncio.run_coroutine_threadsafe(
|
||
self._queue.put(command), self._loop
|
||
)
|
||
|
||
def button_pressed(self):
|
||
current_time = time.time()
|
||
time_since_last_press = current_time - self.last_press_time
|
||
|
||
if time_since_last_press <= 0.25:
|
||
self.press_count += 1
|
||
else:
|
||
self.press_count = 1
|
||
|
||
self.last_press_time = current_time
|
||
|
||
if self.press_count == 2:
|
||
self.handle_double_press()
|
||
elif self.press_count == 3:
|
||
self.handle_triple_press()
|
||
elif time_since_last_press > 1: # Long press considered after 1 second
|
||
self.handle_long_press()
|
||
|
||
def handle_double_press(self):
|
||
# Call LVA peripheral API for double press
|
||
self._send('button_double_press')
|
||
|
||
def handle_triple_press(self):
|
||
# Call LVA peripheral API for triple press
|
||
self._send('button_triple_press')
|
||
|
||
def handle_long_press(self):
|
||
# Call LVA peripheral API for long press
|
||
self._send('button_long_press')
|
||
|
||
# ===========================================================================
|
||
# WebSocket client
|
||
# ===========================================================================
|
||
|
||
class LVAClient:
|
||
"""
|
||
Maintains a persistent WebSocket connection to the LVA peripheral API.
|
||
|
||
- Receives events and updates SharedState + LEDRing.
|
||
- Drains command_queue and sends commands.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
host: str,
|
||
port: int,
|
||
state: SharedState,
|
||
leds: LEDRing,
|
||
command_queue: asyncio.Queue,
|
||
) -> None:
|
||
self._uri = f"ws://{host}:{port}"
|
||
self._state = state
|
||
self._leds = leds
|
||
self._queue = command_queue
|
||
|
||
async def run_forever(self) -> None:
|
||
while True:
|
||
try:
|
||
await self._connect()
|
||
except Exception as exc: # pylint: disable=broad-except
|
||
_LOGGER.warning(
|
||
"LVA connection lost (%s) – retrying in %.0fs",
|
||
exc, RECONNECT_DELAY_S,
|
||
)
|
||
self._state.update(
|
||
ha_connected=False,
|
||
assist_state=AssistState.NOT_READY,
|
||
)
|
||
self._leds.set_animation(LEDRing.ANIM_TWINKLE)
|
||
await asyncio.sleep(RECONNECT_DELAY_S)
|
||
|
||
async def _connect(self) -> None:
|
||
_LOGGER.info("Connecting to LVA at %s …", self._uri)
|
||
async with websockets.connect(self._uri) as ws:
|
||
_LOGGER.info("Connected to LVA peripheral API")
|
||
# Register the LED Light entity with LVA so HA can control it.
|
||
# LVA treats repeat registrations for the same object_id as a
|
||
# no-op, so it's safe to send this on every connect.
|
||
await ws.send(json.dumps({
|
||
"command": "register_light",
|
||
"data": {
|
||
"name": LIGHT_NAME,
|
||
"object_id": LIGHT_OBJECT_ID,
|
||
"icon": LIGHT_ICON,
|
||
"effects": [EFFECT_VOICE_ASSISTANT],
|
||
"supports_rgb": True,
|
||
"supports_brightness": True,
|
||
},
|
||
}))
|
||
# Declare that this peripheral has physical buttons so LVA
|
||
# creates a Button Press event entity in Home Assistant.
|
||
# Idempotent: safe to send on every reconnect.
|
||
await ws.send(json.dumps({"command": "register_button"}))
|
||
recv_task = asyncio.create_task(self._recv_loop(ws))
|
||
send_task = asyncio.create_task(self._send_loop(ws))
|
||
done, pending = await asyncio.wait(
|
||
[recv_task, send_task],
|
||
return_when=asyncio.FIRST_EXCEPTION,
|
||
)
|
||
for task in pending:
|
||
task.cancel()
|
||
# Re-raise the first exception so run_forever() can handle it
|
||
for task in done:
|
||
if not task.cancelled() and task.exception():
|
||
raise task.exception()
|
||
|
||
async def _recv_loop(self, ws) -> None:
|
||
async for raw in ws:
|
||
try:
|
||
msg = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
await self._handle_event(msg)
|
||
|
||
async def _send_loop(self, ws) -> None:
|
||
while True:
|
||
item = await self._queue.get()
|
||
try:
|
||
# Item can be either a string (command) or a pre-formatted JSON dict (light_command)
|
||
if isinstance(item, str):
|
||
# Try to parse as JSON (for light_command) or wrap as command
|
||
try:
|
||
msg = json.loads(item)
|
||
except json.JSONDecodeError:
|
||
# Plain string command
|
||
msg = {"command": item}
|
||
else:
|
||
msg = item
|
||
|
||
await ws.send(json.dumps(msg))
|
||
except Exception as exc: # pylint: disable=broad-except
|
||
_LOGGER.warning("Failed to send item %s: %s", item, exc)
|
||
# Put it back so it's not silently dropped
|
||
self._queue.put_nowait(item)
|
||
raise
|
||
|
||
async def _handle_event(self, msg: dict) -> None:
|
||
event = msg.get("event", "")
|
||
data = msg.get("data") or {}
|
||
|
||
_LOGGER.debug("Event: %s data=%s", event, data)
|
||
|
||
# --- Snapshot (sent on connect) ------------------------------------
|
||
if event == "snapshot":
|
||
self._state.update(
|
||
muted=data.get("muted", False),
|
||
volume=data.get("volume", 1.0),
|
||
ha_connected=data.get("ha_connected", False),
|
||
)
|
||
snap = self._state.snapshot
|
||
anim = state_to_animation(
|
||
snap["assist_state"], snap["ha_connected"], snap["muted"]
|
||
)
|
||
self._leds.set_animation(anim)
|
||
return
|
||
|
||
# --- Voice pipeline events -----------------------------------------
|
||
elif event == "wake_word_detected":
|
||
self._state.update(assist_state=AssistState.WAKE_WORD)
|
||
|
||
elif event == "listening":
|
||
self._state.update(assist_state=AssistState.LISTENING)
|
||
|
||
elif event == "thinking":
|
||
self._state.update(assist_state=AssistState.THINKING)
|
||
|
||
elif event == "tts_speaking":
|
||
self._state.update(assist_state=AssistState.SPEAKING)
|
||
|
||
elif event in ("tts_finished", "idle"):
|
||
self._state.update(assist_state=AssistState.IDLE)
|
||
|
||
elif event == "muted":
|
||
# Carries the mic mute state in both directions. Default True so a
|
||
# bare "muted" event (no data) still reads as muted.
|
||
muted = data.get("muted", True)
|
||
self._state.update(muted=muted)
|
||
if muted:
|
||
self._state.update(assist_state=AssistState.MUTED)
|
||
elif self._state.assist_state == AssistState.MUTED:
|
||
self._state.update(assist_state=AssistState.IDLE)
|
||
|
||
elif event == "pipeline_error":
|
||
_LOGGER.warning("LVA pipeline error: %s", data.get("reason", ""))
|
||
self._state.update(assist_state=AssistState.ERROR)
|
||
|
||
elif event == "disconnected":
|
||
_LOGGER.warning("Home Assistant disconnected")
|
||
self._state.update(
|
||
ha_connected=False,
|
||
assist_state=AssistState.NOT_READY,
|
||
)
|
||
|
||
elif event == "error":
|
||
reason = data.get("reason", "")
|
||
_LOGGER.warning("LVA error: %s", reason)
|
||
if reason == "ha_disconnected":
|
||
self._state.update(
|
||
ha_connected=False,
|
||
assist_state=AssistState.NOT_READY,
|
||
)
|
||
else:
|
||
self._state.update(assist_state=AssistState.ERROR)
|
||
|
||
# --- Timer events --------------------------------------------------
|
||
elif event == "timer_ticking":
|
||
self._state.update(
|
||
assist_state=AssistState.TIMER_TICKING,
|
||
timer_total_seconds=data.get("total_seconds", 0),
|
||
timer_seconds_left=data.get("seconds_left", 0),
|
||
)
|
||
|
||
elif event == "timer_updated":
|
||
self._state.update(
|
||
timer_total_seconds=data.get("total_seconds", 0),
|
||
timer_seconds_left=data.get("seconds_left", 0),
|
||
)
|
||
|
||
elif event == "timer_ringing":
|
||
self._state.update(
|
||
assist_state=AssistState.TIMER_RINGING,
|
||
timer_total_seconds=data.get("total_seconds", 0),
|
||
timer_seconds_left=data.get("seconds_left", 0),
|
||
)
|
||
|
||
# --- Media / volume events -----------------------------------------
|
||
elif event == "media_player_playing":
|
||
self._state.update(assist_state=AssistState.MEDIA_PLAYING)
|
||
|
||
elif event == "volume_changed":
|
||
self._state.update(volume=data.get("volume", 1.0))
|
||
|
||
elif event == "volume_muted":
|
||
self._state.update(muted=data.get("muted", False))
|
||
|
||
# --- Zeroconf / connection events ----------------------------------
|
||
elif event == "zeroconf":
|
||
status = data.get("status", "")
|
||
if status == "connected":
|
||
self._state.update(ha_connected=True)
|
||
_LOGGER.info("Home Assistant connected")
|
||
elif status == "getting_started":
|
||
_LOGGER.info("LVA starting up, waiting for HA …")
|
||
|
||
# --- Light command events ------------------------------------------
|
||
elif event == "light_command":
|
||
# LVA broadcasts to every connected peripheral; only act on
|
||
# commands targeting our registered Light. The Light exposes a
|
||
# single "Voice Assistant" effect, so there is no effect to
|
||
# switch on: we apply on/off, brightness, and color and let the
|
||
# pipeline animations run.
|
||
if data.get("object_id") != LIGHT_OBJECT_ID:
|
||
return
|
||
|
||
r = float(data.get("red", 0.094))
|
||
g = float(data.get("green", 0.733))
|
||
b = float(data.get("blue", 0.949))
|
||
|
||
# Convert RGB back to HSV and track hue for color change mode
|
||
hue, sat, val = rgb_to_hsv(
|
||
int(r * 255), int(g * 255), int(b * 255)
|
||
)
|
||
|
||
self._state.update(
|
||
light_is_on=bool(data.get("state", True)),
|
||
light_brightness=float(data.get("brightness", 0.66)),
|
||
light_red=r,
|
||
light_green=g,
|
||
light_blue=b,
|
||
hsv_hue=hue,
|
||
hsv_saturation=sat,
|
||
hsv_value=val,
|
||
)
|
||
# Force re-render animation with new light state
|
||
snap = self._state.snapshot
|
||
anim = state_to_animation(
|
||
snap["assist_state"], snap["ha_connected"], snap["muted"]
|
||
)
|
||
self._leds.set_animation(anim)
|
||
return
|
||
|
||
# --- Recompute LED animation after every event ---------------------
|
||
snap = self._state.snapshot
|
||
anim = state_to_animation(
|
||
snap["assist_state"], snap["ha_connected"], snap["muted"]
|
||
)
|
||
self._leds.set_animation(anim)
|
||
|
||
|
||
# ===========================================================================
|
||
# Main
|
||
# ===========================================================================
|
||
|
||
async def async_main(host: str, port: int) -> None:
|
||
state = SharedState()
|
||
command_queue: asyncio.Queue = asyncio.Queue()
|
||
loop = asyncio.get_running_loop()
|
||
|
||
leds = LEDRing(state)
|
||
buttons = ButtonHandler(state, loop, command_queue)
|
||
client = LVAClient(host, port, state, leds, command_queue)
|
||
|
||
# Start hardware
|
||
leds.start()
|
||
buttons.setup()
|
||
|
||
# Start with "not ready" animation
|
||
leds.set_animation(LEDRing.ANIM_TWINKLE)
|
||
|
||
# Graceful shutdown on SIGINT / SIGTERM
|
||
def _shutdown(signum, frame) -> None:
|
||
_LOGGER.info("Shutdown requested")
|
||
leds.stop()
|
||
buttons.cleanup()
|
||
sys.exit(0)
|
||
|
||
signal.signal(signal.SIGINT, _shutdown)
|
||
signal.signal(signal.SIGTERM, _shutdown)
|
||
|
||
_LOGGER.info(
|
||
"Satellite 1 HAT controller started – connecting to ws://%s:%d",
|
||
host, port,
|
||
)
|
||
|
||
try:
|
||
await client.run_forever()
|
||
finally:
|
||
leds.stop()
|
||
buttons.cleanup()
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="Satellite 1 HAT Board controller for Linux Voice Assistant"
|
||
)
|
||
parser.add_argument(
|
||
"--host", default=DEFAULT_LVA_HOST,
|
||
help=f"LVA container hostname or IP (default: {DEFAULT_LVA_HOST})",
|
||
)
|
||
parser.add_argument(
|
||
"--port", type=int, default=DEFAULT_LVA_PORT,
|
||
help=f"LVA peripheral API port (default: {DEFAULT_LVA_PORT})",
|
||
)
|
||
parser.add_argument(
|
||
"--debug", action="store_true",
|
||
help="Enable DEBUG logging",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=logging.DEBUG if args.debug else logging.INFO,
|
||
format="%(asctime)s %(levelname)-8s %(name)s %(message)s",
|
||
)
|
||
|
||
asyncio.run(async_main(args.host, args.port))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|