| from __future__ import annotations
|
|
|
| import base64
|
| from dataclasses import dataclass, field
|
| from datetime import datetime
|
| import json
|
| from pathlib import Path
|
| import re
|
| from typing import Any
|
| from uuid import uuid4
|
|
|
| from time_machine.domain.events import (
|
| AudioEvent,
|
| ConversationTextEvent,
|
| DestinationEvent,
|
| ErrorEvent,
|
| MachineStateEvent,
|
| NarrationEvent,
|
| PersonaEvent,
|
| PortraitEvent,
|
| SceneEvent,
|
| SouvenirEvent,
|
| TranscriptEvent,
|
| TemporalEvent,
|
| VisualArtifactEvent,
|
| )
|
| from time_machine.domain.models import Destination, Persona, Souvenir
|
|
|
|
|
| def new_session_state() -> dict[str, Any]:
|
| return {
|
| "session_id": str(uuid4()),
|
| "encounter_id": None,
|
| "persona_name": "Contact",
|
| "machine_state": "dormant",
|
| "target_year": None,
|
| "destination_place": "",
|
| "destination_mode": "",
|
| "destination_atmosphere": "",
|
| "destination_visual_preset": "",
|
| "destination_motifs": [],
|
| "persona_role": "",
|
| "persona_situation": "",
|
| "scene": None,
|
| "portrait": None,
|
| "narration": None,
|
| "artifact": None,
|
| "image_generation_ready": False,
|
| "image_generation_warning": None,
|
| "ambient_key": "temporal_storm",
|
| "transcript_lines": [],
|
| "status": "Machine dormant.",
|
| "destination_md": "",
|
| "persona_md": "",
|
| "souvenir_md": "",
|
| "audio_path": None,
|
| "_partial_character_active": False,
|
| }
|
|
|
|
|
| @dataclass
|
| class UiRenderState:
|
| session_state: dict[str, Any]
|
| status: str = "Machine dormant."
|
| destination_md: str = ""
|
| persona_md: str = ""
|
| transcript_lines: list[str] = field(default_factory=list)
|
| souvenir_md: str = ""
|
| audio_path: str | None = None
|
|
|
| @classmethod
|
| def from_state(cls, session_state: dict[str, Any] | None) -> "UiRenderState":
|
| state = session_state or new_session_state()
|
| return cls(
|
| session_state=state,
|
| status=state.get("status", "Machine dormant."),
|
| destination_md=state.get("destination_md", ""),
|
| persona_md=state.get("persona_md", ""),
|
| transcript_lines=list(state.get("transcript_lines", [])),
|
| souvenir_md=state.get("souvenir_md", ""),
|
| audio_path=state.get("audio_path"),
|
| )
|
|
|
| @classmethod
|
| def reset_for_launch(cls, session_state: dict[str, Any] | None) -> "UiRenderState":
|
| state = session_state or new_session_state()
|
| state["encounter_id"] = None
|
| state["persona_name"] = "Contact"
|
| state["machine_state"] = "launching"
|
| state["target_year"] = None
|
| state["destination_place"] = ""
|
| state["destination_mode"] = ""
|
| state["destination_atmosphere"] = ""
|
| state["destination_visual_preset"] = ""
|
| state["destination_motifs"] = []
|
| state["persona_role"] = ""
|
| state["persona_situation"] = ""
|
| state["scene"] = None
|
| state["portrait"] = None
|
| state["narration"] = None
|
| state["artifact"] = None
|
| state["ambient_key"] = "temporal_storm"
|
| state["transcript_lines"] = []
|
| state["destination_md"] = ""
|
| state["persona_md"] = ""
|
| state["souvenir_md"] = ""
|
| state["audio_path"] = None
|
| state["_partial_character_active"] = False
|
| return cls.from_state(state)
|
|
|
| def apply(self, event: TemporalEvent) -> None:
|
| encounter_id = event.metadata.get("encounter_id")
|
| if encounter_id:
|
| self.session_state["encounter_id"] = encounter_id
|
|
|
| if isinstance(event, MachineStateEvent):
|
| self.session_state["machine_state"] = event.state
|
| self.status = event.message or event.state
|
| elif isinstance(event, DestinationEvent):
|
| self.destination_md = render_destination(event.destination)
|
| self._apply_destination(event.destination)
|
| self.session_state["machine_state"] = "destination"
|
| self.status = event.message or "Destination resolved."
|
| elif isinstance(event, PersonaEvent):
|
| self.persona_md = render_persona(event.persona)
|
| self.session_state["persona_name"] = event.persona.name
|
| self.session_state["persona_role"] = event.persona.role
|
| self.session_state["persona_situation"] = event.persona.immediate_situation
|
| self.session_state["machine_state"] = "persona"
|
| self.status = event.message or "Signal locked."
|
| elif isinstance(event, TranscriptEvent):
|
| self.transcript_lines.append(f"**You:** {event.transcript.text}")
|
| self.status = event.message or "Transcript decoded."
|
| elif isinstance(event, ConversationTextEvent):
|
| name = self.session_state.get("persona_name", "Contact")
|
| line = f"**{name}:** {event.text}"
|
| if event.is_final:
|
| if self.session_state.get("_partial_character_active") and self.transcript_lines:
|
| self.transcript_lines[-1] = line
|
| else:
|
| self.transcript_lines.append(line)
|
| self.session_state["_partial_character_active"] = False
|
| elif self.session_state.get("_partial_character_active") and self.transcript_lines:
|
| self.transcript_lines[-1] = line
|
| else:
|
| self.transcript_lines.append(line)
|
| self.session_state["_partial_character_active"] = True
|
| self.status = event.message or "Character replied."
|
| elif isinstance(event, AudioEvent):
|
| self.audio_path = event.audio.path
|
| self.status = event.message or event.audio.description or "Audio ready."
|
| elif isinstance(event, SceneEvent):
|
| self.session_state["scene"] = event.scene.model_dump(mode="json")
|
| self.session_state["ambient_key"] = event.scene.ambient_key
|
| self.status = event.message or "World image stabilized."
|
| elif isinstance(event, PortraitEvent):
|
| self.session_state["portrait"] = event.portrait.model_dump(mode="json")
|
| self.status = event.message or "Character image resolved."
|
| elif isinstance(event, NarrationEvent):
|
| self.session_state["narration"] = event.narration.model_dump(mode="json")
|
| self.status = event.message or "Narrator signal prepared."
|
| elif isinstance(event, VisualArtifactEvent):
|
| self.session_state["artifact"] = event.artifact.model_dump(mode="json")
|
| self.status = event.message or "Temporal artifact materialized."
|
| elif isinstance(event, SouvenirEvent):
|
| self.souvenir_md = render_souvenir(event.souvenir)
|
| self.status = event.message or "Souvenir ready."
|
| elif isinstance(event, ErrorEvent):
|
| self.status = event.message or "Recoverable error."
|
|
|
| timing_summary = render_timing_summary(event.metadata)
|
| if timing_summary:
|
| self.status = f"{self.status} ({timing_summary})"
|
|
|
| self._sync_state()
|
|
|
| def set_error(self, message: str) -> None:
|
| self.status = message
|
| self._sync_state()
|
|
|
| def to_outputs(self) -> tuple:
|
| self._sync_state()
|
| return (
|
| self.session_state,
|
| self.status,
|
| self.destination_md,
|
| self.persona_md,
|
| "\n\n".join(self.transcript_lines),
|
| self.souvenir_md,
|
| self.audio_path,
|
| render_realtime_config(self.session_state),
|
| render_immersive_payload(self.session_state),
|
| )
|
|
|
| def _apply_destination(self, destination: Destination) -> None:
|
| self.session_state["target_year"] = _parse_year(destination.year)
|
| self.session_state["destination_place"] = destination.place
|
| self.session_state["destination_mode"] = destination.mode
|
| self.session_state["destination_atmosphere"] = destination.atmosphere
|
| self.session_state["destination_visual_preset"] = destination.visual_preset_key
|
| self.session_state["destination_motifs"] = [motif.label for motif in destination.visual_motifs]
|
|
|
| def _sync_state(self) -> None:
|
| self.session_state["status"] = self.status
|
| self.session_state["destination_md"] = self.destination_md
|
| self.session_state["persona_md"] = self.persona_md
|
| self.session_state["transcript_lines"] = list(self.transcript_lines)
|
| self.session_state["souvenir_md"] = self.souvenir_md
|
| self.session_state["audio_path"] = self.audio_path
|
|
|
|
|
| def render_destination(destination: Destination) -> str:
|
| motifs = ", ".join(motif.label for motif in destination.visual_motifs) or "none"
|
| oddity = destination.temporal_oddity or "No visible anomaly."
|
| return (
|
| f"### {destination.place}, {destination.year}\n"
|
| f"**Mode:** {destination.mode}\n\n"
|
| f"{destination.atmosphere}\n\n"
|
| f"**Visual preset:** `{destination.visual_preset_key}`\n\n"
|
| f"**Motifs:** {motifs}\n\n"
|
| f"**Oddity:** {oddity}"
|
| )
|
|
|
|
|
| def render_persona(persona: Persona) -> str:
|
| constraints = "\n".join(f"- {item}" for item in persona.worldview_constraints)
|
| return (
|
| f"### {persona.name}\n"
|
| f"**Role:** {persona.role}\n\n"
|
| f"**Situation:** {persona.immediate_situation}\n\n"
|
| f"**Concern:** {persona.daily_concern}\n\n"
|
| f"**Theory about you:** {persona.voice_theory}\n\n"
|
| f"**Speaking style:** {persona.speaking_style}\n\n"
|
| f"**Worldview constraints:**\n{constraints}"
|
| )
|
|
|
|
|
| def render_souvenir(souvenir: Souvenir) -> str:
|
| return (
|
| f"### {souvenir.stamp_name}\n"
|
| f"**Destination:** {souvenir.destination}\n\n"
|
| f"**Contact:** {souvenir.contact}\n\n"
|
| f"**Quote:** \"{souvenir.quote}\"\n\n"
|
| f"**Artifact:** {souvenir.artifact}\n\n"
|
| f"{souvenir.summary}"
|
| )
|
|
|
|
|
| def render_timing_summary(metadata: dict[str, Any]) -> str:
|
| timings = metadata.get("timings")
|
| if not isinstance(timings, list):
|
| return ""
|
|
|
| parts: list[str] = []
|
| for timing in timings:
|
| if not isinstance(timing, dict):
|
| continue
|
| stage = timing.get("stage")
|
| elapsed_ms = timing.get("elapsed_ms")
|
| if not isinstance(stage, str) or not isinstance(elapsed_ms, int | float):
|
| continue
|
| parts.append(f"{stage}: {elapsed_ms / 1000:.1f}s")
|
| return ", ".join(parts)
|
|
|
|
|
| def render_realtime_config(session_state: dict[str, Any]) -> str:
|
| return json.dumps(
|
| {
|
| "session_id": session_state.get("session_id"),
|
| "encounter_id": session_state.get("encounter_id"),
|
| "persona_name": session_state.get("persona_name", "Contact"),
|
| "ready": bool(session_state.get("encounter_id")),
|
| "path": "/realtime/voice",
|
| },
|
| ensure_ascii=True,
|
| )
|
|
|
|
|
| def render_immersive_payload(session_state: dict[str, Any]) -> str:
|
| current_year = datetime.now().year
|
| target_year = session_state.get("target_year")
|
| direction = "hold"
|
| if isinstance(target_year, int):
|
| direction = "rewind" if target_year < current_year else "fast_forward" if target_year > current_year else "hold"
|
|
|
| return json.dumps(
|
| {
|
| "session_id": session_state.get("session_id"),
|
| "encounter_id": session_state.get("encounter_id"),
|
| "machine_state": session_state.get("machine_state", "dormant"),
|
| "current_year": current_year,
|
| "target_year": target_year,
|
| "direction": direction,
|
| "destination": {
|
| "place": session_state.get("destination_place", ""),
|
| "mode": session_state.get("destination_mode", ""),
|
| "atmosphere": session_state.get("destination_atmosphere", ""),
|
| "visual_preset": session_state.get("destination_visual_preset", ""),
|
| "motifs": session_state.get("destination_motifs", []),
|
| },
|
| "persona": {
|
| "name": session_state.get("persona_name", "Contact"),
|
| "role": session_state.get("persona_role", ""),
|
| "situation": session_state.get("persona_situation", ""),
|
| },
|
| "scene": session_state.get("scene"),
|
| "portrait": session_state.get("portrait"),
|
| "narration": _render_narration_payload(session_state.get("narration")),
|
| "artifact": session_state.get("artifact"),
|
| "image_generation": _render_image_generation_payload(session_state),
|
| "ambient_key": session_state.get("ambient_key", "temporal_storm"),
|
| },
|
| ensure_ascii=True,
|
| )
|
|
|
|
|
| def _render_image_generation_payload(session_state: dict[str, Any]) -> dict[str, Any]:
|
| scene = session_state.get("scene")
|
| portrait = session_state.get("portrait")
|
| return {
|
| "ready": bool(session_state.get("image_generation_ready")),
|
| "warning": session_state.get("image_generation_warning") or "",
|
| "scene_image_ready": _has_png_image(scene),
|
| "portrait_image_ready": _has_png_image(portrait),
|
| }
|
|
|
|
|
| def _has_png_image(value: object) -> bool:
|
| if not isinstance(value, dict):
|
| return False
|
| return bool(value.get("image_b64")) and value.get("mime_type") == "image/png"
|
|
|
|
|
| def _render_narration_payload(value: object) -> dict[str, Any] | None:
|
| if not isinstance(value, dict):
|
| return None
|
| audio = value.get("audio")
|
| audio_path = None
|
| mime_type = "audio/wav"
|
| if isinstance(audio, dict):
|
| mime_type = str(audio.get("mime_type") or mime_type)
|
| raw_path = audio.get("path")
|
| if isinstance(raw_path, str) and raw_path:
|
| audio_path = _audio_data_url(raw_path, mime_type)
|
| return {
|
| "text": value.get("text", ""),
|
| "audio_path": audio_path,
|
| "mime_type": mime_type,
|
| }
|
|
|
|
|
| def _audio_data_url(path_value: str, mime_type: str) -> str | None:
|
| path = Path(path_value)
|
| if not path.exists():
|
| return None
|
| encoded = base64.b64encode(path.read_bytes()).decode("ascii")
|
| return f"data:{mime_type};base64,{encoded}"
|
|
|
|
|
| def _parse_year(value: str) -> int | None:
|
| match = re.search(r"-?\d{1,5}", value)
|
| if not match:
|
| return None
|
| return int(match.group(0))
|
|
|
