#!/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 import sys 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 = [CONFIG.get("entity_ids", {}).get(k, "") for k in caldav_keys] caldav_entity_ids = [eid for eid in caldav_entity_ids if eid and eid.startswith("calendar.")] local_sensor_id = CONFIG.get("entity_ids", {}).get("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: start_raw = ev.get("start", "?") if isinstance(start_raw, dict): start_raw = start_raw.get("dateTime", start_raw.get("date", "?")) end_raw = ev.get("end", "?") if isinstance(end_raw, dict): end_raw = end_raw.get("dateTime", end_raw.get("date", "?")) all_events.append({ "summary": ev.get("summary", "?"), "start": start_raw, "end": end_raw, "source": "caldav", }) except Exception as exc: print(f"calendar fetch error for {cal_id}: {exc}", file=sys.stderr) 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) # Vergangene Termine herausfiltern (ab jetzt) now_iso = now.isoformat() all_events = [ev for ev in all_events if str(ev.get("start", "")) >= now_iso] # 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) draw.line((24, 260, DISPLAY_WIDTH - 24, 260), fill=0, width=2) events = feed.get("calendar_events", []) y = 284 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) import threading import time as _time def sync_shelly_cloud() -> None: """Fetch WS90 data from Shelly Cloud and push to HA as sensors.""" auth_key = str(CONFIG.get("shelly_cloud_key") or "").strip() server = str(CONFIG.get("shelly_cloud_server") or "").strip() device_id = str(CONFIG.get("shelly_ws90_device_id") or "").strip() if not auth_key or not device_id: return try: data = urllib.parse.urlencode({"id": device_id, "auth_key": auth_key}).encode() req = urllib.request.Request(f"{server}/device/status", data=data, method="POST") with urllib.request.urlopen(req, timeout=15) as resp: result = json.loads(resp.read().decode("utf-8")) if not result.get("isok"): print(f"Shelly Cloud error: {result}", file=sys.stderr) return status = result.get("data", {}).get("device_status", {}) sensors = { "sensor.shelly_cloud_ws90_precipitation": ( str(status.get("precipitation:0", {}).get("value", "unknown")), {"unit_of_measurement": "mm", "device_class": "precipitation", "state_class": "total_increasing", "friendly_name": "WS90 Niederschlag (Cloud)"}, ), "sensor.shelly_cloud_ws90_temperature": ( str(status.get("temperature:0", {}).get("tC", "unknown")), {"unit_of_measurement": "°C", "device_class": "temperature", "state_class": "measurement", "friendly_name": "WS90 Temperatur (Cloud)"}, ), "sensor.shelly_cloud_ws90_humidity": ( str(status.get("humidity:0", {}).get("rh", "unknown")), {"unit_of_measurement": "%", "device_class": "humidity", "state_class": "measurement", "friendly_name": "WS90 Luftfeuchtigkeit (Cloud)"}, ), "sensor.shelly_cloud_ws90_wind_speed": ( str(status.get("speed:0", {}).get("value", "unknown")), {"unit_of_measurement": "km/h", "device_class": "wind_speed", "state_class": "measurement", "friendly_name": "WS90 Windgeschwindigkeit (Cloud)"}, ), "sensor.shelly_cloud_ws90_gust_speed": ( str(status.get("speed:1", {}).get("value", "unknown")), {"unit_of_measurement": "km/h", "device_class": "wind_speed", "state_class": "measurement", "friendly_name": "WS90 Böengeschwindigkeit (Cloud)"}, ), "sensor.shelly_cloud_ws90_wind_direction": ( str(status.get("direction:0", {}).get("value", "unknown")), {"unit_of_measurement": "°", "device_class": "wind_direction", "state_class": "measurement", "friendly_name": "WS90 Windrichtung (Cloud)"}, ), "sensor.shelly_cloud_ws90_pressure": ( str(status.get("pressure:0", {}).get("value", "unknown")), {"unit_of_measurement": "hPa", "device_class": "pressure", "state_class": "measurement", "friendly_name": "WS90 Luftdruck (Cloud)"}, ), "sensor.shelly_cloud_ws90_uv_index": ( str(status.get("UV:0", {}).get("value", "unknown")), {"state_class": "measurement", "friendly_name": "WS90 UV-Index (Cloud)"}, ), "sensor.shelly_cloud_ws90_illuminance": ( str(status.get("illuminance:0", {}).get("lux", "unknown")), {"unit_of_measurement": "lx", "device_class": "illuminance", "state_class": "measurement", "friendly_name": "WS90 Beleuchtungsstärke (Cloud)"}, ), "sensor.shelly_cloud_ws90_battery": ( str(status.get("devicepower:0", {}).get("battery", {}).get("percent", "unknown")), {"unit_of_measurement": "%", "device_class": "battery", "state_class": "measurement", "friendly_name": "WS90 Batterie (Cloud)"}, ), "binary_sensor.shelly_cloud_ws90_rain_detected": ( "on" if status.get("moisture_alarm:0", {}).get("value") else "off", {"device_class": "moisture", "friendly_name": "WS90 Regen erkannt (Cloud)"}, ), } client = HomeAssistantClient() for entity_id, (state, attrs) in sensors.items(): payload = {"state": state, "attributes": attrs} body = json.dumps(payload).encode("utf-8") req = urllib.request.Request( f"{client.base_url}/api/states/{entity_id}", data=body, method="POST", headers={"Authorization": f"Bearer {client.token}", "Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=10) as resp: resp.read() precip = status.get("precipitation:0", {}).get("value", "?") print(f"Shelly Cloud sync: precip={precip}mm") except Exception as exc: print(f"Shelly Cloud sync error: {exc}", file=sys.stderr) def shelly_cloud_loop() -> None: """Background thread: sync Shelly Cloud every 10 minutes.""" while True: sync_shelly_cloud() _time.sleep(600) 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)) shelly_thread = threading.Thread(target=shelly_cloud_loop, daemon=True) shelly_thread.start() 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())