Tilman Hampl
1249 lines
48 KiB
Python
1249 lines
48 KiB
Python
#!/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'<text x="24" y="{y}" font-size="{size}" font-weight="{weight}">{html.escape(text)}</text>'
|
|
for y, text, size, weight in rows
|
|
)
|
|
return f"""<svg xmlns="http://www.w3.org/2000/svg" width="{DISPLAY_WIDTH}" height="{DISPLAY_HEIGHT}" viewBox="0 0 {DISPLAY_WIDTH} {DISPLAY_HEIGHT}">
|
|
<rect width="100%" height="100%" fill="white"/>
|
|
<rect x="16" y="16" width="{DISPLAY_WIDTH - 32}" height="{DISPLAY_HEIGHT - 32}" fill="none" stroke="black" stroke-width="3"/>
|
|
<line x1="24" y1="122" x2="{DISPLAY_WIDTH - 24}" y2="122" stroke="black" stroke-width="2"/>
|
|
<line x1="24" y1="300" x2="{DISPLAY_WIDTH - 24}" y2="300" stroke="black" stroke-width="2"/>
|
|
<line x1="24" y1="408" x2="{DISPLAY_WIDTH - 24}" y2="408" stroke="black" stroke-width="2"/>
|
|
<g font-family="Arial, Helvetica, sans-serif" fill="black">
|
|
{text_nodes}
|
|
</g>
|
|
</svg>
|
|
"""
|
|
|
|
|
|
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"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom">
|
|
<id>urn:crosspoint:x3-ha-dashboard</id>
|
|
<title>{html.escape(title)}</title>
|
|
<updated>{html.escape(updated)}</updated>
|
|
<entry>
|
|
<id>urn:crosspoint:x3-ha-dashboard:current</id>
|
|
<title>{html.escape(entry_title)}</title>
|
|
<updated>{html.escape(updated)}</updated>
|
|
<author><name>Home Assistant</name></author>
|
|
<summary>{html.escape(feed["headline"])} · {html.escape(now_local().strftime('%a %H:%M'))}</summary>
|
|
<link rel="http://opds-spec.org/acquisition" href="/x3/status.epub" type="application/epub+zip"/>
|
|
</entry>
|
|
</feed>
|
|
"""
|
|
|
|
|
|
def render_epub(feed: dict[str, Any]) -> bytes:
|
|
png = render_png(feed)
|
|
title = "X3 Home Assistant Dashboard"
|
|
body = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<!DOCTYPE html>
|
|
<html xmlns="http://www.w3.org/1999/xhtml" lang="en">
|
|
<head>
|
|
<title>{html.escape(title)}</title>
|
|
<style>
|
|
body {{ margin: 0; padding: 0; }}
|
|
img {{ display: block; width: 100%; height: auto; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<img src="status.png" alt="Home Assistant dashboard"/>
|
|
</body>
|
|
</html>
|
|
"""
|
|
nav = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<html xmlns="http://www.w3.org/1999/xhtml" xmlns:epub="http://www.idpf.org/2007/ops" lang="en">
|
|
<head><title>{html.escape(title)}</title></head>
|
|
<body>
|
|
<nav epub:type="toc" id="toc">
|
|
<h1>{html.escape(title)}</h1>
|
|
<ol><li><a href="dashboard.xhtml">Dashboard</a></li></ol>
|
|
</nav>
|
|
</body>
|
|
</html>
|
|
"""
|
|
opf = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<package xmlns="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="book-id">
|
|
<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">
|
|
<dc:identifier id="book-id">urn:crosspoint:x3-ha-dashboard:current</dc:identifier>
|
|
<dc:title>{html.escape(title)}</dc:title>
|
|
<dc:creator>Home Assistant</dc:creator>
|
|
<dc:language>en</dc:language>
|
|
<meta property="dcterms:modified">{feed["generated_at"].split("+")[0]}Z</meta>
|
|
</metadata>
|
|
<manifest>
|
|
<item id="nav" href="nav.xhtml" media-type="application/xhtml+xml" properties="nav"/>
|
|
<item id="dashboard" href="dashboard.xhtml" media-type="application/xhtml+xml"/>
|
|
<item id="status-png" href="status.png" media-type="image/png"/>
|
|
</manifest>
|
|
<spine>
|
|
<itemref idref="dashboard"/>
|
|
</spine>
|
|
</package>
|
|
"""
|
|
container = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">
|
|
<rootfiles>
|
|
<rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/>
|
|
</rootfiles>
|
|
</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"""<!doctype html>
|
|
<html><head><meta charset="utf-8"><title>X3 Feed</title>
|
|
<style>body{{font-family:system-ui,sans-serif;margin:24px;background:#eee}} img{{border:1px solid #222;background:white;width:264px;height:396px;image-rendering:auto}} pre{{background:white;padding:16px;max-width:760px;white-space:pre-wrap}}</style>
|
|
</head><body>
|
|
<h1>X3 Feed</h1>
|
|
<p><a href="/x3/dashboard.json">Manifest</a> · <a href="/x3/status.json">JSON</a> · <a href="/x3/status.md">Markdown</a> · <a href="/x3/status.svg">SVG</a> · <a href="/x3/status.png">PNG</a> · <a href="/x3/status.bmp">BMP</a> · <a href="/x3/status.epub">EPUB</a> · <a href="/x3/messages.bmp">Messages</a> · <a href="/x3/opds">OPDS</a></p>
|
|
<img src="/x3/status.png" alt="X3 status preview">
|
|
<pre>{html.escape(render_markdown(feed))}</pre>
|
|
</body></html>"""
|
|
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())
|