Add TCP push server: broadcast new snapshots to connected clients

Each connected TCP client receives every snapshot as a newline-
delimited JSON line the moment it is collected (no polling needed).
The server binds on --host/--port (default 0.0.0.0:9999).
Client connections are accepted in a daemon thread; dead connections
are pruned on the next broadcast.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-25 19:26:05 +02:00
co-authored by Claude Sonnet 4.6
parent eaec158f2f
commit 81ebcd8e0f
+70
View File
@@ -27,6 +27,13 @@ Usage examples
Available domains: charging, climatisation, measurements, readiness, parking, access Available domains: charging, climatisation, measurements, readiness, parking, access
Push server
-----------
The collector listens on --host/--port (default 0.0.0.0:9999).
Each connected TCP client receives every new snapshot as a single
line of JSON (newline-delimited) the moment it is collected.
Connect with: nc localhost 9999 or any TCP client that reads lines.
Credentials file format (JSON) Credentials file format (JSON)
------------------------------- -------------------------------
{ {
@@ -81,7 +88,9 @@ import argparse
import json import json
import logging import logging
import os import os
import socket
import sys import sys
import threading
import time import time
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
@@ -366,6 +375,51 @@ def save_store(path: Path, store: dict, max_records: int | None) -> None:
path.write_text(json.dumps(store, indent=2, default=str)) path.write_text(json.dumps(store, indent=2, default=str))
# ── push server ──────────────────────────────────────────────────────────────
def _accept_loop(
server_sock: socket.socket,
clients: list,
lock: threading.Lock,
) -> None:
while True:
try:
conn, addr = server_sock.accept()
log.info("Push client connected from %s", addr)
with lock:
clients.append(conn)
except OSError:
break
def _broadcast(snapshot: dict, clients: list, lock: threading.Lock) -> None:
line = (json.dumps(snapshot, default=str) + "\n").encode()
with lock:
dead = []
for conn in clients:
try:
conn.sendall(line)
except OSError:
dead.append(conn)
for conn in dead:
clients.remove(conn)
conn.close()
log.info("Push client disconnected")
def _start_push_server(host: str, port: int) -> tuple:
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind((host, port))
server_sock.listen()
clients: list = []
lock = threading.Lock()
t = threading.Thread(target=_accept_loop, args=(server_sock, clients, lock), daemon=True)
t.start()
log.info("Push server listening on %s:%d", host, port)
return clients, lock
# ── main loop ──────────────────────────────────────────────────────────────── # ── main loop ────────────────────────────────────────────────────────────────
def _auto_out_path(logs_dir: Path, vin: str) -> Path: def _auto_out_path(logs_dir: Path, vin: str) -> Path:
@@ -424,6 +478,8 @@ def run(args: argparse.Namespace) -> None:
logs_dir.mkdir(exist_ok=True) logs_dir.mkdir(exist_ok=True)
out = _auto_out_path(logs_dir, vin) out = _auto_out_path(logs_dir, vin)
push_clients, push_lock = _start_push_server(args.host, args.port)
current_day = datetime.now(timezone.utc).date() current_day = datetime.now(timezone.utc).date()
store = load_store(out, vin, args.interval) store = load_store(out, vin, args.interval)
log.info( log.info(
@@ -448,6 +504,7 @@ def run(args: argparse.Namespace) -> None:
snapshot = collect_snapshot(vehicle, domains) snapshot = collect_snapshot(vehicle, domains)
store["records"].append(snapshot) store["records"].append(snapshot)
save_store(out, store, args.max_records) save_store(out, store, args.max_records)
_broadcast(snapshot, push_clients, push_lock)
log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"]) log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"])
except KeyboardInterrupt: except KeyboardInterrupt:
raise raise
@@ -521,6 +578,19 @@ def main() -> None:
metavar="N", metavar="N",
help="Keep only the last N records in the file (default: unlimited)", help="Keep only the last N records in the file (default: unlimited)",
) )
srv = p.add_argument_group("push server")
srv.add_argument(
"--host",
default="0.0.0.0",
help="Bind address for the push server (default: 0.0.0.0)",
)
srv.add_argument(
"--port",
type=int,
default=9999,
help="TCP port for the push server (default: 9999)",
)
p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging") p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
p.add_argument( p.add_argument(
"--list-domains", "--list-domains",