From 81ebcd8e0f0ec663a5910367ca21d91069412050 Mon Sep 17 00:00:00 2001
From: Jens Ahrensfeld <jens@jayfield.org>
Date: Mon, 25 May 2026 19:26:05 +0200
Subject: [PATCH] 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>
---
 collect.py | 70 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 70 insertions(+)

diff --git a/collect.py b/collect.py
index 6cedc33..10af9de 100644
--- a/collect.py
+++ b/collect.py
@@ -27,6 +27,13 @@ Usage examples
 
 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)
 -------------------------------
 {
@@ -81,7 +88,9 @@ import argparse
 import json
 import logging
 import os
+import socket
 import sys
+import threading
 import time
 from datetime import datetime, timezone
 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))
 
 
+# ── 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 ────────────────────────────────────────────────────────────────
 
 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)
         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()
     store = load_store(out, vin, args.interval)
     log.info(
@@ -448,6 +504,7 @@ def run(args: argparse.Namespace) -> None:
                 snapshot = collect_snapshot(vehicle, domains)
                 store["records"].append(snapshot)
                 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"])
             except KeyboardInterrupt:
                 raise
@@ -521,6 +578,19 @@ def main() -> None:
         metavar="N",
         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(
         "--list-domains",