awtrix_display.py

"""MQTT communication with Awtrix display."""

import json
import logging
import time

import paho.mqtt.client as mqtt

from config import AWTRIX_TOPIC, CALENDAR_ICON

log = logging.getLogger(__name__)


class AwtrixDisplay:
    """Manages MQTT connection and display updates for Awtrix."""

    def __init__(self, host: str, port: int, username: str, password: str):
        self.host = host
        self.port = port
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.client.username_pw_set(username, password)
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self._connected = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            log.debug(f"Connected to MQTT broker at {self.host}:{self.port}")
            self._connected = True
        else:
            log.error(f"Failed to connect to MQTT: {reason_code}")

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        log.debug(f"Disconnected from MQTT: {reason_code}")
        self._connected = False

    def connect(self):
        """Connect to MQTT broker."""
        try:
            self.client.connect(self.host, self.port, 60)
            self.client.loop_start()
            # Wait for connection
            time.sleep(1)
        except Exception as e:
            log.error(f"MQTT connection error: {e}")

    def disconnect(self):
        """Disconnect from MQTT broker."""
        self.clear_display()
        self.client.loop_stop()
        self.client.disconnect()

    def show_countdown(self, minutes: int, seconds: int):
        """Display countdown on Awtrix."""
        # Format: "MM:SS"
        text = f"{minutes:02d}:{seconds:02d}"

        payload = {
            "text": text,
            "icon": CALENDAR_ICON,
            "pushIcon": 0,
            "rainbow": False,
        }

        # Use custom app
        topic = f"{AWTRIX_TOPIC}/custom/calendar"
        self.client.publish(topic, json.dumps(payload))

        # Switch to the calendar app to make it visible
        switch_topic = f"{AWTRIX_TOPIC}/switch"
        self.client.publish(switch_topic, json.dumps({"name": "calendar"}))

    def clear_display(self):
        """Clear the calendar display from Awtrix."""
        topic = f"{AWTRIX_TOPIC}/custom/calendar"
        self.client.publish(topic, "")

    def beep(self):
        """Play a beep sound on Awtrix."""
        topic = f"{AWTRIX_TOPIC}/rtttl"
        # Short beep tone
        self.client.publish(topic, "beep:d=4,o=5,b=200:c6")