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

Add LVA Tray Client

Browse Source 
Provide a simple tray client to show the status of the LVA running on a Linux desktop.
This commit is contained in:
imonlinux
2025-11-26 19:51:17 -08:00
committed by GitHub
parent 11c0e4fbb6
commit 0a2c227cad
1 changed files with 499 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lva_tray_client.py
+499
View File
@@ -0,0 +1,499 @@
#!/usr/bin/env python3
"""
LVA Tray Client
- Reads linux_voice_assistant/config.json to discover:
- app.name -> used for device_id
- mqtt.host / port / username / password
- Subscribes to the LVA's MQTT topics to:
- Track availability (online/offline)
- Track mute state
- Track per-state LED color (idle/listening/thinking/responding/error)
- Shows a system tray icon whose color matches the current LVA state.
- Provides a tray menu to start/stop/restart the LVA systemd --user service
and toggle microphone mute.
Idle state now **also uses the MQTT color**, not a fixed purple.
"""
import argparse
import json
import logging
import os
import subprocess
import sys
from pathlib import Path
from typing import Dict, Optional
import paho.mqtt.client as mqtt
from PyQt5 import QtCore, QtGui, QtWidgets
_LOGGER = logging.getLogger("lva_tray_client")
# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------
def slugify_device_id(name: str) -> str:
"""Convert display name to a consistent device_id."""
return name.strip().lower().replace(" ", "_")
def load_config(config_path: Optional[Path]) -> dict:
"""Load LVA config.json."""
if config_path is None:
# Default: same layout you’re using now:
# <repo_root>/linux_voice_assistant/config.json
base = Path(__file__).resolve().parent
config_path = base / "linux_voice_assistant" / "config.json"
_LOGGER.info("Using config file: %s", config_path)
with config_path.open("r", encoding="utf-8") as f:
cfg = json.load(f)
app_cfg = cfg.get("app", {})
mqtt_cfg = cfg.get("mqtt", {})
name = app_cfg.get("name", "Linux Voice Assistant")
device_id = slugify_device_id(name)
host = mqtt_cfg.get("host", "127.0.0.1")
port = int(mqtt_cfg.get("port", 1883))
username = mqtt_cfg.get("username") or None
password = mqtt_cfg.get("password") or None
cfg["_resolved"] = {
"name": name,
"device_id": device_id,
"mqtt_enabled": True,
"host": host,
"port": port,
"username": username,
"password": password,
}
_LOGGER.info(
"Loaded config: name=%s device_id=%s mqtt_enabled=%s host=%s port=%s",
name,
device_id,
True,
host,
port,
)
return cfg
# ---------------------------------------------------------------------------
# Tray client
# ---------------------------------------------------------------------------
class LvaTrayClient(QtWidgets.QSystemTrayIcon):
"""System tray integration for the Linux Voice Assistant."""
STATES = ["idle", "listening", "thinking", "responding", "error"]
def __init__(self, app: QtWidgets.QApplication, cfg: dict):
# QSystemTrayIcon init
super().__init__(parent=None)
self._app = app
resolved = cfg["_resolved"]
self._device_name = resolved["name"]
self._device_id = resolved["device_id"]
self._mqtt_host = resolved["host"]
self._mqtt_port = resolved["port"]
self._mqtt_username = resolved["username"]
self._mqtt_password = resolved["password"]
self._topic_prefix = f"lva/{self._device_id}"
# Current state
self._available: bool = False
self._muted: bool = False
self._current_state: str = "idle"
# Default colors per state (fallbacks)
self._default_colors: Dict[str, QtGui.QColor] = {
"idle": QtGui.QColor(128, 0, 255), # purple
"listening": QtGui.QColor(0, 0, 255), # blue
"thinking": QtGui.QColor(255, 255, 0), # yellow
"responding": QtGui.QColor(0, 255, 0), # green
"error": QtGui.QColor(255, 165, 0), # orange
}
# Last MQTT-derived color per state (idle included!)
self._last_color_by_state: Dict[str, QtGui.QColor] = dict(
self._default_colors
)
# Build context menu
self._build_menu()
# Initial icon: offline
self._set_icon_by_key("offline")
# MQTT setup
self._client = mqtt.Client()
if self._mqtt_username:
self._client.username_pw_set(self._mqtt_username, self._mqtt_password)
self._client.on_connect = self._on_connect
self._client.on_message = self._on_message
self._client.on_disconnect = self._on_disconnect
_LOGGER.debug(
"MQTT connecting to %s:%s", self._mqtt_host, self._mqtt_port
)
try:
self._client.connect(self._mqtt_host, self._mqtt_port, 60)
self._client.loop_start()
except Exception: # noqa: BLE001
_LOGGER.exception("Failed to connect to MQTT broker")
# Show tray icon
self.setVisible(True)
_LOGGER.info(
"LVA Tray Client started for device_id=%s", self._device_id
)
# ------------------------------------------------------------------
# Menu / actions
# ------------------------------------------------------------------
def _build_menu(self):
menu = QtWidgets.QMenu()
# Status label
self._status_action = QtWidgets.QAction("LVA Tray Client", self)
self._status_action.setEnabled(False)
menu.addAction(self._status_action)
menu.addSeparator()
# Service control
start_action = QtWidgets.QAction("Start LVA", self)
start_action.triggered.connect(self._start_lva)
menu.addAction(start_action)
stop_action = QtWidgets.QAction("Stop LVA", self)
stop_action.triggered.connect(self._stop_lva)
menu.addAction(stop_action)
restart_action = QtWidgets.QAction("Restart LVA", self)
restart_action.triggered.connect(self._restart_lva)
menu.addAction(restart_action)
menu.addSeparator()
# Mute toggle
self._mute_action = QtWidgets.QAction("Mute Microphone", self)
self._mute_action.setCheckable(True)
self._mute_action.triggered.connect(self._toggle_mute)
menu.addAction(self._mute_action)
menu.addSeparator()
# Quit
quit_action = QtWidgets.QAction("Quit Tray Client", self)
quit_action.triggered.connect(self._quit)
menu.addAction(quit_action)
self.setContextMenu(menu)
def _start_lva(self):
_LOGGER.info("Running: systemctl --user start linux-voice-assistant.service")
subprocess.run(
["systemctl", "--user", "start", "linux-voice-assistant.service"],
check=False,
)
def _stop_lva(self):
_LOGGER.info("Running: systemctl --user stop linux-voice-assistant.service")
subprocess.run(
["systemctl", "--user", "stop", "linux-voice-assistant.service"],
check=False,
)
def _restart_lva(self):
_LOGGER.info("Running: systemctl --user restart linux-voice-assistant.service")
subprocess.run(
["systemctl", "--user", "restart", "linux-voice-assistant.service"],
check=False,
)
def _toggle_mute(self, checked: bool):
self._mute_action.setChecked(checked)
self._muted = checked
# Publish to LVA
topic = f"{self._topic_prefix}/mute/set"
payload = "ON" if checked else "OFF"
_LOGGER.debug("Publishing mute command: %s -> %s", topic, payload)
try:
self._client.publish(topic, payload, retain=False)
except Exception: # noqa: BLE001
_LOGGER.exception("Failed to publish mute command")
# Update local icon
self._update_tray_icon()
def _quit(self):
try:
self._client.loop_stop()
self._client.disconnect()
except Exception: # noqa: BLE001
pass
self._app.quit()
# ------------------------------------------------------------------
# MQTT callbacks
# ------------------------------------------------------------------
def _on_connect(self, client, userdata, flags, rc): # noqa: ARG002
if rc == 0:
_LOGGER.info("Connected to MQTT broker (tray)")
# Subscribe to availability + mute + all LVA topics (for LED states, num_leds, etc.)
client.subscribe(f"{self._topic_prefix}/availability")
client.subscribe(f"{self._topic_prefix}/mute/state")
client.subscribe(f"{self._topic_prefix}/#")
else:
_LOGGER.error("Failed to connect to MQTT, return code %d", rc)
def _on_disconnect(self, client, userdata, rc): # noqa: ARG002
_LOGGER.warning("MQTT disconnected (rc=%s)", rc)
self._available = False
self._update_tray_icon()
def _on_message(self, client, userdata, msg): # noqa: ARG002
try:
topic = msg.topic
payload = msg.payload.decode()
_LOGGER.debug(
"MQTT message: topic=%s payload=%s", topic, payload
)
# availability
if topic == f"{self._topic_prefix}/availability":
self._handle_availability(payload)
return
# mute state
if topic == f"{self._topic_prefix}/mute/state":
self._handle_mute_state(payload)
return
# effect state (we don't currently use it for color, but log it)
if topic.endswith("_effect/state"):
return
# light state
if topic.endswith("_light/state"):
self._handle_light_state(topic, payload)
return
except Exception: # noqa: BLE001
_LOGGER.exception("Error in _on_message")
# ------------------------------------------------------------------
# MQTT handlers
# ------------------------------------------------------------------
def _handle_availability(self, payload: str):
online = payload.strip().lower() == "online"
_LOGGER.debug("Availability update: %s", payload)
self._available = online
self._update_tray_icon()
def _handle_mute_state(self, payload: str):
new_muted = payload.strip().upper() == "ON"
_LOGGER.debug("Mute update: %s", new_muted)
self._muted = new_muted
# Reflect in menu
self._mute_action.setChecked(self._muted)
self._update_tray_icon()
def _handle_light_state(self, topic: str, payload: str):
"""
Handle JSON from .../<state>_light/state
Example payload:
{"state": "ON", "brightness": 127,
"color": {"r": 0, "g": 0, "b": 255}}
"""
try:
data = json.loads(payload)
except json.JSONDecodeError:
_LOGGER.warning("Invalid JSON on light state: %s", payload)
return
# Extract state name from topic
# lva/<device>/<state>_light/state
parts = topic.split("/")
if len(parts) < 4:
return
state_part = parts[-2] # "<state>_light"
if not state_part.endswith("_light"):
return
state_name = state_part[: -len("_light")]
if state_name not in self.STATES:
return
_LOGGER.debug(
"Decoded light state: state_name=%s data=%s", state_name, data
)
state_flag = data.get("state", "OFF").upper()
color_dict = data.get("color", {}) or {}
brightness = int(data.get("brightness", 255))
brightness = max(0, min(brightness, 255))
r = int(color_dict.get("r", 0))
g = int(color_dict.get("g", 0))
b = int(color_dict.get("b", 0))
# Apply brightness scaling to the MQTT color
scale = brightness / 255.0 if brightness > 0 else 0.0
r_scaled = max(0, min(255, int(r * scale)))
g_scaled = max(0, min(255, int(g * scale)))
b_scaled = max(0, min(255, int(b * scale)))
qcolor = QtGui.QColor(r_scaled, g_scaled, b_scaled)
self._last_color_by_state[state_name] = qcolor
# State transitions:
# - For non-idle: treat "ON" as "this is the active state"
# - For idle: **always** treat updates as "idle is the current baseline"
if state_name == "idle":
_LOGGER.debug(
"Tray state_update (idle): data=%s (using MQTT color for idle)",
data,
)
self._current_state = "idle"
else:
if state_flag == "ON":
_LOGGER.debug(
"Tray state_update: state_name=%s data=%s",
state_name,
data,
)
if self._current_state != state_name:
_LOGGER.debug(
"Tray state changing: %s -> %s",
self._current_state,
state_name,
)
self._current_state = state_name
else:
# Ignore OFF for non-idle states
_LOGGER.debug(
"Ignoring OFF for non-idle state %s", state_name
)
self._update_tray_icon()
# ------------------------------------------------------------------
# Icon rendering
# ------------------------------------------------------------------
def _set_icon_by_key(self, key: str):
"""
Convenience wrapper to set icon for:
- 'offline'
- any state name in STATES
"""
if key == "offline":
color = QtGui.QColor(128, 128, 128) # grey
tooltip_state = "offline"
else:
color = self._last_color_by_state.get(
key, self._default_colors.get(key, QtGui.QColor(128, 128, 128))
)
tooltip_state = key
# If muted, tint red (but keep some info from base color)
if self._muted and key != "offline":
base = color
color = QtGui.QColor(
min(255, base.red() + 120),
max(0, int(base.green() * 0.4)),
max(0, int(base.blue() * 0.4)),
)
tooltip_state += " (muted)"
icon = self._make_circle_icon(color)
self.setIcon(icon)
# Update tooltip
tip = f"{self._device_name} – {tooltip_state}"
self.setToolTip(tip)
if self.contextMenu():
self._status_action.setText(tip)
_LOGGER.debug(
"Setting tray icon: key=%s available=%s muted=%s", key, self._available, self._muted
)
def _make_circle_icon(self, color: QtGui.QColor) -> QtGui.QIcon:
size = 20
pixmap = QtGui.QPixmap(size, size)
pixmap.fill(QtCore.Qt.transparent)
painter = QtGui.QPainter(pixmap)
painter.setRenderHint(QtGui.QPainter.Antialiasing, True)
pen = QtGui.QPen(QtGui.QColor(0, 0, 0, 160))
pen.setWidth(1)
painter.setPen(pen)
painter.setBrush(QtGui.QBrush(color))
radius = size // 2 - 2
painter.drawEllipse(2, 2, radius * 2, radius * 2)
painter.end()
return QtGui.QIcon(pixmap)
def _update_tray_icon(self):
if not self._available:
self._set_icon_by_key("offline")
else:
self._set_icon_by_key(self._current_state)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main(argv=None):
parser = argparse.ArgumentParser(description="LVA Tray Client")
parser.add_argument(
"--config",
type=str,
help="Path to LVA config.json (defaults to linux_voice_assistant/config.json)",
)
parser.add_argument(
"--debug",
action="store_true",
help="Enable debug logging",
)
args = parser.parse_args(argv)
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
config_path = Path(args.config) if args.config else None
cfg = load_config(config_path)
# Make Qt app
app = QtWidgets.QApplication(sys.argv)
app.setQuitOnLastWindowClosed(False)
tray = LvaTrayClient(app, cfg)
# Start event loop
sys.exit(app.exec_())
if __name__ == "__main__":
main()