Update requirements and add MQTT manager

This commit is contained in:
ChuckBuilds
2025-04-12 11:59:07 -05:00
parent 5b47ad15ae
commit 6b466b41e5
4 changed files with 268 additions and 1 deletions

68
config.example.json Normal file
View File

@@ -0,0 +1,68 @@
{
"matrix": {
"width": 64,
"height": 32,
"brightness": 100
},
"display": {
"enabled": true,
"display_durations": {
"clock": 15,
"weather": 15,
"stocks": 45,
"calendar": 30,
"stock_news": 30
}
},
"clock": {
"enabled": true,
"format": "%I:%M:%S %p"
},
"weather": {
"enabled": true,
"api_key": "your_openweather_api_key",
"update_interval": 300,
"units": "imperial"
},
"location": {
"lat": 0.0,
"lon": 0.0,
"city": "Your City"
},
"stocks": {
"enabled": true,
"symbols": ["AAPL", "GOOGL", "MSFT"],
"update_interval": 300
},
"stock_news": {
"enabled": true,
"symbols": ["AAPL", "GOOGL", "MSFT"],
"update_interval": 900,
"max_headlines": 3
},
"calendar": {
"enabled": true,
"credentials_file": "credentials.json",
"token_file": "token.json",
"update_interval": 300,
"max_events": 3
},
"mqtt": {
"broker": "homeassistant.local",
"port": 1883,
"username": "your_mqtt_username",
"password": "your_mqtt_password",
"client_id": "led_matrix",
"use_tls": false,
"topics": [
"homeassistant/sensor/#",
"homeassistant/binary_sensor/#",
"homeassistant/switch/#"
],
"display": {
"scroll_speed": 0.1,
"message_timeout": 60,
"max_messages": 5
}
}
}

1
credentials.json Executable file
View File

@@ -0,0 +1 @@
{"installed":{"client_id":"440006573525-kg1idpl6fj3i17eg6mlucaj10uk4341v.apps.googleusercontent.com","project_id":"homeassistant-309617","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token","auth_provider_x509_cert_url":"https://www.googleapis.com/oauth2/v1/certs","client_secret":"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"}}

View File

@@ -1,4 +1,3 @@
rpi-rgb-led-matrix==1.0.0
Pillow==10.0.0 Pillow==10.0.0
pytz==2023.3 pytz==2023.3
requests==2.31.0 requests==2.31.0

199
src/mqtt_manager.py Normal file
View File

@@ -0,0 +1,199 @@
import paho.mqtt.client as mqtt
import json
from typing import Dict, Any, Optional, List, Callable
from queue import Queue
import threading
import time
class MQTTManager:
def __init__(self, config: Dict[str, Any], display_manager):
"""Initialize MQTT Manager with configuration."""
self.config = config.get('mqtt', {})
self.display_manager = display_manager
self.client = None
self.connected = False
self.message_queue = Queue()
self.subscriptions = {}
self.last_update = 0
self.current_messages = {}
# MQTT Configuration
self.broker = self.config.get('broker', 'localhost')
self.port = self.config.get('port', 1883)
self.username = self.config.get('username')
self.password = self.config.get('password')
self.client_id = self.config.get('client_id', 'led_matrix')
# Display Configuration
self.scroll_speed = self.config.get('scroll_speed', 0.1)
self.message_timeout = self.config.get('message_timeout', 60)
self.max_messages = self.config.get('max_messages', 5)
# Initialize MQTT client
self._setup_mqtt()
# Start message processing thread
self.running = True
self.process_thread = threading.Thread(target=self._process_messages)
self.process_thread.daemon = True
self.process_thread.start()
def _setup_mqtt(self):
"""Setup MQTT client and callbacks."""
self.client = mqtt.Client(client_id=self.client_id)
# Set up callbacks
self.client.on_connect = self._on_connect
self.client.on_message = self._on_message
self.client.on_disconnect = self._on_disconnect
# Set up authentication if configured
if self.username and self.password:
self.client.username_pw_set(self.username, self.password)
# Set up TLS if configured
if self.config.get('use_tls', False):
self.client.tls_set()
def _on_connect(self, client, userdata, flags, rc):
"""Callback for when the client connects to the broker."""
if rc == 0:
print("Connected to MQTT broker")
self.connected = True
# Subscribe to configured topics
for topic in self.config.get('topics', []):
self.subscribe(topic)
else:
print(f"Failed to connect to MQTT broker with code: {rc}")
def _on_message(self, client, userdata, message):
"""Callback for when a message is received."""
try:
payload = message.payload.decode()
try:
# Try to parse as JSON
payload = json.loads(payload)
except json.JSONDecodeError:
# If not JSON, use as plain text
pass
self.message_queue.put({
'topic': message.topic,
'payload': payload,
'timestamp': time.time()
})
except Exception as e:
print(f"Error processing message: {e}")
def _on_disconnect(self, client, userdata, rc):
"""Callback for when the client disconnects from the broker."""
print("Disconnected from MQTT broker")
self.connected = False
if rc != 0:
print(f"Unexpected disconnection. Attempting to reconnect...")
self._reconnect()
def _reconnect(self):
"""Attempt to reconnect to the MQTT broker."""
while not self.connected and self.running:
try:
self.client.connect(self.broker, self.port)
self.client.loop_start()
break
except Exception as e:
print(f"Reconnection failed: {e}")
time.sleep(5)
def start(self):
"""Start the MQTT client connection."""
try:
self.client.connect(self.broker, self.port)
self.client.loop_start()
except Exception as e:
print(f"Failed to connect to MQTT broker: {e}")
self._reconnect()
def stop(self):
"""Stop the MQTT client and message processing."""
self.running = False
if self.client:
self.client.loop_stop()
self.client.disconnect()
def subscribe(self, topic: str, qos: int = 0):
"""Subscribe to an MQTT topic."""
if self.client:
self.client.subscribe(topic, qos)
print(f"Subscribed to topic: {topic}")
def _process_messages(self):
"""Process messages from the queue and update display."""
while self.running:
try:
# Remove expired messages
current_time = time.time()
self.current_messages = {
topic: msg for topic, msg in self.current_messages.items()
if current_time - msg['timestamp'] < self.message_timeout
}
# Process new messages
while not self.message_queue.empty():
message = self.message_queue.get()
self.current_messages[message['topic']] = message
# Update display if we have messages
if self.current_messages:
self._update_display()
time.sleep(0.1)
except Exception as e:
print(f"Error in message processing: {e}")
def _update_display(self):
"""Update the LED matrix display with current messages."""
try:
# Create a new image for drawing
image = Image.new('RGB', (self.display_manager.matrix.width, self.display_manager.matrix.height))
draw = ImageDraw.Draw(image)
# Sort messages by timestamp (newest first)
messages = sorted(
self.current_messages.values(),
key=lambda x: x['timestamp'],
reverse=True
)[:self.max_messages]
# Calculate display layout
section_height = self.display_manager.matrix.height // min(len(messages), self.max_messages)
# Draw each message
for i, message in enumerate(messages):
y = i * section_height
# Format message based on type
if isinstance(message['payload'], dict):
display_text = self._format_dict_message(message['payload'])
else:
display_text = str(message['payload'])
# Draw message text
draw.text((1, y + 1), display_text,
font=self.display_manager.small_font,
fill=(255, 255, 255))
# Update the display
self.display_manager.image = image
self.display_manager.update_display()
except Exception as e:
print(f"Error updating display: {e}")
def _format_dict_message(self, payload: Dict[str, Any]) -> str:
"""Format a dictionary payload for display."""
# Handle Home Assistant state messages
if 'state' in payload:
return f"{payload.get('state', '')}"
# Handle other dictionary messages
return json.dumps(payload, separators=(',', ':'))