pyr0ball
7e14f9135e
FastAPI backend (port 8522):
- Session lifecycle: POST /session/start, DELETE /session/{id}/end, GET /session/{id}
- SSE stream: GET /session/{id}/stream — per-subscriber asyncio.Queue fan-out, 15s heartbeat
- History: GET /session/{id}/history with min_confidence + limit filters
- Audio: WS /session/{id}/audio — binary PCM ingestion stub (real inference in v0.2.x)
- Export: GET /session/{id}/export — downloadable JSON session log
- ContextClassifier background task per session (CF_VOICE_MOCK=1 in dev)
- ToneEvent SSE wire format per cf-core#40 (locked field names)
- Tier gate: CFG-LNNT- prefix check, 402 for paid features
Vue 3 frontend (port 8521, Vite + UnoCSS + Pinia):
- NowPanel: affect-aware border tint, subtext, prosody flags, shift indicator
- HistoryStrip: horizontal scroll, last 8 events with affect color
- ComposeBar: start/stop session, SSE connection lifecycle
- useToneStream: EventSource composable
- useAudioCapture: AudioWorklet → Int16 PCM → WebSocket (v0.1.x stub)
- audio-processor.js: 100ms chunk accumulator in AudioWorklet thread
- Respects prefers-reduced-motion globally
26 tests passing, manage.sh, Dockerfile, compose.yml included.
64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
# tests/test_tone_event.py
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.models.tone_event import ToneEvent
|
|
|
|
|
|
def _event(**kwargs) -> ToneEvent:
|
|
defaults = dict(
|
|
label="Neutral statement",
|
|
confidence=0.8,
|
|
speaker_id="speaker_a",
|
|
shift_magnitude=0.0,
|
|
timestamp=1000.0,
|
|
session_id="sess-1",
|
|
subtext="",
|
|
affect="neutral",
|
|
shift_direction="none",
|
|
prosody_flags=[],
|
|
elcor=False,
|
|
)
|
|
defaults.update(kwargs)
|
|
return ToneEvent(**defaults)
|
|
|
|
|
|
def test_to_sse_format():
|
|
evt = _event()
|
|
sse = evt.to_sse()
|
|
assert sse.startswith("event: tone-event\ndata: ")
|
|
assert sse.endswith("\n\n")
|
|
|
|
|
|
def test_to_sse_json_fields():
|
|
evt = _event(label="Frustration", affect="frustrated", confidence=0.9)
|
|
sse = evt.to_sse()
|
|
data_line = sse.split("data: ", 1)[1].rstrip()
|
|
payload = json.loads(data_line)
|
|
assert payload["label"] == "Frustration"
|
|
assert payload["affect"] == "frustrated"
|
|
assert payload["confidence"] == pytest.approx(0.9)
|
|
assert "timestamp" in payload
|
|
assert "session_id" in payload
|
|
|
|
|
|
def test_is_reliable_above_threshold():
|
|
assert _event(confidence=0.7).is_reliable()
|
|
|
|
|
|
def test_is_reliable_below_threshold():
|
|
assert not _event(confidence=0.5).is_reliable()
|
|
|
|
|
|
def test_is_reliable_custom_threshold():
|
|
assert _event(confidence=0.4).is_reliable(threshold=0.3)
|
|
assert not _event(confidence=0.4).is_reliable(threshold=0.5)
|
|
|
|
|
|
def test_elcor_subtext_in_sse():
|
|
evt = _event(elcor=True, subtext="[Neutral — matter-of-fact delivery]")
|
|
payload = json.loads(evt.to_sse().split("data: ", 1)[1].rstrip())
|
|
assert payload["subtext"] == "[Neutral — matter-of-fact delivery]"
|