Claude (homelab agent)andClaude Opus 5.5
56eb5af0ff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""Live push (SSE).
|
|
|
|
One watcher thread polls the classifier indexes (.cls.json, rewritten atomically) of the
|
|
two newest day dirs every second and appends newly classified recordings to a ring buffer.
|
|
SSE handlers block on a Condition and stream new entries. Clients are capped.
|
|
"""
|
|
import json
|
|
import threading
|
|
import os
|
|
import time
|
|
|
|
from radio.config import MAX_SSE, REC_DIR
|
|
from radio.recordings import cls_index, list_days, make_item
|
|
from web.status import status
|
|
|
|
_ev = {"seq": 0, "buf": [], "known": {}, "mt": {}, "clients": 0}
|
|
_ev_cond = threading.Condition()
|
|
|
|
|
|
def watch():
|
|
first = True
|
|
while True:
|
|
try:
|
|
new = []
|
|
for day in list_days()[:2]:
|
|
p = os.path.join(REC_DIR, day, ".cls.json")
|
|
try:
|
|
mt = os.stat(p).st_mtime
|
|
except OSError:
|
|
continue
|
|
if _ev["mt"].get(day) == mt:
|
|
continue
|
|
_ev["mt"][day] = mt
|
|
idx = cls_index(day)
|
|
known = _ev["known"].setdefault(day, set())
|
|
for n in sorted(set(idx) - known):
|
|
known.add(n)
|
|
if first:
|
|
continue
|
|
ci = idx[n]
|
|
sub = "" if ci.get("c") == "voice" else "noise/"
|
|
it = make_item(day, n, sub, ci.get("c"), ci)
|
|
if it:
|
|
new.append(it)
|
|
for d in list(_ev["known"]):
|
|
if d not in list_days()[:2]:
|
|
_ev["known"].pop(d, None)
|
|
_ev["mt"].pop(d, None)
|
|
first = False
|
|
if new:
|
|
with _ev_cond:
|
|
for it in new:
|
|
_ev["seq"] += 1
|
|
_ev["buf"].append((_ev["seq"], it))
|
|
del _ev["buf"][:-300]
|
|
_ev_cond.notify_all()
|
|
except Exception: # never let the watcher die
|
|
pass
|
|
time.sleep(1.0)
|
|
|
|
|
|
def stream(h):
|
|
"""Server-Sent Events on handler H: `event: rec` per newly classified recording, `event: status` every 15 s."""
|
|
with _ev_cond:
|
|
if _ev["clients"] >= MAX_SSE:
|
|
return h.send(503, "too many listeners", "text/plain", {"Retry-After": "30"})
|
|
_ev["clients"] += 1
|
|
try:
|
|
try:
|
|
last = int(h.headers.get("Last-Event-ID") or -1)
|
|
except ValueError:
|
|
last = -1
|
|
with _ev_cond:
|
|
if last < 0 or last > _ev["seq"]:
|
|
last = _ev["seq"]
|
|
h.close_connection = True
|
|
h.send_response(200)
|
|
h.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
|
h.send_header("Cache-Control", "no-cache")
|
|
h.send_header("X-Accel-Buffering", "no")
|
|
h.send_header("Connection", "close")
|
|
h.end_headers()
|
|
h.wfile.write(b"retry: 3000\n\n")
|
|
h.wfile.flush()
|
|
next_status = 0
|
|
while True:
|
|
with _ev_cond:
|
|
pend = [(q, it) for q, it in _ev["buf"] if q > last]
|
|
if not pend:
|
|
_ev_cond.wait(timeout=max(0.5, next_status - time.time()))
|
|
pend = [(q, it) for q, it in _ev["buf"] if q > last]
|
|
out = []
|
|
for q, it in pend:
|
|
last = q
|
|
out.append(f"id: {q}\nevent: rec\ndata: {json.dumps(it, ensure_ascii=False)}\n\n")
|
|
if time.time() >= next_status:
|
|
next_status = time.time() + 15
|
|
out.append(f"event: status\ndata: {json.dumps(status(), ensure_ascii=False)}\n\n")
|
|
if out:
|
|
h.wfile.write("".join(out).encode("utf-8"))
|
|
h.wfile.flush()
|
|
except (BrokenPipeError, ConnectionResetError, TimeoutError, OSError):
|
|
pass
|
|
finally:
|
|
with _ev_cond:
|
|
_ev["clients"] -= 1
|