"""Bounded metadata-only Speech transport traces; never audio or user text.""" from collections import OrderedDict from dataclasses import dataclass import hashlib import json import logging import time logger = logging.getLogger('uvicorn.error.device_gateway.speech') @dataclass class Trace: speech_id: str connection: str source: str started: float stage: str = 'prepare' stage_at: float = 0 pcm_bytes: int = 0 buffered: bool = False synth_ms: int = 0 sent_frames: int = 0 played_frames: int = 0 last_send: float | None = None last_progress: float | None = None last_log: float = 0 max_send_gap_ms: int = 0 max_send_ms: int = 0 max_lock_wait_ms: int = 0 max_write_ms: int = 0 max_progress_gap_ms: int = 0 credit_waits: int = 0 max_credit_wait_ms: int = 0 tx_finished: bool = False stop_seen: bool = False class SpeechDiagnostics: # Retain metadata after transmission ends, because codec ACKs arrive later. # Eviction affects diagnostics only, never playback/credits/cancellation. limit = 32 ttl = 300 def __init__(self, clock=time.monotonic): self.clock = clock self.traces = OrderedDict() def prune(self): now = self.clock() for key, trace in list(self.traces.items()): if now - trace.started > self.ttl: del self.traces[key] while len(self.traces) > self.limit: self.traces.popitem(last=False) def get(self, device, speech_id): self.prune() if not isinstance(speech_id, str): return None return self.traces.get((device.connection_id, speech_id)) def begin(self, device, req, source): now = self.clock() trace = Trace(req.speech_id, hashlib.sha256(str(device.connection_id).encode()).hexdigest()[:12], source if source in ('screen', 'reply', 'summary') else 'unknown', now, stage_at=now, last_log=now) self.traces[(device.connection_id, req.speech_id)] = trace self.prune() self.emit('start', trace) def emit(self, event, trace, **fields): # All fields are generated IDs, enums and numeric metadata from this module. value = dict(event=event, speech=trace.speech_id, connection=trace.connection, source=trace.source, stage=trace.stage, sent_frames=trace.sent_frames, played_frames=trace.played_frames, tx_finished=trace.tx_finished, **fields) logger.info('[SPEECH_DIAG] %s', json.dumps(value, separators=(',', ':'))) def stage(self, device, req, name): if trace := self.get(device, req.speech_id): trace.stage, trace.stage_at = name, self.clock() def pcm(self, device, req, size, buffered): if trace := self.get(device, req.speech_id): trace.pcm_bytes, trace.buffered = size, buffered trace.synth_ms = round((self.clock() - trace.stage_at) * 1000) self.emit('pcm_ready', trace, pcm_bytes=size, audio_ms=round(size / 88.2), synth_ms=trace.synth_ms, buffered=buffered) def sent(self, device, req, kind, started, timing): trace = self.get(device, req.speech_id) if not trace: return now = self.clock() send_ms = round((now - started) * 1000) trace.max_send_ms = max(trace.max_send_ms, send_ms) if isinstance(timing, tuple) and len(timing) == 2: trace.max_lock_wait_ms = max(trace.max_lock_wait_ms, round(timing[0] * 1000)) trace.max_write_ms = max(trace.max_write_ms, round(timing[1] * 1000)) gap_ms = 0 if kind == 'speech_chunk': trace.sent_frames += 1 if trace.last_send is not None: gap_ms = round((now - trace.last_send) * 1000) trace.max_send_gap_ms = max(trace.max_send_gap_ms, gap_ms) else: trace.last_progress = now trace.last_send = now if max(send_ms, gap_ms) >= 500 and now - trace.last_log >= 1: trace.last_log = now self.emit('send_delay', trace, send_ms=send_ms, gap_ms=gap_ms, max_lock_wait_ms=trace.max_lock_wait_ms, max_write_ms=trace.max_write_ms) def progress(self, device, event, ceiling=None): trace = self.get(device, event.get('speech_id')) count = event.get('played_chunks') if not trace or type(count) is not int or not trace.played_frames < count <= (trace.sent_frames if ceiling is None else ceiling): return now = self.clock() gap = round((now - trace.last_progress) * 1000) if trace.last_progress is not None else 0 trace.max_progress_gap_ms = max(trace.max_progress_gap_ms, gap) first = trace.played_frames == 0 trace.played_frames, trace.last_progress = count, now if first or now - trace.last_log >= 1: trace.last_log = now self.emit('codec_progress', trace, progress_gap_ms=gap) def credit_wait(self, device, req, started): if trace := self.get(device, req.speech_id): trace.credit_waits += 1 trace.max_credit_wait_ms = max(trace.max_credit_wait_ms, round((self.clock() - started) * 1000)) def finish(self, device, req, status, error='none'): if trace := self.get(device, req.speech_id): trace.tx_finished = status == 'sent' self.emit('tx_end', trace, status=status, error=error, **self.metrics(trace)) def metrics(self, trace): now = self.clock() return dict(pcm_bytes=trace.pcm_bytes, audio_ms=round(trace.pcm_bytes / 88.2), synth_ms=trace.synth_ms, total_ms=round((now - trace.started) * 1000), stage_ms=round((now - trace.stage_at) * 1000), max_send_gap_ms=trace.max_send_gap_ms, max_send_ms=trace.max_send_ms, max_lock_wait_ms=trace.max_lock_wait_ms, max_write_ms=trace.max_write_ms, max_progress_gap_ms=trace.max_progress_gap_ms, credit_waits=trace.credit_waits, max_credit_wait_ms=trace.max_credit_wait_ms, buffered=trace.buffered) def cancel(self, device, req, reason): if trace := self.get(device, req.speech_id): self.emit('cancel', trace, reason=reason if reason in ('requested', 'replacement', 'target_change', 'approval', 'voice', 'settings', 'mode_change', 'device_stop', 'disconnect', 'transport_failure', 'shutdown') else 'requested') def device_stop(self, device, speech_id): # A late stop after speech_end is visible without reopening active playback. self.prune() if speech_id is not None and not isinstance(speech_id, str): return matches = [t for (connection, sid), t in self.traces.items() if connection == device.connection_id and (speech_id is None or sid == speech_id)] if matches and not matches[-1].stop_seen: matches[-1].stop_seen = True self.emit('device_stop', matches[-1]) def forget(self, device): self.prune() for key, trace in list(self.traces.items()): if key[0] == device.connection_id: self.emit('disconnect', trace, **self.metrics(trace)) del self.traces[key]