diff --git a/linux_voice_assistant/mqtt_controller.py b/linux_voice_assistant/mqtt_controller.py new file mode 100644 index 0000000..388360d --- /dev/null +++ b/linux_voice_assistant/mqtt_controller.py @@ -0,0 +1,167 @@ +import json +import logging +from typing import TYPE_CHECKING, List + +import paho.mqtt.client as mqtt + +from .event_bus import EventHandler, subscribe + +if TYPE_CHECKING: + from .models import ServerState + +_LOGGER = logging.getLogger(__name__) + +class MqttController(EventHandler): + def __init__(self, state: "ServerState", host: str, port: int, username: str, password: str): + super().__init__(state) + self._host = host + self._port = port + self._username = username + self._password = password + + self._device_name = self.state.name + self._device_id = self.state.name.lower().replace(" ", "_") + self._topic_prefix = f"lva/{self._device_id}" + + # A list of all configurable states + self.CONFIGURABLE_STATES: List[str] = ["idle", "listening", "thinking", "responding", "error"] + + # Generate topics for all entities + self.topics = { + "mute": { + "command": f"{self._topic_prefix}/mute/set", + "state": f"{self._topic_prefix}/mute/state", + }, + "num_leds": { + "command": f"{self._topic_prefix}/num_leds/set", + "state": f"{self._topic_prefix}/num_leds/state", + } + } + for state_name in self.CONFIGURABLE_STATES: + self.topics[state_name] = { + "effect_command": f"{self._topic_prefix}/{state_name}_effect/set", + "effect_state": f"{self._topic_prefix}/{state_name}_effect/state", + "light_command": f"{self._topic_prefix}/{state_name}_light/set", + "light_state": f"{self._topic_prefix}/{state_name}_light/state", + } + + self._client = mqtt.Client() + self._client.on_connect = self._on_connect + self._client.on_message = self._on_message + + def start(self): + try: + if self._username: + self._client.username_pw_set(self._username, self._password) + + _LOGGER.debug("Connecting to MQTT broker at %s:%s", self._host, self._port) + self._client.connect(self._host, self._port, 60) + self._client.loop_start() + except Exception: + _LOGGER.exception("Failed to connect to MQTT broker") + + def stop(self): + self._client.publish(f"{self._topic_prefix}/availability", "offline", retain=True) + self._client.loop_stop() + self._client.disconnect() + _LOGGER.debug("Disconnected from MQTT broker") + + def _on_connect(self, client, userdata, flags, rc): + if rc == 0: + _LOGGER.info("Connected to MQTT broker") + # Subscribe to all command and state topics + for entity_type in self.topics.values(): + for topic_type, topic in entity_type.items(): + client.subscribe(topic) + self._publish_discovery_configs() + else: + _LOGGER.error("Failed to connect to MQTT, return code %d", rc) + + def _on_message(self, client, userdata, msg): + payload_str = msg.payload.decode() + _LOGGER.debug("Received MQTT message on topic %s", msg.topic) + + # Mute and Num LEDs + if msg.topic == self.topics["mute"]["command"]: + self.state.event_bus.publish("set_mic_mute", {"state": payload_str.upper() == "ON"}) + elif msg.topic == self.topics["num_leds"]["command"]: + self.state.event_bus.publish("set_num_leds", {"num_leds": int(payload_str)}) + + # State-based controls + for state_name in self.CONFIGURABLE_STATES: + state_topics = self.topics[state_name] + # Handle commands from HA + if msg.topic == state_topics["effect_command"]: + effect_id = payload_str.lower().replace(" ", "_") + self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id}) + elif msg.topic == state_topics["light_command"]: + try: + data = json.loads(payload_str) + if data.get("state", "ON").upper() == "OFF": + self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": "off"}) + else: + self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": "solid"}) + self.state.event_bus.publish(f"set_{state_name}_color", data) + except json.JSONDecodeError: pass + + # Handle retained state messages on startup + elif msg.topic == state_topics["effect_state"]: + effect_id = payload_str.lower().replace(" ", "_") + self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id, "retained": True}) + elif msg.topic == state_topics["light_state"]: + try: + data = json.loads(payload_str) + data["retained"] = True + self.state.event_bus.publish(f"set_{state_name}_color", data) + except json.JSONDecodeError: pass + + + def _publish_discovery_configs(self): + availability_topic = f"{self._topic_prefix}/availability" + device_info = { "identifiers": [self._device_id], "name": self._device_name, "manufacturer": "LVA Project" } + options = ["Off", "Solid", "Slow Pulse", "Medium Pulse", "Fast Pulse", "Slow Blink", "Medium Blink", "Fast Blink", "Spin"] + + # Mute Switch + mute_cfg = { "name": "Mute Microphone", "unique_id": f"{self._device_id}_mute", "command_topic": self.topics["mute"]["command"], "state_topic": self.topics["mute"]["state"], "availability_topic": availability_topic, "icon": "mdi:microphone-off", "device": device_info } + self._client.publish(f"homeassistant/switch/{self._device_id}_mute/config", json.dumps(mute_cfg), retain=True) + + # Number of LEDs + num_leds_cfg = { "name": "Number of LEDs", "unique_id": f"{self._device_id}_num_leds", "command_topic": self.topics["num_leds"]["command"], "state_topic": self.topics["num_leds"]["state"], "availability_topic": availability_topic, "min": 1, "max": 256, "step": 1, "icon": "mdi:counter", "device": device_info, "mode": "box", "entity_category": "config" } + self._client.publish(f"homeassistant/number/{self._device_id}_num_leds/config", json.dumps(num_leds_cfg), retain=True) + + # Loop through states to create select and light entities + for state_name in self.CONFIGURABLE_STATES: + capital_name = state_name.title() + + # Select Entity + select_cfg = { "name": f"{capital_name} Effect", "unique_id": f"{self._device_id}_{state_name}_effect", "command_topic": self.topics[state_name]["effect_command"], "state_topic": self.topics[state_name]["effect_state"], "availability_topic": availability_topic, "options": options, "icon": "mdi:palette-swatch-variant", "device": device_info, "entity_category": "config" } + self._client.publish(f"homeassistant/select/{self._device_id}_{state_name}_effect/config", json.dumps(select_cfg), retain=True) + + # Light Entity + light_cfg = { "name": f"{capital_name} Color", "unique_id": f"{self._device_id}_{state_name}_color", "schema": "json", "command_topic": self.topics[state_name]["light_command"], "state_topic": self.topics[state_name]["light_state"], "availability_topic": availability_topic, "brightness": True, "color_mode": True, "supported_color_modes": ["rgb"], "device": device_info, "entity_category": "config" } + self._client.publish(f"homeassistant/light/{self._device_id}_{state_name}_color/config", json.dumps(light_cfg), retain=True) + + _LOGGER.debug("Published all MQTT discovery configs") + self._client.publish(availability_topic, "online", retain=True) + self.publish_mute_state(self.state.mic_muted) + + def publish_mute_state(self, is_muted: bool): + self._client.publish(self.topics["mute"]["state"], "ON" if is_muted else "OFF", retain=True) + + @subscribe + def publish_state_to_mqtt(self, data: dict): + state_name = data.get("state_name") + if state_name in self.topics: + state_topics = self.topics[state_name] + + # Update Select State + effect_name = data.get("effect", "off").replace("_", " ").title() + self._client.publish(state_topics["effect_state"], effect_name, retain=True) + + # Update Light State + light_state = { + "state": "ON" if data.get("effect") != "off" else "OFF", + "brightness": int(data.get("brightness", 0.5) * 255), + "color": { "r": data.get("color")[0], "g": data.get("color")[1], "b": data.get("color")[2] } + } + self._client.publish(state_topics["light_state"], json.dumps(light_state), retain=True)