Add Tesla view, MQTT camera alerts, Spanish translations

This commit is contained in:
Joan
2025-12-31 13:29:41 +01:00
parent f1dd6d5c49
commit 21b82129f2
6 changed files with 222 additions and 85 deletions

View File

@@ -14,3 +14,14 @@ NETDATA_URL=http://your-netdata-host:19999
# LED Matrix Configuration # LED Matrix Configuration
LED_ROWS=64 LED_ROWS=64
LED_COLS=64 LED_COLS=64
# Tesla Configuration
TESLA_BATTERY_ENTITY=sensor.lovelace_battery
TESLA_RANGE_ENTITY=sensor.lovelace_range
TESLA_CHARGING_ENTITY=binary_sensor.lovelace_charging
TESLA_PLUGGED_ENTITY=binary_sensor.lovelace_plugged_in
# MQTT Configuration
MQTT_BROKER=10.2.10.165
MQTT_PORT=1883
MQTT_CAMERA_TOPIC=iSpy/cameras/exterior lliçà/alert

View File

@@ -24,3 +24,14 @@ NETDATA_URL = os.getenv('NETDATA_URL')
# LED Matrix Configuration # LED Matrix Configuration
LED_ROWS = int(os.getenv('LED_ROWS', '64')) LED_ROWS = int(os.getenv('LED_ROWS', '64'))
LED_COLS = int(os.getenv('LED_COLS', '64')) LED_COLS = int(os.getenv('LED_COLS', '64'))
# Tesla Configuration
TESLA_BATTERY_ENTITY = os.getenv('TESLA_BATTERY_ENTITY', 'sensor.lovelace_battery')
TESLA_RANGE_ENTITY = os.getenv('TESLA_RANGE_ENTITY', 'sensor.lovelace_range')
TESLA_CHARGING_ENTITY = os.getenv('TESLA_CHARGING_ENTITY', 'binary_sensor.lovelace_charging')
TESLA_PLUGGED_ENTITY = os.getenv('TESLA_PLUGGED_ENTITY', 'binary_sensor.lovelace_plugged_in')
# MQTT Configuration
MQTT_BROKER = os.getenv('MQTT_BROKER', '10.2.10.165')
MQTT_PORT = int(os.getenv('MQTT_PORT', '1883'))
MQTT_CAMERA_TOPIC = os.getenv('MQTT_CAMERA_TOPIC', 'iSpy/cameras/exterior lliçà/alert')

View File

@@ -5,7 +5,9 @@ import requests
from config import ( from config import (
HA_TOKEN, HASS_URL, HA_TOKEN, HASS_URL,
BRIGHTNESS_ENTITY_ID, WEATHER_ENTITY_ID, BRIGHTNESS_ENTITY_ID, WEATHER_ENTITY_ID,
INTERIOR_TEMP_ENTITY_ID, INTERIOR_HUMIDITY_ENTITY_ID INTERIOR_TEMP_ENTITY_ID, INTERIOR_HUMIDITY_ENTITY_ID,
TESLA_BATTERY_ENTITY, TESLA_RANGE_ENTITY,
TESLA_CHARGING_ENTITY, TESLA_PLUGGED_ENTITY
) )
@@ -61,3 +63,18 @@ def get_brightness():
if brightness: if brightness:
return brightness.get("state") return brightness.get("state")
return None return None
def get_tesla_status():
"""Get Tesla Model 3 status from Home Assistant."""
battery = get_entity_value(TESLA_BATTERY_ENTITY)
range_km = get_entity_value(TESLA_RANGE_ENTITY)
charging = get_entity_value(TESLA_CHARGING_ENTITY)
plugged = get_entity_value(TESLA_PLUGGED_ENTITY)
return {
"battery": int(float(battery.get("state", 0))) if battery else None,
"range": int(float(range_km.get("state", 0))) if range_km else None,
"charging": charging.get("state") == "on" if charging else False,
"plugged": plugged.get("state") == "on" if plugged else False,
}

206
matrix.py
View File

@@ -3,14 +3,15 @@
Matrix64 LED Display with Multi-View Pagination Matrix64 LED Display with Multi-View Pagination
Views: Views:
0: Main (Weather + Clock + Temps) 0: Principal (Clima + Reloj + Temps)
1: HDD Details 1: Discos (HDD Details)
2: Date + Weather (Spanish) 2: Fecha (Date + Weather)
3: Tesla (Model 3 Status)
Buttons (GPIO 0 & 1): Buttons (GPIO 0 & 1):
- Button 1: Next view - Button 1: Siguiente vista
- Button 2: Previous view - Button 2: Vista anterior
- Long press: Toggle auto-cycle - Mantener: Auto-ciclo
""" """
from samplebase import SampleBase from samplebase import SampleBase
from rgbmatrix import graphics from rgbmatrix import graphics
@@ -18,14 +19,16 @@ from gpiozero import Button
import time import time
import datetime import datetime
import locale import locale
import math
from config import LED_ROWS, LED_COLS from config import LED_ROWS, LED_COLS
from weather_icons import draw_weather_icon from weather_icons import draw_weather_icon
from home_assistant import ( from home_assistant import (
get_weather, get_weather_description, get_weather, get_weather_description,
get_interior_weather, get_brightness get_interior_weather, get_brightness, get_tesla_status
) )
from netdata import get_hdd_temps from netdata import get_hdd_temps
from mqtt_listener import MQTTListener
# Set Spanish locale for dates # Set Spanish locale for dates
try: try:
@@ -34,19 +37,20 @@ except:
try: try:
locale.setlocale(locale.LC_TIME, 'es_ES') locale.setlocale(locale.LC_TIME, 'es_ES')
except: except:
pass # Fall back to default pass
# Number of views # Number of views
NUM_VIEWS = 3 NUM_VIEWS = 4
# View names (Spanish) # View names (Spanish)
VIEW_NAMES = { VIEW_NAMES = {
0: "Principal", 0: "Principal",
1: "Discos", 1: "Discos",
2: "Fecha", 2: "Fecha",
3: "Tesla",
} }
# Spanish day names fallback # Spanish translations
SPANISH_DAYS = { SPANISH_DAYS = {
'Monday': 'Lunes', 'Tuesday': 'Martes', 'Wednesday': 'Miércoles', 'Monday': 'Lunes', 'Tuesday': 'Martes', 'Wednesday': 'Miércoles',
'Thursday': 'Jueves', 'Friday': 'Viernes', 'Saturday': 'Sábado', 'Sunday': 'Domingo' 'Thursday': 'Jueves', 'Friday': 'Viernes', 'Saturday': 'Sábado', 'Sunday': 'Domingo'
@@ -58,7 +62,6 @@ SPANISH_MONTHS = {
def format_temp(value): def format_temp(value):
"""Format temperature to one decimal place."""
try: try:
return f"{float(value):.1f}°" return f"{float(value):.1f}°"
except (ValueError, TypeError): except (ValueError, TypeError):
@@ -66,7 +69,6 @@ def format_temp(value):
def format_humidity(value): def format_humidity(value):
"""Format humidity as integer percentage."""
try: try:
return f"{int(float(value))}%" return f"{int(float(value))}%"
except (ValueError, TypeError): except (ValueError, TypeError):
@@ -74,7 +76,6 @@ def format_humidity(value):
def get_temperature_color(temp): def get_temperature_color(temp):
"""Determine color based on temperature value."""
try: try:
temp_value = float(temp) if isinstance(temp, (int, float)) else float(str(temp).replace('°', '').strip()) temp_value = float(temp) if isinstance(temp, (int, float)) else float(str(temp).replace('°', '').strip())
if temp_value < 0: if temp_value < 0:
@@ -92,31 +93,25 @@ def get_temperature_color(temp):
def get_hdd_color(temp): def get_hdd_color(temp):
"""Color for HDD temps: green <32, yellow 32-40, red >40."""
try: try:
temp_value = int(temp) temp_value = int(temp)
if temp_value < 32: if temp_value < 32:
return graphics.Color(0, 255, 80) # Green return graphics.Color(0, 255, 80)
elif temp_value <= 40: elif temp_value <= 40:
return graphics.Color(255, 200, 0) # Yellow return graphics.Color(255, 200, 0)
else: else:
return graphics.Color(255, 50, 50) # Red return graphics.Color(255, 50, 50)
except: except:
return graphics.Color(255, 255, 255) return graphics.Color(255, 255, 255)
def get_spanish_day(now): def get_spanish_day(now):
"""Get day name in Spanish.""" return SPANISH_DAYS.get(now.strftime("%A"), now.strftime("%A"))
day_en = now.strftime("%A")
return SPANISH_DAYS.get(day_en, day_en)
def get_spanish_date(now): def get_spanish_date(now):
"""Get date in Spanish format.""" month = SPANISH_MONTHS.get(now.strftime("%b"), now.strftime("%b"))
day = now.day return f"{now.day} {month}"
month_en = now.strftime("%b")
month = SPANISH_MONTHS.get(month_en, month_en)
return f"{day} {month}"
class Matrix64Display(SampleBase): class Matrix64Display(SampleBase):
@@ -132,59 +127,52 @@ class Matrix64Display(SampleBase):
self.last_update = time.time() self.last_update = time.time()
self.weather_desc = "unknown" self.weather_desc = "unknown"
self.hdd_temps = None self.hdd_temps = None
self.tesla = None
# View state # View state
self.current_view = 0 self.current_view = 0
self.auto_cycle = False self.auto_cycle = False
self.last_view_change = time.time() self.last_view_change = time.time()
self.view_cycle_interval = 5 # seconds self.view_cycle_interval = 5
# Feedback message # Feedback
self.feedback_message = None self.feedback_message = None
self.feedback_until = 0 self.feedback_until = 0
# Animation # Animation
self.animation_frame = 0 self.animation_frame = 0
# Button setup # MQTT listener
self.mqtt = MQTTListener()
self.mqtt.start()
# Buttons
self.button1 = Button(0, pull_up=True, hold_time=1.5) self.button1 = Button(0, pull_up=True, hold_time=1.5)
self.button2 = Button(1, pull_up=True, hold_time=1.5) self.button2 = Button(1, pull_up=True, hold_time=1.5)
# Button callbacks
self.button1.when_pressed = self.on_button1_press self.button1.when_pressed = self.on_button1_press
self.button1.when_held = self.on_button_held self.button1.when_held = self.on_button_held
self.button2.when_pressed = self.on_button2_press self.button2.when_pressed = self.on_button2_press
self.button2.when_held = self.on_button_held self.button2.when_held = self.on_button_held
def show_feedback(self, message, duration=2): def show_feedback(self, message, duration=2):
"""Show a feedback message for a duration."""
self.feedback_message = message self.feedback_message = message
self.feedback_until = time.time() + duration self.feedback_until = time.time() + duration
def on_button1_press(self): def on_button1_press(self):
"""Next view."""
if not self.button1.is_held: if not self.button1.is_held:
self.current_view = (self.current_view + 1) % NUM_VIEWS self.current_view = (self.current_view + 1) % NUM_VIEWS
self.last_view_change = time.time() self.last_view_change = time.time()
self.show_feedback(VIEW_NAMES[self.current_view], 1.5) self.show_feedback(VIEW_NAMES[self.current_view], 1.5)
print(f"View: {self.current_view}")
def on_button2_press(self): def on_button2_press(self):
"""Previous view."""
if not self.button2.is_held: if not self.button2.is_held:
self.current_view = (self.current_view - 1) % NUM_VIEWS self.current_view = (self.current_view - 1) % NUM_VIEWS
self.last_view_change = time.time() self.last_view_change = time.time()
self.show_feedback(VIEW_NAMES[self.current_view], 1.5) self.show_feedback(VIEW_NAMES[self.current_view], 1.5)
print(f"View: {self.current_view}")
def on_button_held(self): def on_button_held(self):
"""Toggle auto-cycle on long press."""
self.auto_cycle = not self.auto_cycle self.auto_cycle = not self.auto_cycle
if self.auto_cycle: self.show_feedback("AUTO ON" if self.auto_cycle else "AUTO OFF", 2)
self.show_feedback("AUTO ON", 2)
else:
self.show_feedback("AUTO OFF", 2)
print(f"Auto-cycle: {self.auto_cycle}")
def update_data(self): def update_data(self):
try: try:
@@ -198,6 +186,8 @@ class Matrix64Display(SampleBase):
if interior.get("temperature") is not None: if interior.get("temperature") is not None:
self.interior_temperature = float(interior.get("temperature")) self.interior_temperature = float(interior.get("temperature"))
self.interior_humidity = interior.get("humidity") self.interior_humidity = interior.get("humidity")
self.tesla = get_tesla_status()
except Exception as e: except Exception as e:
print(f"Error updating data: {e}") print(f"Error updating data: {e}")
@@ -206,44 +196,56 @@ class Matrix64Display(SampleBase):
brightness = get_brightness() brightness = get_brightness()
if brightness: if brightness:
self.matrix.brightness = int(float(brightness)) / 10 + 1 self.matrix.brightness = int(float(brightness)) / 10 + 1
except Exception as e: except:
self.matrix.brightness = 5 self.matrix.brightness = 5
def update_hdd_temps(self): def update_hdd_temps(self):
self.hdd_temps = get_hdd_temps() self.hdd_temps = get_hdd_temps()
def draw_camera_alert(self, canvas, fonts):
"""Draw flashing camera alert overlay."""
time_font, data_font, small_font = fonts
# Flashing red border
flash = int(time.time() * 4) % 2
if flash:
border_color = graphics.Color(255, 0, 0)
for i in range(2):
graphics.DrawLine(canvas, i, 0, i, 63, border_color)
graphics.DrawLine(canvas, 63-i, 0, 63-i, 63, border_color)
graphics.DrawLine(canvas, 0, i, 63, i, border_color)
graphics.DrawLine(canvas, 0, 63-i, 63, 63-i, border_color)
# Alert text
alert_color = graphics.Color(255, 50, 50)
graphics.DrawText(canvas, data_font, 8, 32, alert_color, "ALERTA")
graphics.DrawText(canvas, small_font, 6, 42, alert_color, "Camara")
def draw_feedback(self, canvas, fonts, colors): def draw_feedback(self, canvas, fonts, colors):
"""Draw feedback message overlay."""
time_font, data_font, small_font = fonts time_font, data_font, small_font = fonts
time_color, humidity_color, hdd_color, label_color, line_color = colors time_color, humidity_color, hdd_color, label_color, line_color = colors
# Dark background
for y in range(24, 44): for y in range(24, 44):
graphics.DrawLine(canvas, 0, y, 63, y, graphics.Color(0, 0, 0)) graphics.DrawLine(canvas, 0, y, 63, y, graphics.Color(0, 0, 0))
# Message
msg_color = graphics.Color(0, 255, 100) msg_color = graphics.Color(0, 255, 100)
msg_len = len(self.feedback_message) * 7 msg_len = len(self.feedback_message) * 7
x = (64 - msg_len) // 2 x = (64 - msg_len) // 2
graphics.DrawText(canvas, time_font, x, 38, msg_color, self.feedback_message) graphics.DrawText(canvas, time_font, x, 38, msg_color, self.feedback_message)
def draw_view_0(self, canvas, fonts, colors): def draw_view_0(self, canvas, fonts, colors):
"""Main view: Weather + Clock + Temps.""" """Vista Principal"""
time_font, data_font, small_font = fonts time_font, data_font, small_font = fonts
time_color, humidity_color, hdd_color, label_color, line_color = colors time_color, humidity_color, hdd_color, label_color, line_color = colors
now = datetime.datetime.now() now = datetime.datetime.now()
time_str = f"{now.hour:02d}:{now.minute:02d}:{now.second:02d}" time_str = f"{now.hour:02d}:{now.minute:02d}:{now.second:02d}"
# Weather Icon (with smooth sine wave animation)
import math
anim_offset = int(math.sin(self.animation_frame * 0.3) * 2) anim_offset = int(math.sin(self.animation_frame * 0.3) * 2)
draw_weather_icon(canvas, anim_offset, 0, self.weather_desc) draw_weather_icon(canvas, anim_offset, 0, self.weather_desc)
# Vertical separator
graphics.DrawLine(canvas, 28, 0, 28, 25, line_color) graphics.DrawLine(canvas, 28, 0, 28, 25, line_color)
# HDD Temps (3 rows of 2)
if self.hdd_temps and len(self.hdd_temps) > 1: if self.hdd_temps and len(self.hdd_temps) > 1:
temps = [str(int(t[0])) for t in self.hdd_temps[1:] if t] temps = [str(int(t[0])) for t in self.hdd_temps[1:] if t]
if len(temps) >= 2: if len(temps) >= 2:
@@ -253,41 +255,36 @@ class Matrix64Display(SampleBase):
if len(temps) > 4: if len(temps) > 4:
graphics.DrawText(canvas, data_font, 34, 24, hdd_color, " ".join(temps[4:6])) graphics.DrawText(canvas, data_font, 34, 24, hdd_color, " ".join(temps[4:6]))
# Horizontal lines around clock
graphics.DrawLine(canvas, 0, 26, 63, 26, line_color) graphics.DrawLine(canvas, 0, 26, 63, 26, line_color)
time_x = (64 - len(time_str) * 7) // 2 time_x = (64 - len(time_str) * 7) // 2
graphics.DrawText(canvas, time_font, time_x, 39, time_color, time_str) graphics.DrawText(canvas, time_font, time_x, 39, time_color, time_str)
graphics.DrawLine(canvas, 0, 42, 63, 42, line_color) graphics.DrawLine(canvas, 0, 42, 63, 42, line_color)
# Outdoor temp/humidity graphics.DrawText(canvas, data_font, 0, 52, label_color, "Ext")
graphics.DrawText(canvas, data_font, 0, 52, label_color, "Out")
if self.temperature is not None: if self.temperature is not None:
temp_color = get_temperature_color(self.temperature) temp_color = get_temperature_color(self.temperature)
graphics.DrawText(canvas, data_font, 20, 52, temp_color, format_temp(self.temperature)) graphics.DrawText(canvas, data_font, 18, 52, temp_color, format_temp(self.temperature))
if self.humidity is not None: if self.humidity is not None:
graphics.DrawText(canvas, data_font, 48, 52, humidity_color, format_humidity(self.humidity)) graphics.DrawText(canvas, data_font, 46, 52, humidity_color, format_humidity(self.humidity))
# Indoor temp/humidity graphics.DrawText(canvas, data_font, 0, 62, label_color, "Int")
graphics.DrawText(canvas, data_font, 0, 62, label_color, "In")
if self.interior_temperature is not None: if self.interior_temperature is not None:
temp_color = get_temperature_color(self.interior_temperature) temp_color = get_temperature_color(self.interior_temperature)
graphics.DrawText(canvas, data_font, 20, 62, temp_color, format_temp(self.interior_temperature)) graphics.DrawText(canvas, data_font, 18, 62, temp_color, format_temp(self.interior_temperature))
if self.interior_humidity is not None: if self.interior_humidity is not None:
graphics.DrawText(canvas, data_font, 48, 62, humidity_color, format_humidity(self.interior_humidity)) graphics.DrawText(canvas, data_font, 46, 62, humidity_color, format_humidity(self.interior_humidity))
def draw_view_1(self, canvas, fonts, colors): def draw_view_1(self, canvas, fonts, colors):
"""HDD Details view - 6 HDDs in 2 columns.""" """Vista Discos"""
time_font, data_font, small_font = fonts time_font, data_font, small_font = fonts
time_color, humidity_color, hdd_color, label_color, line_color = colors time_color, humidity_color, hdd_color, label_color, line_color = colors
# Title
graphics.DrawText(canvas, data_font, 4, 8, time_color, "Discos") graphics.DrawText(canvas, data_font, 4, 8, time_color, "Discos")
graphics.DrawLine(canvas, 0, 10, 63, 10, line_color) graphics.DrawLine(canvas, 0, 10, 63, 10, line_color)
if self.hdd_temps and len(self.hdd_temps) > 1: if self.hdd_temps and len(self.hdd_temps) > 1:
temps = [(i+1, int(t[0])) for i, t in enumerate(self.hdd_temps[1:7]) if t] temps = [(i+1, int(t[0])) for i, t in enumerate(self.hdd_temps[1:7]) if t]
# Two columns: left (1-3), right (4-6) with bigger font
y = 20 y = 20
for idx, (num, temp_val) in enumerate(temps[:3]): for idx, (num, temp_val) in enumerate(temps[:3]):
temp_color = get_hdd_color(temp_val) temp_color = get_hdd_color(temp_val)
@@ -302,45 +299,87 @@ class Matrix64Display(SampleBase):
graphics.DrawText(canvas, data_font, 48, y, temp_color, f"{temp_val}C") graphics.DrawText(canvas, data_font, 48, y, temp_color, f"{temp_val}C")
y += 10 y += 10
# Auto-cycle indicator
if self.auto_cycle: if self.auto_cycle:
graphics.DrawText(canvas, small_font, 58, 62, label_color, "A") graphics.DrawText(canvas, small_font, 58, 62, label_color, "A")
def draw_view_2(self, canvas, fonts, colors): def draw_view_2(self, canvas, fonts, colors):
"""Date + Weather view (Spanish).""" """Vista Fecha"""
time_font, data_font, small_font = fonts time_font, data_font, small_font = fonts
time_color, humidity_color, hdd_color, label_color, line_color = colors time_color, humidity_color, hdd_color, label_color, line_color = colors
now = datetime.datetime.now() now = datetime.datetime.now()
# Day name in Spanish # Day name centered
day_str = get_spanish_day(now) day_str = get_spanish_day(now)
graphics.DrawText(canvas, data_font, 4, 10, label_color, day_str) day_len = len(day_str) * 5
day_x = (64 - day_len) // 2
graphics.DrawText(canvas, data_font, day_x, 12, label_color, day_str)
# Date in Spanish # Date centered
date_str = get_spanish_date(now) date_str = get_spanish_date(now)
graphics.DrawText(canvas, time_font, 4, 24, time_color, date_str) date_len = len(date_str) * 7
date_x = (64 - date_len) // 2
graphics.DrawText(canvas, time_font, date_x, 26, time_color, date_str)
graphics.DrawLine(canvas, 0, 28, 63, 28, line_color) graphics.DrawLine(canvas, 0, 30, 63, 30, line_color)
# Weather icon with smooth animation
import math
anim_offset = int(math.sin(self.animation_frame * 0.3) * 2) anim_offset = int(math.sin(self.animation_frame * 0.3) * 2)
draw_weather_icon(canvas, 4 + anim_offset, 32, self.weather_desc) draw_weather_icon(canvas, 4 + anim_offset, 34, self.weather_desc)
# Temperature
if self.temperature is not None: if self.temperature is not None:
temp_color = get_temperature_color(self.temperature) temp_color = get_temperature_color(self.temperature)
graphics.DrawText(canvas, time_font, 32, 50, temp_color, f"{self.temperature:.0f}C") graphics.DrawText(canvas, time_font, 32, 52, temp_color, f"{self.temperature:.0f}C")
if self.auto_cycle:
graphics.DrawText(canvas, small_font, 58, 62, label_color, "A")
def draw_view_3(self, canvas, fonts, colors):
"""Vista Tesla"""
time_font, data_font, small_font = fonts
time_color, humidity_color, hdd_color, label_color, line_color = colors
# Title
graphics.DrawText(canvas, data_font, 14, 8, time_color, "Tesla")
graphics.DrawLine(canvas, 0, 10, 63, 10, line_color)
if self.tesla:
# Battery
battery = self.tesla.get("battery")
if battery is not None:
battery_color = graphics.Color(0, 255, 80) if battery > 20 else graphics.Color(255, 50, 50)
graphics.DrawText(canvas, data_font, 0, 22, label_color, "Bat:")
graphics.DrawText(canvas, time_font, 24, 24, battery_color, f"{battery}%")
# Range
range_km = self.tesla.get("range")
if range_km is not None:
graphics.DrawText(canvas, data_font, 0, 36, label_color, "Km:")
graphics.DrawText(canvas, time_font, 20, 38, time_color, f"{range_km}")
# Status
charging = self.tesla.get("charging", False)
plugged = self.tesla.get("plugged", False)
if charging:
status = "Cargando"
status_color = graphics.Color(0, 255, 100)
elif plugged:
status = "Conectado"
status_color = graphics.Color(255, 200, 0)
else:
status = "Listo"
status_color = graphics.Color(100, 100, 100)
graphics.DrawText(canvas, data_font, 4, 54, status_color, status)
else:
graphics.DrawText(canvas, data_font, 4, 32, label_color, "Sin datos")
# Auto-cycle indicator
if self.auto_cycle: if self.auto_cycle:
graphics.DrawText(canvas, small_font, 58, 62, label_color, "A") graphics.DrawText(canvas, small_font, 58, 62, label_color, "A")
def run(self): def run(self):
canvas = self.matrix.CreateFrameCanvas() canvas = self.matrix.CreateFrameCanvas()
# Fonts
font_dir = "/home/pi/rpi-rgb-led-matrix/fonts" font_dir = "/home/pi/rpi-rgb-led-matrix/fonts"
time_font = graphics.Font() time_font = graphics.Font()
time_font.LoadFont(f"{font_dir}/7x13.bdf") time_font.LoadFont(f"{font_dir}/7x13.bdf")
@@ -351,7 +390,6 @@ class Matrix64Display(SampleBase):
fonts = (time_font, data_font, small_font) fonts = (time_font, data_font, small_font)
# Colors
time_color = graphics.Color(30, 90, 220) time_color = graphics.Color(30, 90, 220)
humidity_color = graphics.Color(80, 160, 255) humidity_color = graphics.Color(80, 160, 255)
hdd_color = graphics.Color(140, 120, 100) hdd_color = graphics.Color(140, 120, 100)
@@ -360,7 +398,6 @@ class Matrix64Display(SampleBase):
colors = (time_color, humidity_color, hdd_color, label_color, line_color) colors = (time_color, humidity_color, hdd_color, label_color, line_color)
# Initial fetch
self.update_brightness() self.update_brightness()
self.update_data() self.update_data()
self.update_hdd_temps() self.update_hdd_temps()
@@ -368,33 +405,34 @@ class Matrix64Display(SampleBase):
while True: while True:
current_time = time.time() current_time = time.time()
# Update data every 60 seconds
if current_time - self.last_update >= 60: if current_time - self.last_update >= 60:
self.update_data() self.update_data()
self.update_brightness() self.update_brightness()
self.update_hdd_temps() self.update_hdd_temps()
self.last_update = current_time self.last_update = current_time
# Auto-cycle views
if self.auto_cycle and (current_time - self.last_view_change >= self.view_cycle_interval): if self.auto_cycle and (current_time - self.last_view_change >= self.view_cycle_interval):
self.current_view = (self.current_view + 1) % NUM_VIEWS self.current_view = (self.current_view + 1) % NUM_VIEWS
self.last_view_change = current_time self.last_view_change = current_time
# Animation frame
self.animation_frame += 1 self.animation_frame += 1
canvas.Clear() canvas.Clear()
# Draw current view
if self.current_view == 0: if self.current_view == 0:
self.draw_view_0(canvas, fonts, colors) self.draw_view_0(canvas, fonts, colors)
elif self.current_view == 1: elif self.current_view == 1:
self.draw_view_1(canvas, fonts, colors) self.draw_view_1(canvas, fonts, colors)
elif self.current_view == 2: elif self.current_view == 2:
self.draw_view_2(canvas, fonts, colors) self.draw_view_2(canvas, fonts, colors)
elif self.current_view == 3:
self.draw_view_3(canvas, fonts, colors)
# Draw feedback message if active # Camera alert overlay (highest priority)
if self.feedback_message and current_time < self.feedback_until: if self.mqtt.is_alert_active():
self.draw_camera_alert(canvas, fonts)
# Feedback overlay
elif self.feedback_message and current_time < self.feedback_until:
self.draw_feedback(canvas, fonts, colors) self.draw_feedback(canvas, fonts, colors)
else: else:
self.feedback_message = None self.feedback_message = None

59
mqtt_listener.py Normal file
View File

@@ -0,0 +1,59 @@
"""
MQTT listener for camera alerts.
Runs in a background thread and sets alert flag when message received.
"""
import threading
import time
import paho.mqtt.client as mqtt
from config import MQTT_BROKER, MQTT_PORT, MQTT_CAMERA_TOPIC
class MQTTListener:
def __init__(self):
self.alert_active = False
self.alert_until = 0
self.alert_duration = 5 # seconds
self.client = None
self.connected = False
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print(f"MQTT connected to {MQTT_BROKER}:{MQTT_PORT}")
client.subscribe(MQTT_CAMERA_TOPIC)
print(f"Subscribed to: {MQTT_CAMERA_TOPIC}")
self.connected = True
else:
print(f"MQTT connection failed: {rc}")
def on_message(self, client, userdata, msg):
print(f"MQTT Alert received: {msg.topic}")
self.alert_active = True
self.alert_until = time.time() + self.alert_duration
def on_disconnect(self, client, userdata, rc):
print("MQTT disconnected")
self.connected = False
def is_alert_active(self):
"""Check if alert is currently active."""
if self.alert_active and time.time() > self.alert_until:
self.alert_active = False
return self.alert_active
def start(self):
"""Start MQTT listener in background thread."""
def run():
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.on_disconnect = self.on_disconnect
try:
self.client.connect(MQTT_BROKER, MQTT_PORT, 60)
self.client.loop_forever()
except Exception as e:
print(f"MQTT error: {e}")
thread = threading.Thread(target=run, daemon=True)
thread.start()
print("MQTT listener started")

View File

@@ -1,3 +1,4 @@
python-dotenv>=1.0.0 python-dotenv>=1.0.0
requests>=2.28.0 requests>=2.28.0
gpiozero>=1.6.0 gpiozero>=1.6.0
paho-mqtt>=1.6.0