"""
Audio-piece timing tracer  (ADD-ON, diagnostic only — safe to remove).

Records, per call, two timestamps for every audio piece:

    Alibaba cloud  --recv-->  THIS server  --send-->  bridge  ----->  caller
                   ^ recv gap               ^ send gap        ^ (bridge's own log)

  * record_recv() is called the moment an audio delta ARRIVES from the cloud
    model (in callbacks.py, response.audio.delta).
  * record_send() is called the moment that server SENDS a piece out to the
    phone bridge (in qwen_pipeline_server.py, _sender()).

Comparing the two gap timelines (plus the bridge's own AudioTrace receive log)
tells you WHERE a speech "freeze" is introduced:

  * a big RECV gap  ............... the cloud paused generating (Alibaba side)
  * RECV smooth but SEND gappy .... this server delayed relaying it
                                    (e.g. the event loop was blocked)
  * RECV + SEND both smooth, but
    the bridge still saw a gap ..... the network between here and the bridge

This module is a pure add-on. It touches no call logic; every public function
swallows its own errors and returns fast, so it can never disturb a live call.
View it at  /audio_timing.html  on this server's dashboard.
"""

import threading
import time
from collections import deque, OrderedDict

# Same gap size the bridge flags as a "WSS gap", so the two views line up.
GAP_THRESHOLD_MS = 250

_LOCK = threading.Lock()
_MAX_CALLS = 40           # keep only the most recent N calls in memory
_MAX_EVENTS = 30000       # hard safety cap per stream, per call

# call_id -> {started_wall, started_mono, recv: deque[(t_ms, bytes)], send: deque}
_CALLS = OrderedDict()


def _bucket(call_id):
    b = _CALLS.get(call_id)
    if b is None:
        b = {
            'started_wall': time.time(),
            'started_mono': time.monotonic(),
            'recv': deque(maxlen=_MAX_EVENTS),
            'send': deque(maxlen=_MAX_EVENTS),
        }
        _CALLS[call_id] = b
        while len(_CALLS) > _MAX_CALLS:
            _CALLS.popitem(last=False)   # drop the oldest call
    return b


def _rel_ms(b):
    return (time.monotonic() - b['started_mono']) * 1000.0


def record_recv(call_id, nbytes=0):
    """An audio piece arrived from the cloud model."""
    try:
        if not call_id:
            return
        with _LOCK:
            b = _bucket(call_id)
            b['recv'].append((_rel_ms(b), int(nbytes)))
    except Exception:
        pass


def record_send(call_id, nbytes=0):
    """An audio piece was sent out to the phone bridge."""
    try:
        if not call_id:
            return
        with _LOCK:
            b = _bucket(call_id)
            b['send'].append((_rel_ms(b), int(nbytes)))
    except Exception:
        pass


def _gaps(events):
    out, prev = [], None
    for t, _ in events:
        if prev is not None:
            out.append(t - prev)
        prev = t
    return out


def _summary(events):
    gaps = _gaps(events)
    big = [g for g in gaps if g >= GAP_THRESHOLD_MS]
    return {
        'pieces': len(events),
        'max_gap_ms': round(max(gaps), 1) if gaps else 0.0,
        'gaps_over_thresh': len(big),
        'span_s': round(events[-1][0] / 1000.0, 1) if events else 0.0,
    }


def _verdict(recv_summary, send_summary):
    """Best-guess of where the worst gap came from (see module docstring)."""
    r = recv_summary['max_gap_ms']
    s = send_summary['max_gap_ms']
    if r < GAP_THRESHOLD_MS and s < GAP_THRESHOLD_MS:
        return 'clean'
    # Gap already present when the audio arrived from the cloud.
    if r >= GAP_THRESHOLD_MS and r >= s * 0.7:
        return 'cloud'
    # Cloud delivered smoothly but this server was slow to relay it.
    if s >= GAP_THRESHOLD_MS and s > r * 1.5:
        return 'this-server'
    return 'mixed'


def get_calls():
    """Summary row per call, newest first."""
    try:
        with _LOCK:
            items = [(cid, list(b['recv']), list(b['send']), b['started_wall'])
                     for cid, b in _CALLS.items()]
        out = []
        for cid, recv, send, started in items:
            rs = _summary(recv)
            ss = _summary(send)
            out.append({
                'call_id': cid,
                'started': time.strftime('%Y-%m-%d %H:%M:%S',
                                         time.localtime(started)),
                'recv': rs,
                'send': ss,
                'verdict': _verdict(rs, ss),
            })
        out.sort(key=lambda c: c['call_id'], reverse=True)
        return out
    except Exception:
        return []


def get_series(call_id):
    """Full per-piece timeline for one call (recv + send streams)."""
    try:
        with _LOCK:
            b = _CALLS.get(call_id)
            if not b:
                return None
            recv = list(b['recv'])
            send = list(b['send'])
            started = b['started_wall']

        def rows(events):
            out, prev = [], None
            for t, nb in events:
                gap = (t - prev) if prev is not None else 0.0
                out.append({'t_ms': round(t, 1),
                            'gap_ms': round(gap, 1),
                            'bytes': nb})
                prev = t
            return out

        rs = _summary(recv)
        ss = _summary(send)
        return {
            'call_id': call_id,
            'started': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(started)),
            'threshold_ms': GAP_THRESHOLD_MS,
            'verdict': _verdict(rs, ss),
            'recv_summary': rs,
            'send_summary': ss,
            'recv': rows(recv),
            'send': rows(send),
        }
    except Exception:
        return None