You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 

122 lines
6.1 KiB

"""UTF-8 BOM CSV logging for Phase 5 voice prompts and responses."""
from __future__ import annotations
import csv
import json
from datetime import datetime
from pathlib import Path
from typing import TextIO
from .config_loader import VoiceLoggingSettings
from .exceptions import VoiceLogError
from .models import VoicePromptDecision, VoicePromptEvent, VoicePromptTrackState
class VoiceLogger:
"""Write voice decisions, events, and one summary per track."""
DECISION_FIELDS = ["experiment_id", "frame_number", "timestamp", "elapsed_time_sec", "track_id", "mode",
"eligible", "should_prompt", "prompt_played", "pseudo_prompt", "reason", "audio_file"]
EVENT_FIELDS = ["experiment_id", "event_type", "track_id", "frame_number", "timestamp", "elapsed_time_sec",
"mode", "prompt_played", "pseudo_prompt", "audio_file", "turn_confirmed", "turn_level",
"reaction_time_sec", "metadata_json"]
SUMMARY_FIELDS = ["experiment_id", "track_id", "mode", "eligible", "eligibility_frame", "eligibility_time",
"prompt_attempted", "prompt_played", "prompt_failed", "pseudo_prompt", "prompt_frame",
"prompt_time", "prompt_elapsed_time_sec", "audio_file", "response_window_started",
"response_window_start_frame", "response_window_start_time", "response_window_end_time_sec",
"response_detected", "response_frame", "response_time", "reaction_time_sec",
"response_turn_level", "max_yaw_delta_after_prompt", "final_turn_level",
"valid_for_voice_analysis", "exclusion_reason"]
def __init__(self, config: VoiceLoggingSettings, experiment_id: str, run_stamp: str | None = None) -> None:
self.config, self.experiment_id = config, experiment_id
self.run_stamp = run_stamp or datetime.now().strftime("%Y%m%d_%H%M%S")
self.directory = Path(config.directory)
self._files: dict[str, TextIO] = {}
self._writers: dict[str, csv.DictWriter[str]] = {}
self._summary_ids: set[int] = set()
self._frames = 0
def open(self) -> None:
if not self.config.enabled:
return
self.directory.mkdir(parents=True, exist_ok=True)
try:
self._open("decisions", "voice_decisions", self.DECISION_FIELDS, self.config.log_decisions)
self._open("events", "voice_events", self.EVENT_FIELDS, self.config.log_events)
self._open("summary", "voice_summary", self.SUMMARY_FIELDS, self.config.log_summaries)
except OSError as exc:
self.close()
raise VoiceLogError(f"Failed to open voice logs: {exc}") from exc
def _open(self, key: str, suffix: str, fields: list[str], enabled: bool) -> None:
if not enabled:
return
stream = (self.directory / f"{self.run_stamp}_{self.experiment_id}_{suffix}.csv").open(
"w", encoding="utf-8-sig", newline="",
)
writer = csv.DictWriter(stream, fieldnames=fields)
writer.writeheader()
self._files[key], self._writers[key] = stream, writer
def write_decisions(self, items: list[VoicePromptDecision]) -> None:
writer = self._writers.get("decisions")
if writer is not None:
for item in items:
writer.writerow({"experiment_id": self.experiment_id, "frame_number": item.frame_number,
"timestamp": item.timestamp_iso, "elapsed_time_sec": self._number(item.elapsed_time_sec),
"track_id": item.track_id, "mode": item.mode, "eligible": item.eligible,
"should_prompt": item.should_prompt, "prompt_played": item.prompt_played,
"pseudo_prompt": item.pseudo_prompt, "reason": item.reason,
"audio_file": item.audio_file or ""})
self._frames += 1
if self._frames >= self.config.flush_interval_frames:
self.flush()
def write_events(self, items: list[VoicePromptEvent]) -> None:
writer = self._writers.get("events")
if writer is None:
return
for item in items:
writer.writerow({"experiment_id": self.experiment_id, "event_type": item.event_type,
"track_id": item.track_id, "frame_number": item.frame_number,
"timestamp": item.timestamp_iso, "elapsed_time_sec": self._number(item.elapsed_time_sec),
"mode": item.mode, "prompt_played": item.prompt_played,
"pseudo_prompt": item.pseudo_prompt, "audio_file": item.audio_file or "",
"turn_confirmed": item.turn_confirmed, "turn_level": item.turn_level or "",
"reaction_time_sec": self._number(item.reaction_time_sec),
"metadata_json": json.dumps(item.metadata, ensure_ascii=False)})
def write_summary(self, state: VoicePromptTrackState) -> None:
writer = self._writers.get("summary")
if writer is None or state.track_id in self._summary_ids:
return
row = {field: getattr(state, field, "") for field in self.SUMMARY_FIELDS if field != "experiment_id"}
row["experiment_id"] = self.experiment_id
for field in ("prompt_elapsed_time_sec", "response_window_end_time_sec", "reaction_time_sec", "max_yaw_delta_after_prompt"):
row[field] = self._number(getattr(state, field))
writer.writerow(row)
self._summary_ids.add(state.track_id)
@staticmethod
def _number(value: float | None) -> str:
return "" if value is None else f"{value:.6f}"
def flush(self) -> None:
for stream in self._files.values():
stream.flush()
self._frames = 0
def close(self) -> None:
for stream in self._files.values():
stream.close()
self._files.clear()
self._writers.clear()
def __enter__(self) -> "VoiceLogger":
self.open()
return self
def __exit__(self, *_args: object) -> None:
self.close()