b8950b38a4
Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config.
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""Latency logging — see package docstring (REQ-011).
|
|
|
|
``LatencyLogger`` lives here so ``session`` imports ``..metrics.latency``
|
|
stably; the package ``__init__`` re-exports it for convenience.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
|
|
class LatencyLogger:
|
|
def __init__(self) -> None:
|
|
self.t: dict[str, float] = {}
|
|
|
|
def _stamp(self, key: str) -> None:
|
|
if key not in self.t:
|
|
self.t[key] = time.monotonic()
|
|
|
|
def speech_start(self) -> None:
|
|
self._stamp("speech_start")
|
|
|
|
def speech_end(self) -> None:
|
|
self._stamp("speech_end")
|
|
|
|
def stt_first_partial(self) -> None:
|
|
self._stamp("stt_first_partial")
|
|
|
|
def stt_final(self) -> None:
|
|
self._stamp("stt_final")
|
|
|
|
def hermes_request(self) -> None:
|
|
self._stamp("hermes_request")
|
|
|
|
def qwen_first_token(self) -> None:
|
|
self._stamp("qwen_first_token")
|
|
|
|
def first_audio_packet(self) -> None:
|
|
self._stamp("first_audio_packet")
|
|
|
|
def playback_start(self) -> None:
|
|
self._stamp("playback_start")
|
|
|
|
def eos_to_first_audio(self) -> float | None:
|
|
"""End-of-speech → first audibly-played audio (seconds)."""
|
|
start = self.t.get("speech_end")
|
|
end = self.t.get("playback_start", self.t.get("first_audio_packet"))
|
|
if start is None or end is None:
|
|
return None
|
|
return end - start
|
|
|
|
def eos_to_first_token(self) -> float | None:
|
|
start = self.t.get("speech_end")
|
|
end = self.t.get("qwen_first_token")
|
|
if start is None or end is None:
|
|
return None
|
|
return end - start
|
|
|
|
def reset(self) -> None:
|
|
self.t = {}
|