| 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)) |
|
|
