Kent
7c521f8d1f
Sonene trenger ikke stå i [zones] på sentralen. En sone sender hjerteslag hvert 30. s (POST /api/mesh/register med navn, egen adresse og mesh-nøkkelen), dukker opp i lista med en gang, og tas ut etter 90 s uten slag. Riktig nøkkel er godkjenningen. Statisk [zones] vinner over et navn som melder seg (409) – en skrivefeil i config skal ikke kunne overstyres av en boks på nettet. Hvorfor push og ikke pull fra NetBird: netbird status --json gir navn, IP og status per peer, men ikke grupper. Pull ville krevd en service-token mot management-serveren og virket bare over NetBird. Push virker likt over NetBird og LAN, uten token, og gir liveness gratis. Sonen finner sentralen lagvis: ANNOUNCER_SENTRAL hvis satt (NetBird-DNS-navn, konvensjon announcer-<navn>.skynet.corp), ellers mDNS (_announcer._tcp, discovery.py). Egen adresse regnes ut som ruten mot sentralen – over NetBird blir det wt0-adressen, på LAN LAN-adressen – med ANNOUNCER_URL som overstyring. Hjerteslag hvert 5. s til første OK, så oppstart ikke tar et halvt minutt. Containeren bruker nå vertens nett. mDNS-multicast når ikke LAN-et fra bridge-nettet, så det var ingen vei utenom – og det ga tilbake ufw: det gamle Docker-smutthullet er borte, portregler virker igjen, og BIND_IP er appens eget --host. Appen lytter alltid også på 127.0.0.1 for helsesjekken. install.sh tar NETBIRD_SETUP_KEY: installerer NetBird, melder maskina inn som announcer-<navn>, venter på wt0, binder til NetBird-adressen. Management-URL leses fra /etc/netbird/config.json når den finnes. Rollen avgjøres av config (Config.role): sentral hvis [zones] eller ANNOUNCER_ROLE=sentral, sone hvis bare nøkkelen er satt, ellers som før. Dispatcher og ZonePoller slår opp soner ved hver bruk, siden de kommer og går. register følger, som play, aldri med «alt åpent». Flask sorterte JSON-nøkler alfabetisk, så sonene kom i feil rekkefølge i /api/status – app.json.sort_keys = False. 25 nye tester, 173 totalt. Verifisert i compose på host-nett: to soner meldte seg innen 4 s uten noe [zones], fordeling til en av dem havnet i kø, avahi-browse på verten ser sentralen, en stoppet sone ble tatt ut etter 90 s og var tilbake 2 s etter start. Underveis fant live-testen at testkonfigen fortsatt hadde statiske a/b – og sentralen avviste hjerteslagene med 409, som den skal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
628 lines
24 KiB
Python
628 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
"""Mesh: én sentral som fordeler utrop til flere soner.
|
||
|
||
Hver sone er en vanlig announcer med egen kø og eget anlegg. Sentralen er en
|
||
vanlig announcer som i tillegg har [zones] i config. Det er hele forskjellen;
|
||
det finnes ingen «mode», og en sone kan brukes alene hvis sentralen er nede.
|
||
|
||
Bærende valg:
|
||
|
||
- Sentralen syntetiserer én gang og sender ferdig WAV. Sonen trenger da verken
|
||
nett, edge-tts eller Pipers stemmer for å spille det den får.
|
||
- Køene forblir per sone. Er hovedscenen opptatt med en lang melding, skal
|
||
ikke en resepsjonsmelding vente. Dispatcheren fordeler og går videre; hver
|
||
sone køer selv.
|
||
- Faller én sone ut, feiler bare den. De andre får meldingen.
|
||
|
||
Modulen kjenner verken Flask eller Player. Den får funksjoner inn:
|
||
synthesize, play_local og send. Det er det som gjør den testbar uten nett.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import queue
|
||
import re
|
||
import secrets
|
||
import socket
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from dataclasses import asdict, dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
log = logging.getLogger("announcer.mesh")
|
||
|
||
ZONE_NAME = re.compile(r"[A-Za-z0-9_æøåÆØÅ-]{1,32}")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Soner
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Zone:
|
||
name: str
|
||
url: str = "" # tom = sentralens eget anlegg
|
||
|
||
@property
|
||
def local(self) -> bool:
|
||
return not self.url
|
||
|
||
|
||
def zones_from_config(raw: dict[str, str]) -> dict[str, Zone]:
|
||
"""[zones] i config: navn -> "local" eller http(s)://…. Feiler høyt på
|
||
noe annet, så en skrivefeil oppdages ved oppstart og ikke ved første utrop."""
|
||
out: dict[str, Zone] = {}
|
||
for name, target in raw.items():
|
||
name = str(name).strip()
|
||
if not ZONE_NAME.fullmatch(name):
|
||
raise ValueError(f"[zones]: ugyldig sonenavn {name!r}")
|
||
target = str(target).strip()
|
||
if target.lower() == "local":
|
||
out[name] = Zone(name)
|
||
elif target.startswith(("http://", "https://")):
|
||
out[name] = Zone(name, target.rstrip("/"))
|
||
else:
|
||
raise ValueError(f'[zones] {name}: {target!r} er verken "local" eller en http-URL')
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Én fordeling = én melding til N soner
|
||
# --------------------------------------------------------------------------
|
||
|
||
# sending -> queued (sonen tok imot) | unreachable (nådde ikke fram)
|
||
# | rejected (sonen sa nei, f.eks. feil nøkkel) | failed (syntese/lokalt)
|
||
ZONE_STATUSES = ("sending", "queued", "unreachable", "rejected", "failed")
|
||
|
||
|
||
@dataclass
|
||
class ZoneResult:
|
||
status: str = "sending"
|
||
item_id: int | None = None # kø-id på sonen
|
||
error: str = ""
|
||
|
||
|
||
@dataclass
|
||
class Dispatch:
|
||
id: str
|
||
text: str
|
||
voice: str
|
||
speed: float
|
||
chime: str
|
||
zones: dict[str, ZoneResult]
|
||
user: str = ""
|
||
subject: str = ""
|
||
method: str = "none"
|
||
sender: str = "" # IP-en forespørselen kom fra
|
||
created: float = field(default_factory=time.time)
|
||
schedule_id: str = ""
|
||
error: str = "" # syntesefeil, da feiler alle soner likt
|
||
|
||
@property
|
||
def pending(self) -> bool:
|
||
return any(z.status == "sending" for z in self.zones.values())
|
||
|
||
@property
|
||
def accepted(self) -> list[str]:
|
||
return [n for n, z in self.zones.items() if z.status == "queued"]
|
||
|
||
def summary(self) -> str:
|
||
"""«hovedscene: kø #3, crew: nede» – til planen og loggen."""
|
||
parts = []
|
||
for name, z in self.zones.items():
|
||
if z.status == "queued":
|
||
parts.append(f"{name}: kø #{z.item_id}")
|
||
elif z.status == "unreachable":
|
||
parts.append(f"{name}: nede")
|
||
else:
|
||
parts.append(f"{name}: {z.status}")
|
||
return ", ".join(parts)
|
||
|
||
def public(self, *, show_ip: bool = False) -> dict[str, Any]:
|
||
return {
|
||
"id": self.id,
|
||
"text": self.text,
|
||
"voice": self.voice,
|
||
"user": self.user,
|
||
"sender": self.sender if show_ip else "",
|
||
"created": self.created,
|
||
"error": self.error,
|
||
"zones": {n: asdict(z) for n, z in self.zones.items()},
|
||
}
|
||
|
||
|
||
class Dispatcher:
|
||
"""Egen tråd og egen kø, atskilt fra Player. Gjør bare fordeling.
|
||
|
||
synthesize(text, voice, speed, chime) -> Path
|
||
play_local(dispatch, path) -> int kø-id lokalt
|
||
send(zone, path, dispatch) -> (status, item_id, error)
|
||
"""
|
||
|
||
def __init__(self, zones: "dict[str, Zone] | Callable[[], dict[str, Zone]]", *,
|
||
synthesize: Callable, play_local: Callable, send: Callable,
|
||
timeout: float = 5.0, on_finished: Callable | None = None) -> None:
|
||
# Soner kommer og går (hjerteslag), så de slås opp ved hver bruk.
|
||
# Et fast dict godtas også, for tester og statisk config.
|
||
self._zones = zones if callable(zones) else (lambda: zones)
|
||
self.synthesize = synthesize
|
||
self.play_local = play_local
|
||
self.send = send
|
||
self.timeout = float(timeout)
|
||
self.on_finished = on_finished
|
||
self.q: queue.Queue[Dispatch] = queue.Queue()
|
||
self.history: list[Dispatch] = []
|
||
self.lock = threading.Lock()
|
||
self._thread: threading.Thread | None = None
|
||
|
||
# -- oppslag -----------------------------------------------------------
|
||
@property
|
||
def zones(self) -> dict[str, Zone]:
|
||
return self._zones()
|
||
|
||
def names(self) -> list[str]:
|
||
return list(self.zones)
|
||
|
||
def resolve(self, requested: Any) -> list[str]:
|
||
"""Tom/None = alle. Ukjent navn kaster ValueError før noe sendes."""
|
||
zones = self.zones
|
||
if not requested:
|
||
return list(zones)
|
||
if isinstance(requested, str):
|
||
requested = [requested]
|
||
if not isinstance(requested, (list, tuple)):
|
||
raise ValueError("zones må være en liste med sonenavn")
|
||
out: list[str] = []
|
||
for name in requested:
|
||
name = str(name).strip()
|
||
if name not in zones:
|
||
raise ValueError(f"Ukjent sone: {name}")
|
||
if name not in out:
|
||
out.append(name)
|
||
return out
|
||
|
||
# -- innlevering -------------------------------------------------------
|
||
def submit(self, text: str, voice: str, speed: float, chime: str, zone_names: list[str],
|
||
*, user: str = "", subject: str = "", method: str = "none", sender: str = "",
|
||
schedule_id: str = "") -> Dispatch:
|
||
d = Dispatch(
|
||
id=secrets.token_hex(4), text=text, voice=voice, speed=speed, chime=chime,
|
||
zones={n: ZoneResult() for n in zone_names},
|
||
user=user, subject=subject, method=method, sender=sender, schedule_id=schedule_id,
|
||
)
|
||
with self.lock:
|
||
self.history.append(d)
|
||
del self.history[:-50]
|
||
self.q.put(d)
|
||
return d
|
||
|
||
def snapshot(self, *, show_ip: bool = False) -> list[dict]:
|
||
with self.lock:
|
||
return [d.public(show_ip=show_ip) for d in reversed(self.history)][:15]
|
||
|
||
# -- tråden ------------------------------------------------------------
|
||
def start(self) -> None:
|
||
if self._thread is None:
|
||
self._thread = threading.Thread(target=self._loop, daemon=True, name="dispatcher")
|
||
self._thread.start()
|
||
|
||
def _loop(self) -> None:
|
||
while True:
|
||
try:
|
||
self.run_once()
|
||
except Exception: # noqa: BLE001
|
||
log.exception("dispatcheren feilet")
|
||
|
||
def run_once(self, timeout: float | None = None) -> Dispatch | None:
|
||
"""Én fordeling, fra syntese til svar fra hver sone. Returnerer den."""
|
||
try:
|
||
d = self.q.get(timeout=timeout)
|
||
except queue.Empty:
|
||
return None
|
||
try:
|
||
path = self.synthesize(d.text, d.voice, d.speed, d.chime)
|
||
except Exception as e: # noqa: BLE001
|
||
log.exception("syntese feilet for fordeling #%s", d.id)
|
||
d.error = str(e)[:200]
|
||
for z in d.zones.values():
|
||
z.status, z.error = "failed", d.error
|
||
self._finish(d)
|
||
return d
|
||
|
||
threads: list[threading.Thread] = []
|
||
zones = self.zones
|
||
for name, result in d.zones.items():
|
||
zone = zones.get(name)
|
||
if zone is None:
|
||
# Meldt av mellom innlevering og fordeling.
|
||
result.status, result.error = "unreachable", "sonen er borte"
|
||
continue
|
||
if zone.local:
|
||
try:
|
||
result.item_id = self.play_local(d, path)
|
||
result.status = "queued"
|
||
except Exception as e: # noqa: BLE001
|
||
result.status, result.error = "failed", str(e)[:200]
|
||
else:
|
||
# Hver sone i sin tråd, så en som henger ikke holder igjen de
|
||
# andre lenger enn timeouten.
|
||
t = threading.Thread(target=self._send_one, args=(zone, path, d, result),
|
||
daemon=True, name=f"send-{name}")
|
||
t.start()
|
||
threads.append(t)
|
||
for t in threads:
|
||
t.join(self.timeout + 1.0)
|
||
for name, result in d.zones.items():
|
||
if result.status == "sending":
|
||
result.status, result.error = "unreachable", "svarte ikke i tide"
|
||
self._finish(d)
|
||
return d
|
||
|
||
def _send_one(self, zone: Zone, path: Path, d: Dispatch, result: ZoneResult) -> None:
|
||
try:
|
||
status, item_id, error = self.send(zone, path, d)
|
||
except Exception as e: # noqa: BLE001
|
||
status, item_id, error = "unreachable", None, str(e)[:200]
|
||
result.status, result.item_id, result.error = status, item_id, error
|
||
|
||
def _finish(self, d: Dispatch) -> None:
|
||
log.info("fordeling #%s: %s", d.id, d.summary())
|
||
if self.on_finished:
|
||
try:
|
||
self.on_finished(d)
|
||
except Exception: # noqa: BLE001
|
||
log.exception("on_finished feilet")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Transport: sentral -> sone, stdlib urllib
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def send_to_zone(zone: Zone, path: Path, d: Dispatch, *, secret: str,
|
||
timeout: float = 5.0) -> tuple[str, int | None, str]:
|
||
"""POST /api/play på sonen: multipart med fila og metadata.
|
||
|
||
Returnerer (status, kø-id på sonen, feilmelding). Kaster ikke; alt som
|
||
kan gå galt blir en status, siden kallet skjer i en tråd uten noen å
|
||
kaste til.
|
||
"""
|
||
meta = json.dumps({
|
||
"origin_id": d.id, "text": d.text, "voice": d.voice,
|
||
"user": d.user, "subject": d.subject, "method": d.method,
|
||
}, ensure_ascii=False).encode()
|
||
boundary = "----announcer-" + secrets.token_hex(8)
|
||
nl = b"\r\n"
|
||
body = b"".join([
|
||
b"--" + boundary.encode() + nl,
|
||
b'Content-Disposition: form-data; name="meta"' + nl,
|
||
b"Content-Type: application/json; charset=utf-8" + nl + nl,
|
||
meta + nl,
|
||
b"--" + boundary.encode() + nl,
|
||
b'Content-Disposition: form-data; name="wav"; filename="' + path.name.encode() + b'"' + nl,
|
||
b"Content-Type: audio/wav" + nl + nl,
|
||
path.read_bytes() + nl,
|
||
b"--" + boundary.encode() + b"--" + nl,
|
||
])
|
||
req = urllib.request.Request(
|
||
f"{zone.url}/api/play",
|
||
data=body,
|
||
headers={
|
||
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
||
"X-Mesh-Key": secret,
|
||
"Accept": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310
|
||
data = json.loads(r.read().decode())
|
||
return "queued", int(data["id"]), ""
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
msg = json.loads(e.read().decode()).get("error", "")
|
||
except Exception: # noqa: BLE001
|
||
msg = ""
|
||
return "rejected", None, f"{e.code} {msg or e.reason}".strip()
|
||
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
||
reason = getattr(e, "reason", e)
|
||
return "unreachable", None, str(reason)[:120]
|
||
except (ValueError, KeyError) as e:
|
||
return "rejected", None, f"uventet svar ({e})"
|
||
|
||
|
||
def fetch_zone_status(zone: Zone, *, secret: str, timeout: float = 2.0) -> dict[str, Any]:
|
||
"""GET /api/status på sonen. Returnerer det UI-et trenger, aldri kaster."""
|
||
req = urllib.request.Request(
|
||
f"{zone.url}/api/status",
|
||
headers={"X-Mesh-Key": secret, "Accept": "application/json"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310
|
||
d = json.loads(r.read().decode())
|
||
return {
|
||
"reachable": True,
|
||
"current": d.get("current"),
|
||
"queued": len(d.get("queued") or []),
|
||
}
|
||
except urllib.error.HTTPError as e:
|
||
return {"reachable": False, "error": f"{e.code} {e.reason}"}
|
||
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
||
return {"reachable": False, "error": str(getattr(e, "reason", e))[:120]}
|
||
except ValueError as e:
|
||
return {"reachable": False, "error": f"uventet svar ({e})"}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Statuspoller: én i bakgrunnen, så N nettlesere ikke gir N x soner kall
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class ZonePoller:
|
||
"""Henter status fra hver ekstern sone med jevne mellomrom og cacher den.
|
||
|
||
Den lokale sonen slås opp ferskt ved hvert snapshot via local_status, så
|
||
eget anlegg aldri er to sekunder på etterskudd.
|
||
"""
|
||
|
||
def __init__(self, zones: "dict[str, Zone] | Callable[[], dict[str, Zone]]", *,
|
||
fetch: Callable, interval: float = 2.0,
|
||
local_status: Callable[[], dict] | None = None) -> None:
|
||
self._zones = zones if callable(zones) else (lambda: zones)
|
||
self.fetch = fetch
|
||
self.interval = float(interval)
|
||
self.local_status = local_status
|
||
self.lock = threading.Lock()
|
||
self.cache: dict[str, dict] = {}
|
||
self._thread: threading.Thread | None = None
|
||
|
||
@property
|
||
def zones(self) -> dict[str, Zone]:
|
||
return self._zones()
|
||
|
||
def poll_once(self) -> None:
|
||
remote = [z for z in self.zones.values() if not z.local]
|
||
results: dict[str, dict] = {}
|
||
|
||
def one(z: Zone) -> None:
|
||
try:
|
||
results[z.name] = self.fetch(z)
|
||
except Exception as e: # noqa: BLE001
|
||
results[z.name] = {"reachable": False, "error": str(e)[:120]}
|
||
|
||
threads = [threading.Thread(target=one, args=(z,), daemon=True) for z in remote]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(5.0)
|
||
with self.lock:
|
||
for z in remote:
|
||
self.cache[z.name] = results.get(z.name, {"reachable": False, "error": "svarte ikke"})
|
||
self.cache[z.name]["checked"] = time.time()
|
||
# Soner som er meldt av skal ikke bli liggende i cachen.
|
||
for gone in set(self.cache) - {z.name for z in remote}:
|
||
del self.cache[gone]
|
||
|
||
def snapshot(self) -> dict[str, dict]:
|
||
out: dict[str, dict] = {}
|
||
for z in self.zones.values():
|
||
if z.local:
|
||
out[z.name] = {"reachable": True, **(self.local_status() if self.local_status else {})}
|
||
else:
|
||
with self.lock:
|
||
out[z.name] = dict(self.cache.get(z.name, {"reachable": False, "error": "ikke spurt ennå"}))
|
||
return out
|
||
|
||
def start(self) -> None:
|
||
if self._thread is None:
|
||
self._thread = threading.Thread(target=self._loop, daemon=True, name="zone-poller")
|
||
self._thread.start()
|
||
|
||
def _loop(self) -> None:
|
||
while True:
|
||
try:
|
||
self.poll_once()
|
||
except Exception: # noqa: BLE001
|
||
log.exception("sonepolleren feilet")
|
||
time.sleep(self.interval)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Register: sonene sentralen kjenner akkurat nå
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class Registered:
|
||
zone: Zone
|
||
first_seen: float
|
||
last_seen: float
|
||
version: str = ""
|
||
|
||
|
||
class Registry:
|
||
"""Sonene en sentral kjenner: eget anlegg, statisk [zones], og de som har
|
||
meldt seg med hjerteslag. En som slutter å slå, tas ut etter stale_seconds.
|
||
|
||
Statisk config vinner alltid over et navn som melder seg – en skrivefeil
|
||
i config skal ikke kunne overstyres av en boks på nettet.
|
||
"""
|
||
|
||
def __init__(self, static: dict[str, Zone], *, local: Zone | None = None,
|
||
stale_seconds: float = 90.0) -> None:
|
||
self.local = local
|
||
self.static = dict(static)
|
||
self.stale = float(stale_seconds)
|
||
self.registered: dict[str, Registered] = {}
|
||
self.lock = threading.Lock()
|
||
|
||
def register(self, name: str, url: str, version: str = "") -> Zone:
|
||
name = str(name).strip()
|
||
if not ZONE_NAME.fullmatch(name):
|
||
raise ValueError(f"ugyldig sonenavn {name!r}")
|
||
if self.local and name == self.local.name:
|
||
raise ValueError(f"{name} er sentralens eget navn")
|
||
if name in self.static:
|
||
raise ValueError(f"{name} er satt i [zones] på sentralen")
|
||
url = str(url).strip().rstrip("/")
|
||
if not url.startswith(("http://", "https://")):
|
||
raise ValueError("url må være http(s)://…")
|
||
zone = Zone(name, url)
|
||
now = time.time()
|
||
with self.lock:
|
||
prev = self.registered.get(name)
|
||
if prev and prev.zone == zone:
|
||
prev.last_seen, prev.version = now, version
|
||
else:
|
||
if prev:
|
||
log.info("sone %s byttet adresse: %s -> %s", name, prev.zone.url, url)
|
||
else:
|
||
log.info("ny sone meldte seg: %s på %s", name, url)
|
||
self.registered[name] = Registered(zone, now, now, version)
|
||
return zone
|
||
|
||
def prune(self, now: float | None = None) -> list[str]:
|
||
now = time.time() if now is None else now
|
||
with self.lock:
|
||
gone = [n for n, r in self.registered.items() if now - r.last_seen > self.stale]
|
||
for n in gone:
|
||
del self.registered[n]
|
||
for n in gone:
|
||
log.warning("sone %s har ikke meldt seg på %d s – tatt ut", n, int(self.stale))
|
||
return gone
|
||
|
||
def zones(self) -> dict[str, Zone]:
|
||
out: dict[str, Zone] = {}
|
||
if self.local:
|
||
out[self.local.name] = self.local
|
||
out.update(self.static)
|
||
with self.lock:
|
||
for n in sorted(self.registered):
|
||
out[n] = self.registered[n].zone
|
||
return out
|
||
|
||
def info(self) -> dict[str, dict]:
|
||
"""Til UI-et: hvor sonen kommer fra, og når den sist sa noe."""
|
||
out: dict[str, dict] = {}
|
||
if self.local:
|
||
out[self.local.name] = {"source": "local", "url": ""}
|
||
for n, z in self.static.items():
|
||
out[n] = {"source": "static", "url": z.url}
|
||
with self.lock:
|
||
for n in sorted(self.registered):
|
||
r = self.registered[n]
|
||
out[n] = {"source": "registered", "url": r.zone.url,
|
||
"last_seen": r.last_seen, "version": r.version}
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Hjerteslag: sonens side
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def local_ip_towards(host: str, port: int = 8090) -> str:
|
||
"""IP-en denne maskina bruker for å nå host – altså den sentralen må
|
||
svare til. Ingen pakke sendes; connect() på en UDP-socket velger bare
|
||
rute. Over NetBird blir det wt0-adressen, på LAN LAN-adressen."""
|
||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
||
sock.connect((host, port))
|
||
return sock.getsockname()[0]
|
||
|
||
|
||
def post_register(sentral_url: str, *, name: str, url: str, secret: str,
|
||
version: str = "", timeout: float = 5.0) -> tuple[bool, str]:
|
||
body = json.dumps({"name": name, "url": url, "version": version}).encode()
|
||
req = urllib.request.Request(
|
||
f"{sentral_url.rstrip('/')}/api/mesh/register",
|
||
data=body,
|
||
headers={"Content-Type": "application/json", "X-Mesh-Key": secret, "Accept": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310
|
||
r.read()
|
||
return True, ""
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
msg = json.loads(e.read().decode()).get("error", "")
|
||
except Exception: # noqa: BLE001
|
||
msg = ""
|
||
return False, f"{e.code} {msg or e.reason}".strip()
|
||
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
||
return False, str(getattr(e, "reason", e))[:120]
|
||
|
||
|
||
class Heartbeat:
|
||
"""Melder denne sonen hos sentralen med jevne mellomrom.
|
||
|
||
sentral() gir sentralens URL akkurat nå – fra config, eller fra mDNS –
|
||
eller None når ingen er kjent. Da venter vi og prøver igjen; en sone skal
|
||
kunne starte før sentralen finnes.
|
||
"""
|
||
|
||
def __init__(self, *, name: str, port: int, secret: str, sentral: Callable[[], str | None],
|
||
url: str = "", interval: float = 30.0, version: str = "",
|
||
post: Callable = post_register) -> None:
|
||
self.name = name
|
||
self.port = int(port)
|
||
self.secret = secret
|
||
self.sentral = sentral
|
||
self.fixed_url = url.strip().rstrip("/")
|
||
self.interval = float(interval)
|
||
self.version = version
|
||
self.post = post
|
||
self.last_ok: float | None = None
|
||
self.last_error = ""
|
||
self.last_sentral = ""
|
||
self._thread: threading.Thread | None = None
|
||
|
||
def my_url(self, sentral_url: str) -> str:
|
||
if self.fixed_url:
|
||
return self.fixed_url
|
||
host = urllib.request.urlparse(sentral_url).hostname or "127.0.0.1"
|
||
return f"http://{local_ip_towards(host, self.port)}:{self.port}"
|
||
|
||
def beat_once(self) -> bool:
|
||
sentral = self.sentral()
|
||
if not sentral:
|
||
self.last_error = "ingen sentral kjent"
|
||
return False
|
||
try:
|
||
url = self.my_url(sentral)
|
||
except OSError as e:
|
||
self.last_error = f"finner ikke egen adresse mot {sentral}: {e}"
|
||
return False
|
||
ok, err = self.post(sentral, name=self.name, url=url, secret=self.secret, version=self.version)
|
||
if ok:
|
||
if self.last_sentral != sentral:
|
||
log.info("meldt hos sentralen %s som %s (%s)", sentral, self.name, url)
|
||
self.last_ok, self.last_error, self.last_sentral = time.time(), "", sentral
|
||
else:
|
||
if err != self.last_error:
|
||
log.warning("fikk ikke meldt meg hos %s: %s", sentral, err)
|
||
self.last_error = err
|
||
return ok
|
||
|
||
def status(self) -> dict:
|
||
return {"sentral": self.last_sentral or None, "last_ok": self.last_ok, "error": self.last_error}
|
||
|
||
def start(self) -> None:
|
||
if self._thread is None:
|
||
self._thread = threading.Thread(target=self._loop, daemon=True, name="heartbeat")
|
||
self._thread.start()
|
||
|
||
def _loop(self) -> None:
|
||
while True:
|
||
try:
|
||
self.beat_once()
|
||
except Exception: # noqa: BLE001
|
||
log.exception("hjerteslaget feilet")
|
||
# Raskere når vi ikke har fått kontakt ennå, så oppstart ikke tar et halvt minutt.
|
||
time.sleep(self.interval if self.last_ok else min(self.interval, 5.0))
|