diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..452d555 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,21 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + fonts-dejavu-core \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +COPY run.sh /app/run.sh +COPY server.py /app/server.py + +RUN chmod +x /app/run.sh + +CMD ["/app/run.sh"] diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..2150450 --- /dev/null +++ b/config.yaml @@ -0,0 +1,103 @@ +name: X3 Dashboard Custom +version: "0.3.0" +slug: x3_dashboard_custom +description: Custom XTeink X3 dashboard for kitchen with solar/battery/calendar +url: https://github.com/CraigBell/xteink-x3-home-assistant-dashboard +arch: + - aarch64 + - amd64 + - armv7 +startup: application +boot: auto +init: false +homeassistant_api: true +ports: + 8787/tcp: 8787 +ports_description: + 8787/tcp: X3 dashboard HTTP feed +webui: "http://[HOST]:[PORT:8787]/" +watchdog: "http://[HOST]:[PORT:8787]/x3/status.json" +options: + title: "Küche Dashboard" + subtitle: "Home Assistant Status" + refresh_seconds: 300 + calendar_blacklist: "Gemeinsame Mittagspause,Gemeinsamer Mensa,Block ZDI Jour Fixe,Terminblock ZDI,Regelmäßiger Austausch SC Würzburg" + message_history_hours: 168 + message_history_limit: 50 + messages_per_page: 5 + entity_ids: + room_temp: "sensor.thermo_zigbee_temperature" + outside_temp: "sensor.shelly_ecowitt_ws90_temperature" + solar_daily: "sensor.solar_wechselrichter_solar_tagesenergie" + battery_soc: "sensor.pylontech_battery_id_512_charge" + power_consumption: "sensor.tibber_pulse_kirchstrasse_6_kumulierter_verbrauch" + local_calendar_sensor: "sensor.x3_local_calendar" + calendar_1: "calendar.personlich" + calendar_2: "calendar.tilman_personlich" + wellbeing_days: "" + steps_primary: "" + steps_total: "" + room_lux: "" + room_presence: "" + desk_presence: "" + window_open: "" + projector: "" + phone_battery: "" + phone_battery_state: "" + network_present: "" + network_rssi: "" + network_last_seen: "" + network_direct_clients: "" + network_direct_repeaters: "" + network_node_one_online: "" + network_node_two_online: "" + latest_bridge: "" + latest_external: "" + latest_message: "" + message_entity_ids: + dashboard_pubsub: "" + bridge: "" + external: "" + secure_external: "" +schema: + title: str + subtitle: str + refresh_seconds: int(30,86400) + calendar_blacklist: str + message_history_hours: int(1,720) + message_history_limit: int(1,500) + messages_per_page: int(1,10) + entity_ids: + room_temp: str? + outside_temp: str? + solar_daily: str? + battery_soc: str? + power_consumption: str? + local_calendar_sensor: str? + calendar_1: str? + calendar_2: str? + wellbeing_days: str? + steps_primary: str? + steps_total: str? + room_lux: str? + room_presence: str? + desk_presence: str? + window_open: str? + projector: str? + phone_battery: str? + phone_battery_state: str? + network_present: str? + network_rssi: str? + network_last_seen: str? + network_direct_clients: str? + network_direct_repeaters: str? + network_node_one_online: str? + network_node_two_online: str? + latest_bridge: str? + latest_external: str? + latest_message: str? + message_entity_ids: + dashboard_pubsub: str? + bridge: str? + external: "" + secure_external: str? \ No newline at end of file diff --git a/repository.yaml b/repository.yaml new file mode 100644 index 0000000..c812f31 --- /dev/null +++ b/repository.yaml @@ -0,0 +1,3 @@ +name: X3 Dashboard Custom +url: https://gitea.sonith.de/tilman_hampl/x3-dashboard +maintainer: Tilman Hampl diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..750ca02 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +Pillow>=10.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..30e9d85 --- /dev/null +++ b/server.py @@ -0,0 +1,1242 @@ +#!/usr/bin/env python3 +"""Local XTeink X3 status feed. + +Serves a small set of endpoints that can be consumed by current reader firmware +or future custom firmware: + + /x3/status.json + /x3/status.md + /x3/status.svg + /x3/status.png + /x3/status.bmp + /x3/status.epub + /x3/opds +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import html +import io +import json +import os +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import textwrap +from typing import Any +import urllib.error +import urllib.parse +import urllib.request +import zipfile + +try: + from PIL import Image, ImageDraw, ImageFont +except Exception: # pragma: no cover - exercised on machines without Pillow + Image = None + ImageDraw = None + ImageFont = None + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_ENV_FILE = PROJECT_ROOT / ".env" +DEFAULT_OPTIONS_FILE = Path("/data/options.json") +DISPLAY_WIDTH = 528 +DISPLAY_HEIGHT = 792 +DEFAULT_REFRESH_SECONDS = 300 +MESSAGE_HISTORY_HOURS = 168 +MESSAGE_HISTORY_LIMIT = 50 +MESSAGES_PER_PAGE = 5 + +DEFAULT_ENTITY_IDS = { + "wellbeing_days": "", + "steps_primary": "", + "steps_total": "", + "room_temp": "", + "outside_temp": "", + "solar_daily": "", + "battery_soc": "", + "power_consumption": "", + "local_calendar_sensor": "", + "calendar_1": "", + "calendar_2": "", + "room_lux": "", + "room_presence": "", + "desk_presence": "", + "window_open": "", + "projector": "", + "phone_battery": "", + "phone_battery_state": "", + "network_present": "", + "network_rssi": "", + "network_last_seen": "", + "network_direct_clients": "", + "network_direct_repeaters": "", + "network_node_one_online": "", + "network_node_two_online": "", + "latest_bridge": "", + "latest_external": "", + "latest_message": "", +} + +DEFAULT_MESSAGE_ENTITY_IDS = { + "dashboard_pubsub": "", + "bridge": "", + "external": "", + "secure_external": "", +} + +DEFAULT_OPTIONS: dict[str, Any] = { + "title": "X3 Dashboard", + "subtitle": "Home Assistant status", + "network_label": "Network", + "wellbeing_label": "Wellbeing", + "refresh_seconds": DEFAULT_REFRESH_SECONDS, + "message_history_hours": MESSAGE_HISTORY_HOURS, + "message_history_limit": MESSAGE_HISTORY_LIMIT, + "messages_per_page": MESSAGES_PER_PAGE, + "entity_ids": DEFAULT_ENTITY_IDS, + "message_entity_ids": DEFAULT_MESSAGE_ENTITY_IDS, +} + +CONFIG: dict[str, Any] = DEFAULT_OPTIONS.copy() + + +def load_env(path: Path = DEFAULT_ENV_FILE) -> None: + if not path.exists(): + return + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + key = key.strip() + value = value.strip().strip('"').strip("'") + os.environ.setdefault(key, value) + + +def normalise_entity_map(values: dict[str, Any] | None, defaults: dict[str, str]) -> dict[str, str]: + merged = defaults.copy() + for key, raw_value in (values or {}).items(): + target_key = str(key) + if target_key not in merged: + continue + merged[target_key] = str(raw_value or "").strip() + return merged + + +def load_options(path: Path = DEFAULT_OPTIONS_FILE) -> dict[str, Any]: + options = DEFAULT_OPTIONS.copy() + if path.exists(): + loaded = json.loads(path.read_text(encoding="utf-8")) + if isinstance(loaded, dict): + options.update({key: value for key, value in loaded.items() if key not in {"entity_ids", "message_entity_ids"}}) + options["entity_ids"] = normalise_entity_map(loaded.get("entity_ids"), DEFAULT_ENTITY_IDS) + options["message_entity_ids"] = normalise_entity_map( + loaded.get("message_entity_ids"), DEFAULT_MESSAGE_ENTITY_IDS + ) + else: + options["entity_ids"] = DEFAULT_ENTITY_IDS.copy() + options["message_entity_ids"] = DEFAULT_MESSAGE_ENTITY_IDS.copy() + return options + + +def active_entity_ids() -> dict[str, str]: + return { + key: value + for key, value in (CONFIG.get("entity_ids") or {}).items() + if isinstance(value, str) and value.strip() + } + + +def active_message_entity_ids() -> dict[str, str]: + return { + key: value + for key, value in (CONFIG.get("message_entity_ids") or {}).items() + if isinstance(value, str) and value.strip() + } + + +def option_int(key: str, default: int, minimum: int = 1, maximum: int = 9999) -> int: + try: + parsed = int(str(CONFIG.get(key, default)).strip()) + except (TypeError, ValueError): + return default + return max(minimum, min(maximum, parsed)) + + +def now_local() -> dt.datetime: + return dt.datetime.now().astimezone() + + +def parse_ha_time(value: str | None) -> dt.datetime | None: + if not value or value in {"unknown", "unavailable", "none"}: + return None + try: + if value.endswith("Z"): + value = value[:-1] + "+00:00" + return dt.datetime.fromisoformat(value).astimezone() + except ValueError: + return None + + +def human_age(value: str | None, now: dt.datetime | None = None) -> str: + parsed = parse_ha_time(value) + if not parsed: + return "unknown" + now = now or now_local() + seconds = max(0, int((now - parsed).total_seconds())) + if seconds < 60: + return f"{seconds}s ago" + minutes = seconds // 60 + if minutes < 60: + return f"{minutes}m ago" + hours = minutes // 60 + if hours < 48: + return f"{hours}h ago" + return f"{hours // 24}d ago" + + +def clean_state(value: Any, fallback: str = "unknown") -> str: + if value is None: + return fallback + text = str(value).strip() + return text if text else fallback + + +def state_bool(value: str | None) -> bool | None: + if value == "on": + return True + if value == "off": + return False + return None + + +class HomeAssistantClient: + def __init__(self) -> None: + supervisor_token = os.getenv("SUPERVISOR_TOKEN", "") + ha_token = os.getenv("HA_TOKEN", "") + if supervisor_token and not ha_token: + self.base_url = os.getenv("HA_URL", "http://supervisor/core").rstrip("/") + self.token = supervisor_token + else: + self.base_url = os.getenv("HA_URL", "").rstrip("/") + self.token = ha_token + if not self.token: + raise RuntimeError("HA_TOKEN or SUPERVISOR_TOKEN is required.") + if not self.base_url: + raise RuntimeError("HA_URL is required when running outside the Home Assistant add-on.") + + def state(self, entity_id: str) -> dict[str, Any] | None: + url = f"{self.base_url}/api/states/{urllib.parse.quote(entity_id, safe='.')}" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + }, + ) + try: + with urllib.request.urlopen(req, timeout=8) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + if exc.code == 404: + return None + raise + + def states(self, entity_ids: dict[str, str]) -> dict[str, dict[str, Any] | None]: + return {key: self.state(entity_id) for key, entity_id in entity_ids.items() if entity_id} + + def history_for_entity(self, entity_id: str, start: dt.datetime) -> list[list[dict[str, Any]]]: + start_utc = start.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") + query = urllib.parse.urlencode( + { + "filter_entity_id": entity_id, + "significant_changes_only": "0", + } + ) + url = f"{self.base_url}/api/history/period/{urllib.parse.quote(start_utc, safe='')}?{query}" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(req, timeout=12) as resp: + data = json.loads(resp.read().decode("utf-8")) + return [series for series in data if isinstance(series, list)] if isinstance(data, list) else [] + + def history(self, entity_ids: list[str], start: dt.datetime) -> list[list[dict[str, Any]]]: + all_series: list[list[dict[str, Any]]] = [] + if not entity_ids: + return all_series + fallback_start = max(start, now_local() - dt.timedelta(hours=6)) + for entity_id in entity_ids: + series = self.history_for_entity(entity_id, start) + if not series and fallback_start > start: + series = self.history_for_entity(entity_id, fallback_start) + all_series.extend(series) + return all_series + + +def st(states: dict[str, dict[str, Any] | None], key: str, fallback: str = "unknown") -> str: + return clean_state((states.get(key) or {}).get("state"), fallback) + + +def attr(states: dict[str, dict[str, Any] | None], key: str, name: str, fallback: Any = None) -> Any: + return ((states.get(key) or {}).get("attributes") or {}).get(name, fallback) + + +def parse_bridge_signature(value: str) -> dict[str, str]: + """Parse HA bridge signatures like '1783779161|g8py<->#g8py||message'.""" + if value in {"unknown", "unavailable", "", "none"}: + return {"time": "unknown", "route": "", "message": "No recent message"} + + parts = value.split("|", 2) + timestamp = parts[0] if parts else "" + route = parts[1] if len(parts) > 2 else "" + message = parts[2] if len(parts) > 2 else parts[1] if len(parts) > 1 else value + message = message.lstrip("|").strip() + + time_text = "unknown" + try: + ts = int(timestamp) + time_text = dt.datetime.fromtimestamp(ts).astimezone().strftime("%H:%M") + except ValueError: + pass + + return { + "time": time_text, + "route": route.strip(), + "message": message or "No recent message", + } + + +def epoch_to_local(timestamp: str | int | None) -> dt.datetime | None: + try: + return dt.datetime.fromtimestamp(int(str(timestamp).strip())).astimezone() + except (TypeError, ValueError, OSError): + return None + + +def message_time_text(timestamp: str | int | None, fallback: str | None = None) -> str: + parsed = epoch_to_local(timestamp) + if parsed: + return parsed.strftime("%a %H:%M") + fallback_dt = parse_ha_time(fallback) + if fallback_dt: + return fallback_dt.strftime("%a %H:%M") + return "unknown" + + +def parse_x3_message_signature(value: str, fallback_time: str | None = None) -> dict[str, Any] | None: + """Parse 'timestamp|source|title|message' values from x3_message_feed_last.""" + if value in {"unknown", "unavailable", "", "none"}: + return None + parts = value.split("|", 3) + if len(parts) < 4: + return None + + timestamp, source, title, message = parts + parsed_time = epoch_to_local(timestamp) or parse_ha_time(fallback_time) + title = title.strip() + message = message.strip() + if not message and not title: + return None + return { + "timestamp": int(parsed_time.timestamp()) if parsed_time else 0, + "time": message_time_text(timestamp, fallback_time), + "source": source.strip() or "HA", + "title": title, + "message": message or title, + "kind": "pubsub", + } + + +def parse_bridge_message_signature(source: str, value: str, fallback_time: str | None = None) -> dict[str, Any] | None: + if value in {"unknown", "unavailable", "", "none"}: + return None + parts = value.split("|", 2) + timestamp = parts[0] if parts else "" + parsed_time = epoch_to_local(timestamp) or parse_ha_time(fallback_time) + bridge = parse_bridge_signature(value) + message = bridge["message"].strip() + if not message or message == "No recent message": + return None + + title = "Message" + if source == "bridge" and bridge["route"]: + title = bridge["route"].replace("<->", " to ") + elif source in {"external", "secure_external"}: + title = "External message" + + return { + "timestamp": int(parsed_time.timestamp()) if parsed_time else 0, + "time": message_time_text(timestamp, fallback_time), + "source": title, + "title": title, + "message": message, + "kind": "mesh", + } + + +def parse_message_state(source: str, state: str, fallback_time: str | None = None) -> dict[str, Any] | None: + if source == "dashboard_pubsub": + return parse_x3_message_signature(state, fallback_time) + return parse_bridge_message_signature(source, state, fallback_time) + + +def compact_message_key(message: dict[str, Any]) -> tuple[int, str]: + text = str(message.get("message", "")) + return int(message.get("timestamp") or 0), " ".join(text.lower().split()) + + +def dedupe_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + seen: set[tuple[int, str]] = set() + unique: list[dict[str, Any]] = [] + for message in sorted(messages, key=lambda item: int(item.get("timestamp") or 0), reverse=True): + key = compact_message_key(message) + if key in seen: + continue + seen.add(key) + unique.append(message) + return unique + + +def positive_int(value: str | None, default: int, minimum: int = 1, maximum: int = 999) -> int: + try: + parsed = int(str(value or "").strip()) + except ValueError: + return default + return max(minimum, min(maximum, parsed)) + + +def query_positive_int(values: list[str] | None, default: int, minimum: int = 1, maximum: int = 999) -> int: + for value in reversed(values or []): + try: + parsed = int(str(value or "").strip()) + except ValueError: + continue + return max(minimum, min(maximum, parsed)) + return default + + +def page_count(total_items: int, per_page: int) -> int: + return max(1, (total_items + per_page - 1) // per_page) + + +def choose_steps(states: dict[str, dict[str, Any] | None]) -> str: + primary = st(states, "steps_primary", "") + if primary and primary not in {"unknown", "unavailable"}: + return primary + total = st(states, "steps_total", "") + return total if total else "unknown" + + +def build_calendar_events(states: dict[str, dict[str, Any] | None], now: dt.datetime, days_ahead: int = 14) -> list[dict[str, Any]]: + """Combine events from CalDAV calendars (HA service) and local sensor.""" + import urllib.parse + + all_events: list[dict[str, Any]] = [] + + caldav_keys = ["calendar_1", "calendar_2"] + caldav_entity_ids = [st(states, k, "") for k in caldav_keys if st(states, k, "")] + local_sensor_id = st(states, "local_calendar_sensor", "") + + blacklist_str = str(CONFIG.get("calendar_blacklist") or "") + blacklist_keywords = [kw.strip().lower() for kw in blacklist_str.split(",") if kw.strip()] + + client = None + + end_date = now + dt.timedelta(days=days_ahead) + start_iso = now.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") + end_iso = end_date.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") + + for cal_id in caldav_entity_ids: + try: + if client is None: + client = HomeAssistantClient() + payload = { + "entity_id": cal_id, + "start_date_time": start_iso, + "end_date_time": end_iso, + } + url = f"{client.base_url}/api/services/calendar/get_events?return_response" + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={ + "Authorization": f"Bearer {client.token}", + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(req, timeout=12) as resp: + data = json.loads(resp.read().decode("utf-8")) + events = data.get("service_response", {}).get(cal_id, {}).get("events", []) + for ev in events: + all_events.append({ + "summary": ev.get("summary", "?"), + "start": ev.get("start", "?"), + "end": ev.get("end", "?"), + "source": "caldav", + }) + except Exception: + pass + + if local_sensor_id: + sensor_state = states.get("local_calendar_sensor") + if sensor_state and sensor_state.get("attributes", {}).get("events"): + for ev in sensor_state["attributes"]["events"]: + all_events.append({ + "summary": ev.get("summary", "?"), + "start": ev.get("start", "?"), + "end": ev.get("end", "?"), + "source": "local", + }) + + def sort_key(ev: dict[str, Any]) -> str: + s = str(ev.get("start", "")) + try: + return dt.datetime.fromisoformat(s).isoformat() + except (ValueError, TypeError): + return "9999" + + all_events.sort(key=sort_key) + + # Blacklist: Termine mit bestimmten Stichworten ausblenden + if blacklist_keywords: + all_events = [ev for ev in all_events if not any(kw in str(ev.get("summary", "")).lower() for kw in blacklist_keywords)] + + # Dedupliziere Termine mit gleichem summary + start + seen = set() + unique_events = [] + for ev in all_events: + key = (str(ev.get("summary", "")).strip().lower(), str(ev.get("start", ""))[:10]) + if key not in seen: + seen.add(key) + unique_events.append(ev) + + return unique_events[:15] + + +def build_rain_forecast(now: dt.datetime) -> str: + """Check weather forecast for rain in next 24h, return 'vorauss. Regen um HH:MM' or 'kein Regen erwartet'.""" + try: + client = HomeAssistantClient() + payload = {"entity_id": "weather.forecast_home", "type": "hourly"} + url = f"{client.base_url}/api/services/weather/get_forecasts?return_response" + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={ + "Authorization": f"Bearer {client.token}", + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(req, timeout=12) as resp: + data = json.loads(resp.read().decode("utf-8")) + forecast = data.get("service_response", {}).get("weather.forecast_home", {}).get("forecast", []) + for f in forecast[:24]: + cond = str(f.get("condition", "")) + precip = f.get("precipitation", 0) or 0 + if cond in ("rainy", "pouring", "snowy", "snowy-rainy") or precip > 0.1: + dt_str = f.get("datetime", "") + try: + dt_obj = dt.datetime.fromisoformat(dt_str) + return f"vorauss. Regen um {dt_obj.strftime('%H:%M')} Uhr" + except (ValueError, TypeError): + return "vorauss. Regen" + return "kein Regen (24h)" + except Exception: + return "Wetter unbekannt" + + +def build_feed() -> dict[str, Any]: + now = now_local() + client = HomeAssistantClient() + entity_ids = active_entity_ids() + states = client.states(entity_ids) + network_label = str(CONFIG.get("network_label") or "Network") + wellbeing_label = str(CONFIG.get("wellbeing_label") or "Wellbeing") + + latest_bridge = parse_bridge_signature(st(states, "latest_bridge", "")) + latest_external = parse_bridge_signature(st(states, "latest_external", "")) + latest_x3_message = parse_message_state( + "dashboard_pubsub", + st(states, "latest_message", ""), + (states.get("latest_message") or {}).get("last_changed"), + ) + + room_presence = state_bool(st(states, "room_presence", "unknown")) + projector = state_bool(st(states, "projector", "unknown")) + window_open = state_bool(st(states, "window_open", "unknown")) + network_present = state_bool(st(states, "network_present", "unknown")) + node_one_online = state_bool(st(states, "network_node_one_online", "unknown")) + node_two_online = state_bool(st(states, "network_node_two_online", "unknown")) + + office_status = "occupied" if room_presence else "empty" if room_presence is False else "unknown" + projector_status = "projector on" if projector else "projector off" if projector is False else "projector unknown" + window_status = "window open" if window_open else "window closed" if window_open is False else "window unknown" + mesh_status = "online" if network_present else "offline" if network_present is False else "unknown" + + room_temp_val = st(states, "room_temp", "unknown") + outside_temp_val = st(states, "outside_temp", "unknown") + solar_daily_val = st(states, "solar_daily", "unknown") + battery_soc_val = st(states, "battery_soc", "unknown") + power_consumption_val = st(states, "power_consumption", "unknown") + + rain_forecast = build_rain_forecast(now) + + headline_parts = [] + if network_present: + headline_parts.append(f"{network_label} OK") + if st(states, "wellbeing_days", "unknown") not in {"unknown", "unavailable"}: + headline_parts.append(f"{st(states, 'wellbeing_days')} days") + headline = " · ".join(headline_parts) or str(CONFIG.get("title") or "X3 status") + + payload = { + "generated_at": now.isoformat(timespec="seconds"), + "display": {"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT, "orientation": "portrait"}, + "headline": headline, + "summary": [ + {"label": wellbeing_label, "value": f"{st(states, 'wellbeing_days')} days"}, + {"label": "Steps", "value": choose_steps(states)}, + {"label": "Room", "value": f"{office_status}, {room_temp_val}C"}, + {"label": network_label, "value": f"{mesh_status}, RSSI {st(states, 'network_rssi')} dBm"}, + ], + "office": { + "presence": office_status, + "desk": st(states, "desk_presence"), + "temperature_c": room_temp_val, + "outside_temp_c": outside_temp_val, + "rain_forecast": rain_forecast, + "solar_daily": solar_daily_val, + "battery_soc": battery_soc_val, + "power_consumption": power_consumption_val, + "lux": st(states, "room_lux"), + "projector": projector_status, + "window": window_status, + }, + "network": { + "label": network_label, + "status": mesh_status, + "rssi_dbm": st(states, "network_rssi"), + "last_seen": st(states, "network_last_seen"), + "last_seen_age": human_age(st(states, "network_last_seen"), now), + "direct_clients": st(states, "network_direct_clients"), + "direct_repeaters": st(states, "network_direct_repeaters"), + "loft_online": bool(node_one_online), + "loft_yagi_online": bool(node_two_online), + "latest_bridge": latest_bridge, + "latest_external": latest_external, + }, + "latest_message": latest_x3_message + or { + "time": latest_bridge["time"], + "source": latest_bridge["route"] or "Message", + "title": latest_bridge["route"] or "Message", + "message": latest_bridge["message"], + "kind": "bridge", + }, + "phone": { + "battery_percent": st(states, "phone_battery"), + "battery_state": st(states, "phone_battery_state"), + }, + "calendar_events": build_calendar_events(states, now), + "raw_entity_ids": entity_ids, + } + return payload + + +def build_message_feed( + hours: int | None = None, + limit: int | None = None, + page: int = 1, + per_page: int | None = None, +) -> dict[str, Any]: + now = now_local() + client = HomeAssistantClient() + message_entity_ids = active_message_entity_ids() + hours = hours or option_int("message_history_hours", MESSAGE_HISTORY_HOURS, minimum=1, maximum=720) + limit = limit or option_int("message_history_limit", MESSAGE_HISTORY_LIMIT, minimum=1, maximum=500) + per_page = per_page or option_int("messages_per_page", MESSAGES_PER_PAGE, minimum=1, maximum=10) + start = now - dt.timedelta(hours=hours) + entity_lookup = {entity_id: key for key, entity_id in message_entity_ids.items()} + messages: list[dict[str, Any]] = [] + + try: + history = client.history(list(message_entity_ids.values()), start) + except Exception: + history = [] + + for series in history: + for row in series: + if not isinstance(row, dict): + continue + source = entity_lookup.get(str(row.get("entity_id", ""))) + if not source: + continue + parsed = parse_message_state(source, clean_state(row.get("state"), ""), row.get("last_changed")) + if parsed: + messages.append(parsed) + + states = client.states(message_entity_ids) + for source, state_data in states.items(): + parsed = parse_message_state(source, clean_state((state_data or {}).get("state"), ""), (state_data or {}).get("last_changed")) + if parsed: + messages.append(parsed) + + all_messages = dedupe_messages(messages)[:limit] + total_messages = len(all_messages) + total_pages = page_count(total_messages, per_page) + page = max(1, min(page, total_pages)) + start_index = (page - 1) * per_page + page_messages = all_messages[start_index : start_index + per_page] + return { + "generated_at": now.isoformat(timespec="seconds"), + "display": {"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT, "orientation": "portrait"}, + "title": "X3 Messages", + "subtitle": str(CONFIG.get("subtitle") or "Home Assistant status"), + "hours": hours, + "pagination": { + "page": page, + "page_count": total_pages, + "per_page": per_page, + "total": total_messages, + "has_previous": page > 1, + "has_next": page < total_pages, + }, + "messages": page_messages, + "raw_entity_ids": message_entity_ids, + } + + +def build_dashboard_manifest(base_url: str = "") -> dict[str, Any]: + message_feed = build_message_feed(page=1) + pagination = message_feed["pagination"] + prefix = base_url.rstrip("/") + refresh_seconds = option_int("refresh_seconds", DEFAULT_REFRESH_SECONDS, minimum=30, maximum=86400) + return { + "version": 1, + "generated_at": now_local().isoformat(timespec="seconds"), + "display": {"width": DISPLAY_WIDTH, "height": DISPLAY_HEIGHT, "orientation": "portrait"}, + "refresh_seconds": refresh_seconds, + "default_page": "status", + "controls": { + "next_page": "right", + "previous_page": "left", + "next_message_page": "down", + "previous_message_page": "up", + "refresh": "confirm", + "home": "back", + }, + "pages": [ + { + "id": "status", + "title": "Status", + "type": "image", + "content_type": "image/bmp", + "url": f"{prefix}/x3/status.bmp", + "refresh_seconds": refresh_seconds, + }, + { + "id": "messages", + "title": "Messages", + "type": "paged_image", + "content_type": "image/bmp", + "url": f"{prefix}/x3/messages.bmp?page={{page}}", + "page": 1, + "page_count": pagination["page_count"], + "refresh_seconds": refresh_seconds, + }, + ], + } + + +def render_markdown(feed: dict[str, Any]) -> str: + mesh = feed["network"] + office = feed["office"] + phone = feed["phone"] + latest = feed.get("latest_message") or mesh["latest_bridge"] + lines = [ + f"# {feed['headline']}", + "", + f"Updated: {feed['generated_at']}", + "", + "## Now", + ] + for item in feed["summary"]: + lines.append(f"- {item['label']}: {item['value']}") + lines.extend( + [ + "", + "## Küche", + f"- Innen: {office['temperature_c']} C", + f"- Aussen: {office['outside_temp_c']} C", + f"- Solar heute: {office['solar_daily']} kWh", + f"- Batterie: {office['battery_soc']}%", + f"- Netz: {office['power_consumption']} kWh", + "", + f"## {mesh.get('label', 'Network')}", + f"- {mesh['status']}, RSSI {mesh['rssi_dbm']} dBm", + f"- Seen {mesh['last_seen_age']}", + f"- Direct: {mesh['direct_clients']} clients, {mesh['direct_repeaters']} repeaters", + f"- Node 1: {'online' if mesh['loft_online'] else 'offline'}", + f"- Node 2: {'online' if mesh['loft_yagi_online'] else 'offline'}", + "", + "## Letzte Meldung", + f"{latest.get('time', 'unknown')} {latest.get('source', '')}", + "", + latest.get("message", "No recent message"), + "", + ] + ) + return "\n".join(lines) + + +def render_messages_markdown(feed: dict[str, Any]) -> str: + pagination = feed.get("pagination", {}) + lines = [ + f"# {feed['title']}", + "", + f"Page {pagination.get('page', 1)} of {pagination.get('page_count', 1)}", + "", + f"Updated: {feed['generated_at']}", + "", + ] + messages = feed.get("messages", []) + if not messages: + lines.append("No recent messages.") + for message in messages: + title = message.get("title") or message.get("source") or "Message" + lines.extend( + [ + f"## {message.get('time', 'unknown')} {title}", + message.get("message", ""), + "", + ] + ) + return "\n".join(lines) + + +def wrap_svg_text(text: str, max_chars: int) -> list[str]: + wrapped: list[str] = [] + for line in text.splitlines(): + if not line: + wrapped.append("") + else: + wrapped.extend(textwrap.wrap(line, width=max_chars, break_long_words=False) or [""]) + return wrapped + + +def render_svg(feed: dict[str, Any]) -> str: + mesh = feed["network"] + office = feed["office"] + latest = feed.get("latest_message") or mesh["latest_bridge"] + rows = [ + (40, feed["headline"], 34, "bold"), + (92, now_local().strftime("%a %H:%M"), 54, "bold"), + (168, f"Küche: Innen {office['temperature_c']}C Aussen {office['outside_temp_c']}C", 28, "bold"), + (210, f"Solar {office['solar_daily']} kWh · Batterie {office['battery_soc']}%", 26, "normal"), + (264, f"Netz {office['power_consumption']} kWh · {mesh.get('label', 'Network')} {mesh['status']}", 28, "bold"), + (330, f"Seen {mesh['last_seen_age']} · {mesh['direct_clients']} clients", 24, "normal"), + (430, "Letzte Meldung", 28, "bold"), + ] + message_lines = wrap_svg_text(f"{latest.get('time', 'unknown')} {latest.get('message', '')}", 24)[:7] + y = 472 + for line in message_lines: + rows.append((y, line, 25, "normal")) + y += 34 + rows.extend( + [ + (758, feed["generated_at"].split("+")[0].replace("T", " "), 18, "normal"), + ] + ) + text_nodes = "\n".join( + f'{html.escape(text)}' + for y, text, size, weight in rows + ) + return f""" + + + + + + +{text_nodes} + + +""" + + +def load_font(size: int, bold: bool = False) -> Any: + assert ImageFont is not None + candidates = [ + "/System/Library/Fonts/Supplemental/Arial Bold.ttf" if bold else "/System/Library/Fonts/Supplemental/Arial.ttf", + "/System/Library/Fonts/Supplemental/Verdana Bold.ttf" if bold else "/System/Library/Fonts/Supplemental/Verdana.ttf", + "/System/Library/Fonts/Helvetica.ttc", + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" if bold else "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + "/usr/share/fonts/truetype/liberation2/LiberationSans-Bold.ttf" if bold else "/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf", + ] + for path in candidates: + if path and Path(path).exists(): + try: + return ImageFont.truetype(path, size=size) + except Exception: + continue + return ImageFont.load_default(size=size) + + +def draw_wrapped(draw: Any, xy: tuple[int, int], text: str, font: Any, width: int, line_gap: int = 8, max_lines: int | None = None) -> int: + x, y = xy + words = text.split() + lines: list[str] = [] + current = "" + for word in words: + trial = f"{current} {word}".strip() + bbox = draw.textbbox((0, 0), trial, font=font) + if bbox[2] - bbox[0] <= width or not current: + current = trial + else: + lines.append(current) + current = word + if current: + lines.append(current) + if max_lines is not None: + lines = lines[:max_lines] + for line in lines: + draw.text((x, y), line, font=font, fill=0) + bbox = draw.textbbox((x, y), line, font=font) + y = bbox[3] + line_gap + return y + + +def render_status_image(feed: dict[str, Any]) -> Any: + if Image is None or ImageDraw is None or ImageFont is None: + raise RuntimeError("Pillow is not installed. Run: python3 -m pip install -r x3-feed/requirements.txt") + + img = Image.new("L", (DISPLAY_WIDTH, DISPLAY_HEIGHT), 255) + draw = ImageDraw.Draw(img) + + title_font = load_font(34, bold=True) + clock_font = load_font(58, bold=True) + section_font = load_font(28, bold=True) + body_font = load_font(25, bold=False) + small_font = load_font(18, bold=False) + + draw.rectangle((16, 16, DISPLAY_WIDTH - 16, DISPLAY_HEIGHT - 16), outline=0, width=3) + draw.text((24, 30), now_local().strftime("%a %H:%M"), font=clock_font, fill=0) + draw.line((24, 100, DISPLAY_WIDTH - 24, 100), fill=0, width=2) + + office = feed["office"] + y = 120 + draw.text((24, y), "Wetter", font=section_font, fill=0) + y += 42 + draw.text((24, y), f"Regen: {office.get('rain_forecast', 'unbekannt')}", font=body_font, fill=0) + y += 32 + draw.text((24, y), f"Innen: {office['temperature_c']} C", font=body_font, fill=0) + y += 32 + draw.text((24, y), f"Aussen: {office['outside_temp_c']} C", font=body_font, fill=0) + y += 32 + draw.text((24, y), f"Solar: {office.get('solar_daily', 'unknown')} kWh", font=body_font, fill=0) + y += 32 + draw.text((24, y), f"Batterie: {office.get('battery_soc', 'unknown')}%", font=body_font, fill=0) + y += 32 + draw.text((24, y), f"Netz: {office.get('power_consumption', 'unknown')} kWh", font=body_font, fill=0) + + draw.line((24, 330, DISPLAY_WIDTH - 24, 330), fill=0, width=2) + events = feed.get("calendar_events", []) + y = 350 + if not events: + draw.text((24, y), "Keine Termine", font=body_font, fill=0) + else: + ev_font = load_font(22, bold=False) + date_font = load_font(22, bold=True) + grouped: dict[str, list[str]] = {} + order: list[str] = [] + for ev in events[:15]: + start = str(ev.get("start", "")) + try: + start_dt = dt.datetime.fromisoformat(start) + date_key = start_dt.strftime("%a %d.%m.") + except (ValueError, TypeError): + date_key = start[:10] + if date_key not in grouped: + grouped[date_key] = [] + order.append(date_key) + grouped[date_key].append(str(ev.get("summary", "?"))) + + for date_key in order: + if y > 760: + break + draw.text((24, y), date_key, font=date_font, fill=0) + y += 28 + summaries = ", ".join(grouped[date_key]) + y = draw_wrapped(draw, (24, y), summaries, ev_font, DISPLAY_WIDTH - 48, line_gap=3, max_lines=2) + y += 12 + + return img + + +def render_png(feed: dict[str, Any]) -> bytes: + img = render_status_image(feed) + out = io.BytesIO() + img.save(out, format="PNG", optimize=True) + return out.getvalue() + + +def render_bmp(feed: dict[str, Any]) -> bytes: + img = render_status_image(feed) + out = io.BytesIO() + img.save(out, format="BMP") + return out.getvalue() + + +def render_messages_image(feed: dict[str, Any]) -> Any: + if Image is None or ImageDraw is None or ImageFont is None: + raise RuntimeError("Pillow is not installed. Run: python3 -m pip install -r x3-feed/requirements.txt") + + img = Image.new("L", (DISPLAY_WIDTH, DISPLAY_HEIGHT), 255) + draw = ImageDraw.Draw(img) + + title_font = load_font(36, bold=True) + section_font = load_font(24, bold=True) + body_font = load_font(21, bold=False) + small_font = load_font(17, bold=False) + + draw.rectangle((16, 16, DISPLAY_WIDTH - 16, DISPLAY_HEIGHT - 16), outline=0, width=3) + draw.text((24, 34), "Messages", font=title_font, fill=0) + draw.text((24, 80), str(CONFIG.get("subtitle") or "Home Assistant status"), font=body_font, fill=0) + pagination = feed.get("pagination", {}) + page_text = f"Page {pagination.get('page', 1)}/{pagination.get('page_count', 1)}" + draw.text((24, 112), f"{now_local().strftime('%a %H:%M')} {page_text}", font=section_font, fill=0) + draw.line((24, 150, DISPLAY_WIDTH - 24, 150), fill=0, width=2) + + messages = feed.get("messages", []) + y = 170 + if not messages: + draw.text((24, y), "No recent messages", font=body_font, fill=0) + for message in messages: + if y > DISPLAY_HEIGHT - 80: + break + source = message.get("source") or message.get("title") or "Message" + header = f"{message.get('time', 'unknown')} {source}" + draw_wrapped(draw, (24, y), header, section_font, DISPLAY_WIDTH - 48, line_gap=3, max_lines=1) + y += 32 + y = draw_wrapped(draw, (24, y), message.get("message", ""), body_font, DISPLAY_WIDTH - 48, line_gap=4, max_lines=3) + y += 12 + draw.line((24, y, DISPLAY_WIDTH - 24, y), fill=0, width=1) + y += 14 + + nav_hint = "Up/Down pages Left/Right tabs" + if not pagination.get("has_previous") and not pagination.get("has_next"): + nav_hint = "Left/Right tabs" + footer = f"{nav_hint} {feed['generated_at'].split('+')[0].replace('T', ' ')}" + draw.text((24, DISPLAY_HEIGHT - 42), footer, font=small_font, fill=0) + return img + + +def render_messages_png(feed: dict[str, Any]) -> bytes: + img = render_messages_image(feed) + out = io.BytesIO() + img.save(out, format="PNG", optimize=True) + return out.getvalue() + + +def render_messages_bmp(feed: dict[str, Any]) -> bytes: + img = render_messages_image(feed) + out = io.BytesIO() + img.save(out, format="BMP") + return out.getvalue() + + +def render_opds(feed: dict[str, Any]) -> str: + updated = feed["generated_at"] + title = "Home Assistant Dashboard" + entry_title = "X3 Dashboard" + return f""" + + urn:crosspoint:x3-ha-dashboard + {html.escape(title)} + {html.escape(updated)} + + urn:crosspoint:x3-ha-dashboard:current + {html.escape(entry_title)} + {html.escape(updated)} + Home Assistant + {html.escape(feed["headline"])} · {html.escape(now_local().strftime('%a %H:%M'))} + + + +""" + + +def render_epub(feed: dict[str, Any]) -> bytes: + png = render_png(feed) + title = "X3 Home Assistant Dashboard" + body = f""" + + + + {html.escape(title)} + + + + Home Assistant dashboard + + +""" + nav = f""" + +{html.escape(title)} + + + + +""" + opf = f""" + + + urn:crosspoint:x3-ha-dashboard:current + {html.escape(title)} + Home Assistant + en + {feed["generated_at"].split("+")[0]}Z + + + + + + + + + + +""" + container = """ + + + + + +""" + + out = io.BytesIO() + with zipfile.ZipFile(out, "w") as epub: + epub.writestr(zipfile.ZipInfo("mimetype"), "application/epub+zip", compress_type=zipfile.ZIP_STORED) + epub.writestr("META-INF/container.xml", container, compress_type=zipfile.ZIP_DEFLATED) + epub.writestr("OEBPS/content.opf", opf, compress_type=zipfile.ZIP_DEFLATED) + epub.writestr("OEBPS/nav.xhtml", nav, compress_type=zipfile.ZIP_DEFLATED) + epub.writestr("OEBPS/dashboard.xhtml", body, compress_type=zipfile.ZIP_DEFLATED) + epub.writestr("OEBPS/status.png", png, compress_type=zipfile.ZIP_DEFLATED) + return out.getvalue() + + +class X3Handler(BaseHTTPRequestHandler): + server_version = "X3Feed/0.1" + + def log_message(self, fmt: str, *args: Any) -> None: + print(f"{self.address_string()} - {fmt % args}") + + def send_bytes(self, body: bytes, content_type: str, status: HTTPStatus = HTTPStatus.OK) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def send_text(self, text: str, content_type: str, status: HTTPStatus = HTTPStatus.OK) -> None: + self.send_bytes(text.encode("utf-8"), content_type, status) + + def do_GET(self) -> None: # noqa: N802 + parsed_url = urllib.parse.urlparse(self.path) + path = parsed_url.path.rstrip("/") or "/" + query = urllib.parse.parse_qs(parsed_url.query) + message_page = query_positive_int(query.get("page"), 1, minimum=1, maximum=99) + if path in {"/favicon.ico"}: + self.send_response(HTTPStatus.NO_CONTENT) + self.end_headers() + return + try: + feed = build_feed() + if path in {"/", "/x3"}: + html_body = f""" +X3 Feed + + +

X3 Feed

+

Manifest · JSON · Markdown · SVG · PNG · BMP · EPUB · Messages · OPDS

+X3 status preview +
{html.escape(render_markdown(feed))}
+""" + self.send_text(html_body, "text/html; charset=utf-8") + elif path == "/x3/dashboard.json": + host = self.headers.get("Host", "") + base_url = f"http://{host}" if host else "" + self.send_text(json.dumps(build_dashboard_manifest(base_url), indent=2, sort_keys=True) + "\n", "application/json; charset=utf-8") + elif path == "/x3/status.json": + self.send_text(json.dumps(feed, indent=2, sort_keys=True) + "\n", "application/json; charset=utf-8") + elif path in {"/x3/status.md", "/x3/status.txt"}: + self.send_text(render_markdown(feed), "text/markdown; charset=utf-8") + elif path == "/x3/status.svg": + self.send_text(render_svg(feed), "image/svg+xml; charset=utf-8") + elif path == "/x3/status.png": + self.send_bytes(render_png(feed), "image/png") + elif path == "/x3/status.bmp": + self.send_bytes(render_bmp(feed), "image/bmp") + elif path == "/x3/status.epub": + self.send_bytes(render_epub(feed), "application/epub+zip") + elif path == "/x3/messages.json": + message_feed = build_message_feed(page=message_page) + self.send_text(json.dumps(message_feed, indent=2, sort_keys=True) + "\n", "application/json; charset=utf-8") + elif path in {"/x3/messages.md", "/x3/messages.txt"}: + message_feed = build_message_feed(page=message_page) + self.send_text(render_messages_markdown(message_feed), "text/markdown; charset=utf-8") + elif path == "/x3/messages.png": + message_feed = build_message_feed(page=message_page) + self.send_bytes(render_messages_png(message_feed), "image/png") + elif path == "/x3/messages.bmp": + message_feed = build_message_feed(page=message_page) + self.send_bytes(render_messages_bmp(message_feed), "image/bmp") + elif path == "/x3/opds": + self.send_text(render_opds(feed), "application/atom+xml;profile=opds-catalog; charset=utf-8") + else: + self.send_text("Not found\n", "text/plain; charset=utf-8", HTTPStatus.NOT_FOUND) + except Exception as exc: + payload = {"error": str(exc), "type": type(exc).__name__} + self.send_text(json.dumps(payload, indent=2) + "\n", "application/json; charset=utf-8", HTTPStatus.INTERNAL_SERVER_ERROR) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Serve XTeink X3 Home Assistant feed") + parser.add_argument("--host", default=os.getenv("X3_FEED_HOST", "127.0.0.1")) + parser.add_argument("--port", type=int, default=int(os.getenv("X3_FEED_PORT", "8787"))) + parser.add_argument("--env-file", default=os.getenv("X3_ENV_FILE", str(DEFAULT_ENV_FILE))) + parser.add_argument("--options-file", default=os.getenv("X3_OPTIONS_FILE", str(DEFAULT_OPTIONS_FILE))) + args = parser.parse_args() + + global CONFIG + load_env(Path(args.env_file)) + CONFIG = load_options(Path(args.options_file)) + server = ThreadingHTTPServer((args.host, args.port), X3Handler) + print(f"X3 feed listening on http://{args.host}:{args.port}/") + try: + server.serve_forever() + except KeyboardInterrupt: + print("\nStopping X3 feed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())