File size: 1,995 Bytes
6bcddd0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | import json
from collections.abc import Iterable, Iterator
from typing import Any
VALID_TYPES = {
"stage",
"voice",
"skill",
"text_delta",
"sentence_end",
"vision_note",
"audio",
"debug",
"done",
"error",
}
def normalize_event(event: dict[str, Any]) -> dict[str, Any]:
event_type = event.get("type")
if event_type not in VALID_TYPES:
return {"type": "debug", "message": f"unknown event type: {event_type}", "raw": event}
if event_type == "stage":
return {
"type": "stage",
"expression": event.get("expression", "idle"),
"motion": event.get("motion", "breathe"),
"intensity": _clamp(event.get("intensity", 0.5), 0.0, 1.0),
}
if event_type == "voice":
return {
"type": "voice",
"style": event.get("style", "neutral"),
"speed": _clamp(event.get("speed", 1.0), 0.6, 1.4),
"energy": _clamp(event.get("energy", 0.5), 0.0, 1.0),
}
if event_type == "text_delta":
return {"type": "text_delta", "text": str(event.get("text", ""))}
return event
def parse_sse_lines(lines: Iterable[str]) -> Iterator[dict[str, Any]]:
for line in lines:
if not line:
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
yield {"type": "done"}
continue
try:
parsed = json.loads(line)
except json.JSONDecodeError:
yield {"type": "debug", "message": "invalid json event", "raw": line}
continue
yield normalize_event(parsed)
def dumps_event(event: dict[str, Any]) -> str:
return json.dumps(normalize_event(event), ensure_ascii=False)
def _clamp(value: Any, low: float, high: float) -> float:
try:
number = float(value)
except (TypeError, ValueError):
number = (low + high) / 2
return max(low, min(high, number))
|