Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b49fd8dc65 | ||
|
|
f20952f260 |
No files matched your search
@@ -65,6 +65,7 @@ Immediately after connecting, LVA sends a snapshot so your client can synchronis
|
||||
"muted": false,
|
||||
"volume": 0.8,
|
||||
"ha_connected": true,
|
||||
"button_controls_disabled": false,
|
||||
"last_stt_text": "set a timer for five minutes",
|
||||
"last_tts_text": "Sure, I've set a timer for five minutes."
|
||||
}
|
||||
@@ -124,13 +125,14 @@ These are all the events LVA emits. Your peripheral script receives them and rea
|
||||
| `disconnected` | — | The TCP connection to Home Assistant was lost. Show a "no connection" animation and keep it until you see `zeroconf` with `status: connected`. Note: if LVA itself is not running, your client will see a WebSocket connection failure instead — treat that the same way. |
|
||||
| `muted` | `{"muted": bool}` | The microphone mute state changed. `true` = muted (show a muted indicator on your LEDs, e.g. red at mic positions), `false` = unmuted. Emitted on every transition in both directions, so peripherals can track mute state without inferring it from `idle`. |
|
||||
| `zeroconf` | `{"status": "getting_started" \| "connected"}` | Reports LVA's connection lifecycle. `getting_started` is emitted at startup before HA connects; `connected` is emitted once the HA TCP handshake completes. Use `connected` to clear a "no connection" animation. |
|
||||
| `button_controls_disabled` | `{"disabled": bool}` | Reflects the "Disable button controls" switch on the Home Assistant device page. `true` = on-board buttons should be ignored by the peripheral; `false` = normal button behaviour. Off by default. Also included directly in the `snapshot` payload for newly-connecting clients. |
|
||||
|
||||
### Timer events
|
||||
|
||||
| Event | Data | Description |
|
||||
|-------|------|-------------|
|
||||
| `timer_ticking` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A timer has been set and is counting down. Show a countdown animation proportional to `seconds_left / total_seconds`. |
|
||||
| `timer_updated` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A running timer was adjusted. Update your countdown display. |
|
||||
| `timer_updated` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A running timer was adjusted. Update your countdown display. Also sent with `seconds_left` of `0`, followed by `idle`, when a timer is cancelled: treat zero as the timer being gone and clear it. A timer expiring normally arrives as `timer_ringing`. |
|
||||
| `timer_ringing` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | The timer has expired and the alarm sound is playing. Show a repeating alert animation. Send `stop_timer_ringing` to dismiss. |
|
||||
|
||||
### Media and volume events
|
||||
@@ -165,6 +167,7 @@ For HA to see your entity, your peripheral must register before HA enumerates th
|
||||
|---------|------|-------------|
|
||||
| `register_light` | `{"name": str, "object_id": str, "effects": [str], "supports_rgb": bool, "supports_brightness": bool}` | Register a Light entity for an LED strip, ring, or single LED. HA exposes it as `light.<satellite>_<object_id>` with on/off, brightness, RGB, and a selectable effect from the declared list. Subsequent HA changes are delivered as `light_command` events. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_light", "data": {"name": "LEDs", "object_id": "leds", "effects": ["Voice Assistant"], "supports_rgb": true, "supports_brightness": true}}` |
|
||||
| `register_button` | `{"name": Button Press, "button_press_event": str}` | Register a Button entity for a physical button on your peripheral. When the user presses the button in HA, LVA emits a `button_press` event to all connected peripherals. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_button"}` |
|
||||
| `register_button_lock` | `{"name": Button Lock, "button_lock_event": str}` | Register a Button Lock entity for a "Disable button controls" switch on the HA device page. When the user toggles it, LVA emits a `button_lock_changed` event to all connected peripherals. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_button_lock"}` |
|
||||
|
||||
### Voice pipeline
|
||||
|
||||
|
||||
@@ -67,6 +67,28 @@ Togges microphone mute. Sends `mute_mic` when unmuted, `unmute_mic` when muted.
|
||||
|
||||
> **Mute button caveat:** The Jabra Speak 410 does not expose the mute button state over HID when not in an active call. The controller detects mute by reading the raw audio stream from PipeWire — all-zero audio indicates hardware mute has engaged. This requires PipeWire to be exposed to the container (see installation).
|
||||
|
||||
### Disabling button controls from Home Assistant
|
||||
|
||||
On connect, the controller registers a **Disable button controls** switch
|
||||
with LVA, which appears on the device page as
|
||||
`switch.<satellite>_disable_button_controls` (off by default).
|
||||
|
||||
When turned **on** in Home Assistant, the call/hang-up button, volume
|
||||
buttons, and mute button all stop sending commands to LVA (and, when
|
||||
`VOLUME_CONTROL=pipewire` is set, volume buttons also stop adjusting the
|
||||
PipeWire sink directly). When turned **off**, normal button behavior
|
||||
resumes immediately.
|
||||
|
||||
> **Note:** Unlike the other example peripherals, the mute button on this
|
||||
> device is also gated by the lock, since it is a physical button command
|
||||
> like any other rather than a dedicated hardware mute switch. If you'd
|
||||
> prefer mute to remain functional while locked, that's a small change in
|
||||
> `main.py`'s `Telephony.mute` / `Volume.mute` handling.
|
||||
|
||||
The lock state is persisted by LVA across restarts. On (re)connect, the
|
||||
controller reads the current value from the `snapshot` event's
|
||||
`button_controls_locked` field.
|
||||
|
||||
---
|
||||
|
||||
## Installation
|
||||
|
||||
@@ -200,6 +200,9 @@ class JabraSpeak:
|
||||
if not vol_ctrl:
|
||||
print("ignoring volume command, env not set")
|
||||
continue
|
||||
if buttons_locked:
|
||||
print("ignoring volume command, buttons locked")
|
||||
continue
|
||||
if event & Volume.vol_up:
|
||||
print("jabra to lva: volume up detected")
|
||||
match vol_ctrl:
|
||||
@@ -250,6 +253,9 @@ async def write_to_jabra(state: LEDs | LEDState):
|
||||
|
||||
|
||||
async def write_to_lva(command: LVACommand, data: dict = None):
|
||||
if buttons_locked:
|
||||
print(f"to lva: buttons locked, ignoring {command}")
|
||||
return
|
||||
if lva_sock:
|
||||
global last_lva_write
|
||||
last_lva_write = command
|
||||
@@ -299,6 +305,8 @@ current_state: None | LVAEvent = None
|
||||
|
||||
muted: bool = False
|
||||
|
||||
buttons_locked: bool = False
|
||||
|
||||
|
||||
async def run_cmd(cmd: list[str]):
|
||||
print("run_cmd ", cmd)
|
||||
@@ -342,12 +350,19 @@ async def wsloop():
|
||||
async with websockets.connect(LVA_WS_URL) as websocket:
|
||||
print(f"Connected to LVA at {LVA_WS_URL}")
|
||||
lva_sock = websocket
|
||||
await websocket.send(json.dumps({"command": "register_button_lock"}))
|
||||
while True:
|
||||
data = await websocket.recv()
|
||||
print(f"from lva: {data}")
|
||||
json_data = json.loads(data)
|
||||
if json_data["event"] == "snapshot":
|
||||
await set_mute(json_data["data"]["muted"], write_lva=False)
|
||||
global buttons_locked
|
||||
buttons_locked = json_data["data"].get("button_controls_locked", False)
|
||||
elif json_data["event"] == "button_lock_changed":
|
||||
buttons_locked = json_data["data"].get("locked", False)
|
||||
print(f"Button controls {'locked' if buttons_locked else 'unlocked'}")
|
||||
continue
|
||||
global current_state
|
||||
try:
|
||||
current_state = LVAEvent(json_data["event"])
|
||||
|
||||
@@ -126,6 +126,26 @@ Send commands to LVA by clicking buttons:
|
||||
- **Single press**, **Double press**, **Triple press**, **Long press** - Simulate physical button inputs
|
||||
- Useful for testing button-based interactions if LVA is connected to hardware buttons
|
||||
|
||||
### Button Lock Testing
|
||||
|
||||
The console registers a **Disable button controls** switch with LVA
|
||||
(`register_button_lock`) so it can be used to test and demonstrate the
|
||||
button-lock feature that hardware peripherals implement.
|
||||
|
||||
- The **Buttons Locked** field in the status area reflects the current
|
||||
switch state, sourced from the `snapshot` event and updated live on
|
||||
`button_lock_changed` events.
|
||||
- When the switch is turned **on** from the Home Assistant device page,
|
||||
every button in the **Commands** section is disabled (greyed out) and
|
||||
clicking it is a no-op, mirroring how a real peripheral should ignore
|
||||
its physical buttons while locked.
|
||||
- Turning the switch **off** re-enables the command buttons immediately.
|
||||
|
||||
This is useful for verifying the lock behavior end-to-end without needing
|
||||
physical hardware: toggle the switch in Home Assistant and confirm the
|
||||
console's command grid disables/enables accordingly, and that no
|
||||
commands are sent to LVA while locked.
|
||||
|
||||
## Example Workflow
|
||||
|
||||
1. **Start LVA server** with WebSocket peripheral API enabled
|
||||
|
||||
@@ -117,6 +117,10 @@
|
||||
<span class="meta-label">HA Connected</span>
|
||||
<span class="meta-value" id="meta-ha">—</span>
|
||||
</div>
|
||||
<div class="meta-item">
|
||||
<span class="meta-label">Buttons Locked</span>
|
||||
<span class="meta-value" id="meta-locked">—</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Volume bar -->
|
||||
@@ -311,6 +315,9 @@
|
||||
// Register the Button entity with LVA so HA can receive button events.
|
||||
ws.send(JSON.stringify({command: 'register_button'}));
|
||||
log('sent', '↑ register_button');
|
||||
|
||||
ws.send(JSON.stringify({command: 'register_button_lock'}));
|
||||
log('sent', '↑ register_button_lock');
|
||||
};
|
||||
|
||||
ws.onclose = () => {
|
||||
@@ -340,6 +347,8 @@
|
||||
metaMuted.textContent = data.muted !== undefined ? (data.muted ? 'yes' : 'no') : '—';
|
||||
metaHa.textContent = data.ha_connected !== undefined ? (data.ha_connected ? 'yes' : 'no') : '—';
|
||||
volBar.style.width = (data.volume !== undefined ? data.volume * 100 : 0).toFixed(0) + '%';
|
||||
buttonsLocked = !!data.button_controls_locked;
|
||||
applyButtonLockUI();
|
||||
|
||||
// Restore last conversation context
|
||||
if (data.last_stt_text) {
|
||||
@@ -398,6 +407,10 @@
|
||||
timerBox.classList.remove('hidden');
|
||||
}
|
||||
}
|
||||
if (ev === 'button_lock_changed' && data) {
|
||||
buttonsLocked = !!data.locked;
|
||||
applyButtonLockUI();
|
||||
}
|
||||
|
||||
setState(ev);
|
||||
log('recv', ev, dataStr);
|
||||
@@ -416,6 +429,10 @@
|
||||
log('err-item', '✗ not connected — connect first');
|
||||
return;
|
||||
}
|
||||
if (buttonsLocked) {
|
||||
log('err-item', '✗ button controls locked — ignoring ' + command);
|
||||
return;
|
||||
}
|
||||
const msg = { command };
|
||||
if (data) msg.data = data;
|
||||
ws.send(JSON.stringify(msg));
|
||||
@@ -435,4 +452,15 @@
|
||||
send('set_volume', { volume: clamped });
|
||||
volumeInput.value = ''; // Clear after sending
|
||||
}
|
||||
|
||||
let buttonsLocked = false;
|
||||
|
||||
function applyButtonLockUI() {
|
||||
document.getElementById('meta-locked').textContent = buttonsLocked ? 'yes' : 'no';
|
||||
document.querySelectorAll('.cmd-grid .cmd-btn').forEach((el) => {
|
||||
el.disabled = buttonsLocked;
|
||||
el.style.opacity = buttonsLocked ? '0.4' : '';
|
||||
el.style.cursor = buttonsLocked ? 'not-allowed' : '';
|
||||
});
|
||||
}
|
||||
</script>
|
||||
@@ -133,6 +133,22 @@ Multi-press commands are useful for triggering custom Home Assistant automations
|
||||
|
||||
These commands are exposed as button press events to Home Assistant (via the registered button sensor), allowing you to create custom automations via `button_press_event` triggers.
|
||||
|
||||
### Disabling button controls from Home Assistant
|
||||
|
||||
On connect, the controller registers a **Disable button controls** switch
|
||||
with LVA, which appears on the device page as
|
||||
`switch.<satellite>_disable_button_controls` (off by default).
|
||||
|
||||
When turned **on** in Home Assistant, the onboard button's single-press
|
||||
context action and all multi-press gestures (double, triple, long press)
|
||||
stop sending commands to LVA. When turned **off**, normal behavior
|
||||
resumes immediately.
|
||||
|
||||
The lock state is persisted by LVA across restarts. On (re)connect, the
|
||||
controller reads the current value from the `snapshot` event's
|
||||
`button_controls_locked` field, so it never has a stale/unknown lock
|
||||
state after a reconnect.
|
||||
|
||||
---
|
||||
|
||||
## Installation
|
||||
|
||||
@@ -194,6 +194,7 @@ class SharedState:
|
||||
# Monotonic deadline used by _timer_tick to fade brightness
|
||||
# smoothly between sparse timer_updated events.
|
||||
self.timer_ends_at: float = 0.0
|
||||
self.buttons_locked: bool = False
|
||||
|
||||
def update(self, **kwargs) -> None:
|
||||
with self._lock:
|
||||
@@ -224,6 +225,7 @@ class SharedState:
|
||||
"light_green": self.light_green,
|
||||
"light_blue": self.light_blue,
|
||||
"timer_ends_at": self.timer_ends_at,
|
||||
"buttons_locked": self.buttons_locked,
|
||||
}
|
||||
|
||||
|
||||
@@ -601,6 +603,9 @@ class ButtonHandler:
|
||||
self._on_press()
|
||||
|
||||
def _send(self, command: str) -> None:
|
||||
if self._state.snapshot["buttons_locked"]:
|
||||
_LOGGER.debug("Button controls locked — ignoring %s", command)
|
||||
return
|
||||
_LOGGER.info("Button → %s", command)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._queue.put(command), self._loop
|
||||
@@ -665,6 +670,9 @@ class ButtonMultipressHandler:
|
||||
self._single_click_action = action
|
||||
|
||||
def _send(self, command: str) -> None:
|
||||
if self._state.snapshot["buttons_locked"]:
|
||||
_LOGGER.debug("Button controls locked — ignoring %s", command)
|
||||
return
|
||||
_LOGGER.info("Button → %s", command)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._queue.put(command), self._loop
|
||||
@@ -805,7 +813,8 @@ class LVAClient:
|
||||
# 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"}))
|
||||
await ws.send(json.dumps({"command": "register_button"}))
|
||||
await ws.send(json.dumps({"command": "register_button_lock"}))
|
||||
recv_task = asyncio.create_task(self._recv_loop(ws))
|
||||
send_task = asyncio.create_task(self._send_loop(ws))
|
||||
done, pending = await asyncio.wait(
|
||||
@@ -848,6 +857,7 @@ class LVAClient:
|
||||
muted=muted,
|
||||
volume=data.get("volume", 1.0),
|
||||
ha_connected=data.get("ha_connected", False),
|
||||
buttons_locked=data.get("button_controls_locked", False),
|
||||
)
|
||||
if muted:
|
||||
self._animator.set_state(AssistState.MUTED)
|
||||
@@ -857,6 +867,12 @@ class LVAClient:
|
||||
self._animator.set_state(AssistState.IDLE)
|
||||
return
|
||||
|
||||
elif event == "button_lock_changed":
|
||||
locked = data.get("locked", False)
|
||||
self._state.update(buttons_locked=locked)
|
||||
_LOGGER.info("Button controls %s", "locked" if locked else "unlocked")
|
||||
return
|
||||
|
||||
if event == "wake_word_detected":
|
||||
self._state.update(assist_state=AssistState.WAKE_WORD)
|
||||
|
||||
|
||||
@@ -181,6 +181,22 @@ Each press (or hold-repeat step) changes the hue by 10°, allowing smooth naviga
|
||||
3. Release the volume button once you like the color, then release the action button to confirm and exit color mode
|
||||
4. The new color persists in Home Assistant
|
||||
|
||||
#### Disabling button controls from Home Assistant
|
||||
|
||||
On connect, the controller registers a **Disable button controls** switch
|
||||
with LVA, which appears on the device page as
|
||||
`switch.<satellite>_disable_button_controls` (off by default).
|
||||
|
||||
When turned **on** in Home Assistant, the action button (context command), volume buttons, mute button, and the hold + volume color-wheel gesture are all ignored — pressing them has no effect on LVA or the LED ring color.
|
||||
|
||||
When turned **off**, button behavior resumes immediately.
|
||||
|
||||
The switch state is persisted by LVA across restarts, so a lock set from
|
||||
Home Assistant survives a reboot of the satellite. On reconnect, the
|
||||
controller reads the current state from the `snapshot` event's
|
||||
`button_controls_locked` field and applies it immediately, without
|
||||
waiting for a change.
|
||||
|
||||
---
|
||||
|
||||
## LED ring animations
|
||||
|
||||
@@ -291,6 +291,7 @@ class SharedState:
|
||||
# pressed" tactile feedback in LEDRing (only applied when the HA
|
||||
# Light entity is off — see LEDRing._anim_action_held).
|
||||
self.action_button_down: bool = False
|
||||
self.buttons_locked: bool = False
|
||||
|
||||
def update(self, **kwargs) -> None:
|
||||
with self._lock:
|
||||
@@ -318,6 +319,7 @@ class SharedState:
|
||||
"hsv_saturation": self.hsv_saturation,
|
||||
"hsv_value": self.hsv_value,
|
||||
"action_button_down": self.action_button_down,
|
||||
"buttons_locked": self.buttons_locked,
|
||||
}
|
||||
|
||||
|
||||
@@ -1090,8 +1092,12 @@ class ButtonMultipressHandler:
|
||||
self.button_pin = BTN_ACTION
|
||||
self.last_press_time = 0
|
||||
self.press_count = 0
|
||||
self._state = state
|
||||
|
||||
def _send(self, command: str) -> None:
|
||||
if self._state.snapshot["buttons_locked"]:
|
||||
_LOGGER.debug("Button controls locked — ignoring %s", command)
|
||||
return
|
||||
_LOGGER.info("Button → %s", command)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._queue.put(command), self._loop
|
||||
@@ -1189,7 +1195,8 @@ class LVAClient:
|
||||
# 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"}))
|
||||
await ws.send(json.dumps({"command": "register_button"}))
|
||||
await ws.send(json.dumps({"command": "register_button_lock"}))
|
||||
recv_task = asyncio.create_task(self._recv_loop(ws))
|
||||
send_task = asyncio.create_task(self._send_loop(ws))
|
||||
done, pending = await asyncio.wait(
|
||||
@@ -1248,6 +1255,7 @@ class LVAClient:
|
||||
# defensively in case a future LVA version adds it.
|
||||
volume_muted=data.get("volume_muted", False),
|
||||
ha_connected=data.get("ha_connected", False),
|
||||
buttons_locked=data.get("button_controls_locked", False),
|
||||
)
|
||||
snap = self._state.snapshot
|
||||
anim = state_to_animation(
|
||||
@@ -1256,6 +1264,12 @@ class LVAClient:
|
||||
self._leds.set_animation(anim)
|
||||
return
|
||||
|
||||
elif event == "button_lock_changed":
|
||||
locked = data.get("locked", False)
|
||||
self._state.update(buttons_locked=locked)
|
||||
_LOGGER.info("Button controls %s", "locked" if locked else "unlocked")
|
||||
return
|
||||
|
||||
# --- Voice pipeline events -----------------------------------------
|
||||
elif event == "wake_word_detected":
|
||||
self._state.update(assist_state=AssistState.WAKE_WORD)
|
||||
|
||||
@@ -383,6 +383,15 @@ async def main() -> None:
|
||||
initial_threshold = initial_stop_word_threshold(preferences.stop_word_sensitivity)
|
||||
preferences.stop_word_sensitivity = initial_threshold
|
||||
|
||||
# Load button-lock state from preferences on startup. This must happen
|
||||
# before ServerState is constructed (and therefore before the peripheral
|
||||
# WebSocket server can accept any connection), because the very first
|
||||
# snapshot sent to a (re)connecting peripheral must already reflect the
|
||||
# persisted value — peripherals only learn the lock state from that
|
||||
# snapshot or a later button_lock_changed event, and nothing re-sends
|
||||
# a corrected snapshot after the fact.
|
||||
initial_button_controls_locked = bool(preferences.button_controls_locked)
|
||||
|
||||
if args.enable_thinking_sound:
|
||||
preferences.thinking_sound = 1
|
||||
|
||||
@@ -446,6 +455,7 @@ async def main() -> None:
|
||||
download_dir=args.download_dir,
|
||||
volume=initial_volume,
|
||||
stop_word_threshold=initial_threshold,
|
||||
button_controls_locked=initial_button_controls_locked,
|
||||
mic_volume=preferences.mic_volume,
|
||||
mic_auto_gain=preferences.mic_auto_gain,
|
||||
mic_noise_suppression=preferences.mic_noise_suppression,
|
||||
|
||||
@@ -801,6 +801,65 @@ class ButtonEventSensorEntity(ESPHomeEntity):
|
||||
)
|
||||
|
||||
|
||||
class ButtonLockEntity(ESPHomeEntity):
|
||||
"""Switch entity for disabling a peripheral's on-board buttons.
|
||||
|
||||
Registered by a peripheral via register_button_lock, mirroring the
|
||||
opt-in pattern used by register_light. LVA only exposes the switch
|
||||
in Home Assistant and owns its state; it has no opinion on what
|
||||
"locked" means for the hardware. When Home Assistant changes the
|
||||
switch, on_changed fires so the peripheral API server can broadcast
|
||||
a button_lock_changed event back to the peripheral, which decides
|
||||
whether to keep executing button-triggered commands.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
server: APIServer,
|
||||
key: int,
|
||||
name: str,
|
||||
object_id: str,
|
||||
get_locked: Callable[[], bool],
|
||||
set_locked: Callable[[bool], None],
|
||||
icon: str = "mdi:button-pointer",
|
||||
) -> None:
|
||||
ESPHomeEntity.__init__(self, server)
|
||||
self.key = key
|
||||
self.name = name
|
||||
self.object_id = object_id
|
||||
self.icon = icon
|
||||
self._get_locked = get_locked
|
||||
self._set_locked = set_locked
|
||||
self._switch_state = self._get_locked() # Sync internal state with actual value on init
|
||||
|
||||
def update_get_locked(self, get_locked: Callable[[], bool]) -> None:
|
||||
self._get_locked = get_locked
|
||||
|
||||
def update_set_locked(self, set_locked: Callable[[bool], None]) -> None:
|
||||
self._set_locked = set_locked
|
||||
|
||||
def sync_with_state(self) -> None:
|
||||
self._switch_state = self._get_locked()
|
||||
|
||||
def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
|
||||
if isinstance(msg, SwitchCommandRequest) and (msg.key == self.key):
|
||||
new_state = bool(msg.state)
|
||||
self._switch_state = new_state
|
||||
self._set_locked(new_state)
|
||||
yield SwitchStateResponse(key=self.key, state=self._switch_state)
|
||||
elif isinstance(msg, ListEntitiesRequest):
|
||||
yield ListEntitiesSwitchResponse(
|
||||
object_id=self.object_id,
|
||||
key=self.key,
|
||||
name=self.name,
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
icon=self.icon,
|
||||
)
|
||||
elif isinstance(msg, SubscribeHomeAssistantStatesRequest):
|
||||
self.sync_with_state()
|
||||
yield SwitchStateResponse(key=self.key, state=self._switch_state)
|
||||
|
||||
|
||||
# Backward compatibility export aliases
|
||||
__all__ = [
|
||||
"ESPHomeEntity",
|
||||
@@ -809,6 +868,7 @@ __all__ = [
|
||||
"ThinkingSoundEntity",
|
||||
"LEDLightEntity",
|
||||
"ButtonEventSensorEntity",
|
||||
"ButtonLockEntity",
|
||||
"WakeWord1SensitivityNumberEntity",
|
||||
"WakeWord2SensitivityNumberEntity",
|
||||
"StopWordSensitivityNumberEntity",
|
||||
|
||||
@@ -16,6 +16,7 @@ if TYPE_CHECKING:
|
||||
|
||||
from .entity import (
|
||||
ButtonEventSensorEntity,
|
||||
ButtonLockEntity,
|
||||
ESPHomeEntity,
|
||||
LEDLightEntity,
|
||||
MediaPlayerEntity,
|
||||
@@ -85,6 +86,7 @@ class Preferences:
|
||||
active_wake_words: List[Optional[str]] = field(default_factory=list)
|
||||
volume: Optional[float] = None
|
||||
thinking_sound: int = 0 # 0 = disabled, 1 = enabled
|
||||
button_controls_locked: int = 0 # 0 = buttons enabled (default), 1 = buttons locked
|
||||
wake_word_1_sensitivity: Optional[float] = None
|
||||
wake_word_2_sensitivity: Optional[float] = None
|
||||
stop_word_sensitivity: Optional[float] = None
|
||||
@@ -146,6 +148,13 @@ class ServerState:
|
||||
# Survives HA reconnects so the entity is re-registered automatically.
|
||||
pending_button: bool = False
|
||||
|
||||
# True once a peripheral sends register_button_lock. Gates creation of
|
||||
# ButtonLockEntity so the HA device page only shows the "Disable button
|
||||
# controls" switch when hardware that actually supports it is present.
|
||||
# Survives HA reconnects so the entity is re-registered automatically.
|
||||
pending_button_lock: bool = False
|
||||
button_lock_entity: "Optional[ButtonLockEntity]" = None
|
||||
|
||||
# Optional peripheral WebSocket API (LEDs, buttons, HAT boards).
|
||||
# Assigned in __main__ before the event loop starts.
|
||||
peripheral_api: "Optional[Any]" = None # PeripheralAPIServer at runtime
|
||||
@@ -159,6 +168,7 @@ class ServerState:
|
||||
wake_words_changed: bool = False
|
||||
refractory_seconds: float = 2.0
|
||||
thinking_sound_enabled: bool = False
|
||||
button_controls_locked: bool = False
|
||||
output_only: bool = False
|
||||
muted: bool = False
|
||||
connected: bool = False
|
||||
|
||||
@@ -54,6 +54,16 @@ Feedback events emitted by LVA
|
||||
effect names are those the peripheral declared at
|
||||
registration; e.g. "Voice Assistant" runs the pipeline
|
||||
animations.
|
||||
button_lock_changed data: {"locked": bool}
|
||||
Fires when the user toggles the "Disable button controls"
|
||||
switch on the Home Assistant device page for a peripheral
|
||||
that registered it via register_button_lock. The current
|
||||
value is also included as "button_controls_locked" in the
|
||||
snapshot payload for newly-connecting/reconnecting clients.
|
||||
A peripheral should stop executing button-triggered LVA
|
||||
commands (start_listening, stop_pipeline, mute_mic/
|
||||
unmute_mic, volume_up/down, button_*_press, etc.) while
|
||||
locked is true.
|
||||
|
||||
Commands accepted from the peripheral container
|
||||
------------------------------------------------
|
||||
@@ -92,6 +102,17 @@ Commands accepted from the peripheral container
|
||||
events to Home Assistant when the corresponding
|
||||
button_* commands are sent. Send once after connecting;
|
||||
duplicate registrations are ignored.
|
||||
register_button_lock
|
||||
The peripheral declares that it has on-board buttons whose
|
||||
behavior can be toggled from Home Assistant. LVA creates a
|
||||
"Disable button controls" switch entity (visible as
|
||||
switch.<satellite>_disable_button_controls, disabled/off
|
||||
by default) and routes HA changes back to the peripheral
|
||||
as button_lock_changed events. The peripheral is
|
||||
responsible for honoring the lock state — LVA does not
|
||||
gate anything on its end. Send once after connecting;
|
||||
duplicate registrations are ignored and preserve the
|
||||
current lock state.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -139,6 +160,7 @@ class LVAEvent(str, Enum):
|
||||
VOLUME_MUTED = "volume_muted"
|
||||
ZEROCONF = "zeroconf"
|
||||
LIGHT_COMMAND = "light_command"
|
||||
BUTTON_LOCK_CHANGED = "button_lock_changed"
|
||||
|
||||
|
||||
class LVACommand(str, Enum):
|
||||
@@ -161,6 +183,7 @@ class LVACommand(str, Enum):
|
||||
BUTTON_LONG_PRESS = "button_long_press"
|
||||
REGISTER_LIGHT = "register_light"
|
||||
REGISTER_BUTTON = "register_button"
|
||||
REGISTER_BUTTON_LOCK = "register_button_lock"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -280,6 +303,7 @@ class PeripheralAPIServer:
|
||||
"volume": round(state.volume, 3),
|
||||
"volume_muted": state.volume == 0.0,
|
||||
"ha_connected": state.connected,
|
||||
"button_controls_locked": state.button_controls_locked,
|
||||
"last_stt_text": self._last_stt_text,
|
||||
"last_tts_text": self._last_tts_text,
|
||||
},
|
||||
@@ -481,6 +505,9 @@ class PeripheralAPIServer:
|
||||
elif command == LVACommand.REGISTER_BUTTON:
|
||||
self._register_button(satellite)
|
||||
|
||||
elif command == LVACommand.REGISTER_BUTTON_LOCK:
|
||||
self._register_button_lock(satellite)
|
||||
|
||||
def _register_light(self, data: Dict[str, Any], satellite: Any) -> None:
|
||||
"""Register a Light declared by a peripheral.
|
||||
|
||||
@@ -550,6 +577,35 @@ class PeripheralAPIServer:
|
||||
|
||||
self._schedule_ha_reconnect_for_late_entity("button", "button_press_event")
|
||||
|
||||
def _register_button_lock(self, satellite: Any) -> None:
|
||||
"""Register the button-lock switch declared by a peripheral.
|
||||
|
||||
Idempotent: repeat registrations from a reconnecting peripheral
|
||||
are a no-op — the existing entity and its current lock state
|
||||
are preserved.
|
||||
|
||||
When the satellite is already running, the entity is materialised
|
||||
immediately so a subsequent switch toggle routes correctly. HA
|
||||
only sees the new entity after the integration is reloaded, but
|
||||
the same startup-wait window that applies to register_light
|
||||
applies here too (see --peripheral-startup-wait).
|
||||
"""
|
||||
state = self._state
|
||||
if state is None:
|
||||
return
|
||||
|
||||
if state.pending_button_lock:
|
||||
# Already registered; keep the existing entity and its state.
|
||||
return
|
||||
|
||||
state.pending_button_lock = True
|
||||
_LOGGER.info("Button lock switch registered by peripheral")
|
||||
|
||||
if satellite is not None:
|
||||
satellite.register_pending_button_lock()
|
||||
|
||||
self._schedule_ha_reconnect_for_late_entity("button_lock", "disable_button_controls")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@@ -52,6 +52,7 @@ from pyopen_wakeword import OpenWakeWord
|
||||
from .api_server import APIServer
|
||||
from .entity import (
|
||||
ButtonEventSensorEntity,
|
||||
ButtonLockEntity,
|
||||
LEDLightEntity,
|
||||
MediaPlayerEntity,
|
||||
MicSettingEntity,
|
||||
@@ -335,6 +336,9 @@ class VoiceSatelliteProtocol(APIServer):
|
||||
# button support before this satellite was constructed (e.g. on an HA
|
||||
# reconnect while the peripheral container stayed connected to LVA).
|
||||
self.register_pending_button()
|
||||
# Materialise ButtonLockEntity if a peripheral already registered
|
||||
# the button-lock switch before this satellite was constructed.
|
||||
self.register_pending_button_lock()
|
||||
|
||||
# ---- Instance variables ----
|
||||
|
||||
@@ -431,6 +435,40 @@ class VoiceSatelliteProtocol(APIServer):
|
||||
self.state.button_event_sensor_entity = entity
|
||||
_LOGGER.info("Button event sensor entity materialised")
|
||||
|
||||
def register_pending_button_lock(self) -> None:
|
||||
"""Materialise the button-lock switch once a peripheral has registered it.
|
||||
|
||||
Called from __init__ (handles HA reconnects where the peripheral
|
||||
container stayed connected to LVA and pending_button_lock is already
|
||||
True) and from PeripheralAPIServer._register_button_lock() when the
|
||||
command arrives at runtime.
|
||||
|
||||
Safe to call multiple times: idempotent — if the entity already
|
||||
exists it is only reattached to the current satellite server
|
||||
instance, preserving its current lock state.
|
||||
"""
|
||||
if not self.state.pending_button_lock:
|
||||
return
|
||||
|
||||
if self.state.button_lock_entity is not None:
|
||||
self.state.button_lock_entity.server = self
|
||||
if self.state.button_lock_entity not in self.state.entities:
|
||||
self.state.entities.append(self.state.button_lock_entity)
|
||||
return
|
||||
|
||||
entity = ButtonLockEntity(
|
||||
server=self,
|
||||
key=len(self.state.entities),
|
||||
name="Disable button controls",
|
||||
object_id="disable_button_controls",
|
||||
get_locked=lambda: self.state.button_controls_locked,
|
||||
set_locked=self._set_button_controls_locked,
|
||||
)
|
||||
self.state.entities.append(entity)
|
||||
self.state.button_lock_entity = entity
|
||||
entity.sync_with_state()
|
||||
_LOGGER.info("Button lock switch entity materialised")
|
||||
|
||||
def _on_led_light_changed(self, object_id: str) -> None:
|
||||
"""Forward an HA Light entity change to peripherals as light_command.
|
||||
|
||||
@@ -442,6 +480,10 @@ class VoiceSatelliteProtocol(APIServer):
|
||||
return
|
||||
self._emit(LVAEvent.LIGHT_COMMAND, entity.state_dict())
|
||||
|
||||
def _on_button_lock_changed(self, is_locked: bool) -> None:
|
||||
"""Forward a button-lock switch change to peripherals as button_lock_changed."""
|
||||
self._emit(LVAEvent.BUTTON_LOCK_CHANGED, {"locked": is_locked})
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Mute / thinking sound
|
||||
# ------------------------------------------------------------------
|
||||
@@ -457,6 +499,21 @@ class VoiceSatelliteProtocol(APIServer):
|
||||
pass
|
||||
self.state.save_preferences()
|
||||
|
||||
def _set_button_controls_locked(self, new_state: bool) -> None:
|
||||
self.state.button_controls_locked = bool(new_state)
|
||||
self.state.preferences.button_controls_locked = 1 if self.state.button_controls_locked else 0
|
||||
self.state.save_preferences()
|
||||
|
||||
if self.state.button_controls_locked:
|
||||
_LOGGER.debug("On-board button controls locked")
|
||||
else:
|
||||
_LOGGER.debug("On-board button controls unlocked")
|
||||
|
||||
if self.state.button_lock_entity is not None:
|
||||
self.state.button_lock_entity.sync_with_state()
|
||||
|
||||
self._emit(LVAEvent.BUTTON_LOCK_CHANGED, {"locked": self.state.button_controls_locked})
|
||||
|
||||
def _set_sensitivity_1(self, new_value: float) -> None:
|
||||
self.state.wake_word_1_threshold = float(new_value)
|
||||
self.state.preferences.wake_word_1_sensitivity = float(new_value)
|
||||
@@ -599,6 +656,12 @@ class VoiceSatelliteProtocol(APIServer):
|
||||
self._emit(LVAEvent.TIMER_UPDATED, timer_data)
|
||||
|
||||
elif event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED:
|
||||
# Report the cancel as a zeroed update before going idle. Every
|
||||
# shipped peripheral already reads `seconds_left` from
|
||||
# `timer_updated`, and none of them treats zero as "ringing",
|
||||
# that is `timer_ringing`, so the countdown clears without any
|
||||
# peripheral having to learn a new event.
|
||||
self._emit(LVAEvent.TIMER_UPDATED, {**timer_data, "seconds_left": 0})
|
||||
self._emit(LVAEvent.IDLE)
|
||||
|
||||
elif event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_FINISHED:
|
||||
|
||||
@@ -367,3 +367,69 @@ class TestConnectionLost:
|
||||
sat = make_satellite(tmp_path)
|
||||
sat.connection_lost(None)
|
||||
sat.state.tts_player.stop.assert_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Timer events
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _timer_msg(seconds_left=60, total_seconds=300, timer_id="t1", name="pasta"):
|
||||
msg = MagicMock()
|
||||
msg.timer_id = timer_id
|
||||
msg.name = name
|
||||
msg.total_seconds = total_seconds
|
||||
msg.seconds_left = seconds_left
|
||||
return msg
|
||||
|
||||
|
||||
class TestHandleTimerEvent:
|
||||
def _sat(self, tmp_path):
|
||||
sat = make_satellite(tmp_path)
|
||||
sat.state.peripheral_api = MagicMock()
|
||||
return sat
|
||||
|
||||
def _emitted(self, sat):
|
||||
return [c.args for c in sat.state.peripheral_api.emit_event_sync.call_args_list]
|
||||
|
||||
def test_started_emits_ticking_with_data(self, tmp_path):
|
||||
from aioesphomeapi.model import VoiceAssistantTimerEventType
|
||||
|
||||
from linux_voice_assistant.peripheral_api import LVAEvent
|
||||
|
||||
sat = self._sat(tmp_path)
|
||||
sat.handle_timer_event(VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_STARTED, _timer_msg())
|
||||
event, data = self._emitted(sat)[0]
|
||||
assert event == LVAEvent.TIMER_TICKING
|
||||
assert data["seconds_left"] == 60
|
||||
|
||||
def test_cancelled_zeroes_the_countdown(self, tmp_path):
|
||||
from aioesphomeapi.model import VoiceAssistantTimerEventType
|
||||
|
||||
from linux_voice_assistant.peripheral_api import LVAEvent
|
||||
|
||||
sat = self._sat(tmp_path)
|
||||
sat.handle_timer_event(
|
||||
VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED,
|
||||
_timer_msg(seconds_left=42),
|
||||
)
|
||||
emitted = self._emitted(sat)
|
||||
assert [e for e, *_ in emitted] == [LVAEvent.TIMER_UPDATED, LVAEvent.IDLE]
|
||||
|
||||
_, data = emitted[0]
|
||||
assert data["seconds_left"] == 0
|
||||
# The identity is preserved, so a peripheral tracking several timers
|
||||
# knows which one went away.
|
||||
assert data["id"] == "t1"
|
||||
assert data["total_seconds"] == 300
|
||||
|
||||
# Peripherals that re-check a running timer on `idle` must see the zeroed
|
||||
# update first, or they stay in the countdown state.
|
||||
def test_cancelled_still_ends_with_idle(self, tmp_path):
|
||||
from aioesphomeapi.model import VoiceAssistantTimerEventType
|
||||
|
||||
from linux_voice_assistant.peripheral_api import LVAEvent
|
||||
|
||||
sat = self._sat(tmp_path)
|
||||
sat.handle_timer_event(VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED, _timer_msg())
|
||||
assert self._emitted(sat)[-1][0] == LVAEvent.IDLE
|
||||
Reference in new issue
Block a user