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