"""Fixed-column Phase 7 unified CSV writer.""" from __future__ import annotations import csv, json from datetime import datetime from pathlib import Path from typing import TextIO from .config_loader import UnifiedLogSettings from .exceptions import UnifiedLogError from .models import LogConsistencyIssue, PersonAnalysisRecord, SessionSummary, UnifiedEvent class UnifiedEventLogger: """Write unified events and deduplicated Phase 8 input tables.""" EVENT_FIELDS = ["experiment_id","session_id","event_id","source","event_type","track_id","frame_number","timestamp","elapsed_time_sec","person_status","voice_mode","prompt_played","pseudo_prompt","turn_confirmed","turn_level","response_detected","reaction_time_sec","exclusion_reason","severity","metadata_json"] ANALYSIS_FIELDS = ["experiment_id","session_id","track_id","voice_mode","prompt_condition","prompt_played","pseudo_prompt","prompt_time","prompt_elapsed_time_sec","response_window_started","response_detected","reaction_time_sec","response_turn_level","turn_confirmed","turn_level","max_yaw_delta_toward_signage","face_detected_ever","pose_estimated_ever","face_detection_rate","pose_estimation_rate","valid_for_turn_analysis","valid_for_voice_analysis","final_status","completed","excluded","exclusion_reason","track_duration_sec","trigger_crossed","trigger_crossing_direction","movement_direction","camera_position_note","notes"] SESSION_FIELDS = ["experiment_id","session_id","started_at","ended_at","duration_sec","voice_mode","total_tracks","completed_tracks","excluded_tracks","evaluable_turn_tracks","valid_voice_analysis_tracks","prompt_played_count","pseudo_prompt_count","response_detected_count","no_response_count","not_evaluable_count","response_rate","mean_reaction_time_sec","median_reaction_time_sec","weak_or_higher_count","subtle_count","medium_or_higher_count","strong_count","average_fps","camera_position_note","warnings_count","errors_count","notes"] ISSUE_FIELDS = ["experiment_id","session_id","issue_type","severity","track_id","frame_number","message","metadata_json"] def __init__(self, config: UnifiedLogSettings, experiment_id: str, session_id: str, run_stamp: str | None = None) -> None: self.config,self.experiment_id,self.session_id=config,experiment_id,session_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._event_ids:set[str]=set(); self._analysis_ids:set[int]=set(); self._session_written=False; self._frames=0 def open(self) -> None: if not self.config.enabled:return self.directory.mkdir(parents=True,exist_ok=True) try: self._open("events","unified_events",self.EVENT_FIELDS,self.config.log_unified_events) self._open("analysis","person_analysis_table",self.ANALYSIS_FIELDS,self.config.log_person_analysis_table) self._open("session","session_summary",self.SESSION_FIELDS,self.config.log_session_summary) self._open("issues","log_consistency_issues",self.ISSUE_FIELDS,self.config.log_consistency_issues) except OSError as exc:self.close();raise UnifiedLogError(str(exc)) from exc def _open(self,key:str,suffix:str,fields:list[str],enabled:bool)->None: if not enabled:return f=(self.directory/f"{self.run_stamp}_{self.experiment_id}_{suffix}.csv").open("w",encoding="utf-8-sig",newline="");w=csv.DictWriter(f,fieldnames=fields);w.writeheader();self._files[key]=f;self._writers[key]=w def write_events(self,events:list[UnifiedEvent])->None: w=self._writers.get("events") if w: for e in sorted(events,key=lambda x:(x.elapsed_time_sec,x.frame_number,x.source,x.event_id)): if e.event_id in self._event_ids:continue row={k:getattr(e,k,"") for k in self.EVENT_FIELDS if k not in {"timestamp","metadata_json"}};row["timestamp"]=e.timestamp_iso;row["metadata_json"]=json.dumps(e.metadata,ensure_ascii=False);w.writerow(row);self._event_ids.add(e.event_id) self._frames+=1 if self._frames>=self.config.flush_interval_frames:self.flush() def write_analysis(self,items:list[PersonAnalysisRecord])->None: w=self._writers.get("analysis") if not w:return for x in items: if x.track_id in self._analysis_ids:continue w.writerow({k:getattr(x,k) for k in self.ANALYSIS_FIELDS});self._analysis_ids.add(x.track_id) def write_session(self,item:SessionSummary)->None: w=self._writers.get("session") if w and not self._session_written:w.writerow({k:getattr(item,k) for k in self.SESSION_FIELDS});self._session_written=True def write_issues(self,items:list[LogConsistencyIssue])->None: w=self._writers.get("issues") if not w:return for x in items:w.writerow({"experiment_id":self.experiment_id,"session_id":self.session_id,"issue_type":x.issue_type,"severity":x.severity,"track_id":x.track_id,"frame_number":x.frame_number,"message":x.message,"metadata_json":json.dumps(x.metadata,ensure_ascii=False)}) def flush(self)->None: for f in self._files.values():f.flush() self._frames=0 def close(self)->None: for f in self._files.values():f.close() self._files.clear();self._writers.clear() def __enter__(self)->"UnifiedEventLogger":self.open();return self def __exit__(self,*_args:object)->None:self.close()