""" Statistical export (iMotions-class data export for stats / R / Python). Turns a `Project` of participants + sessions into tidy tables ready for analysis in R (ggplot2/lme4), Python (pandas/statsmodels), SPSS, or Excel: * participants (wide) — one row per participant: physiology + gaze metrics, * aoi_long — tidy long: one row per (participant, stimulus, AOI), * events_long — tidy long: signal values at each stimulus onset, * group_summary — per-group means. Wide tables are convenient for a glance; LONG tables are what mixed-models and ggplot want (one observation per row), so those are the R-friendly export. Outputs a CSV bundle (one tidy file each) and/or a single multi-sheet .xlsx. Built on pandas; reuses the per-session analyzers and the aggregation layer. """ from __future__ import annotations import zipfile from pathlib import Path import numpy as np import pandas as pd from . import load as A from . import physio, eyetracking as ET, aoi as aoimod # -------------------------------------------------------------------------- # table builders # -------------------------------------------------------------------------- def participant_table(project) -> pd.DataFrame: """One row per participant: physiology + eye-movement metrics (wide).""" rows = [] for r in project.all(): for fname in r.sessions: sess = A.load_session(str(project.dir / fname)) row = {"participant": r.id, "name": r.name, "group": r.group or "(all)", "session": fname} h = physio.heart_metrics(sess) e = physio.eda_metrics(sess) row.update({"hr_bpm": h.get("hr_bpm"), "sdnn_ms": h.get("sdnn_ms"), "rmssd_ms": h.get("rmssd_ms"), "eda_tonic_uS": e.get("tonic_mean_uS"), "scr_per_min": e.get("scr_per_min")}) if "EyeTracker" in sess["streams"]: g = ET.gaze_metrics(sess) row.update({"fixations": g["fixation_count"], "mean_fixation_ms": g["mean_fixation_ms"], "saccades": g["saccade_count"], "mean_saccade_deg": g["mean_saccade_amp_deg"], "scanpath": g["scanpath_len_norm"]}) rows.append(row) return pd.DataFrame(rows) def aoi_long(project, aois, *, window=2.0) -> pd.DataFrame: """Tidy long AOI table: one row per (participant, session, stimulus, AOI).""" rows = [] for r in project.all(): for fname in r.sessions: sess = A.load_session(str(project.dir / fname)) for stim, trials in aoimod.per_stimulus(sess, aois, window=window).items(): for ti, trial in enumerate(trials): for aoi_name, m in trial.items(): rows.append({"participant": r.id, "group": r.group or "(all)", "session": fname, "stimulus": stim, "trial": ti + 1, "aoi": aoi_name, "dwell_s": m["dwell_time_s"], "ttff_s": m["ttff_s"], "fixations": m["fixation_count"], "hit": int(m["hit"]), "revisits": m["revisits"]}) return pd.DataFrame(rows) def events_long(project) -> pd.DataFrame: """Tidy long: each continuous signal's value at each stimulus onset.""" rows = [] for r in project.all(): for fname in r.sessions: sess = A.load_session(str(project.dir / fname)) streams = [n for n, s in sess["streams"].items() if s["type"] != "Markers"] for t, label in A.markers(sess): if not str(label).startswith("stim_on"): continue stim = str(label).split(":", 1)[-1] for sname in streams: try: val = A.value_at(sess, sname, float(t), channel=0) except Exception: continue rows.append({"participant": r.id, "group": r.group or "(all)", "session": fname, "stimulus": stim, "onset_s": float(t), "stream": sname, "value": val}) return pd.DataFrame(rows) def responses_long(project) -> pd.DataFrame: """Tidy long survey/website responses: one row per (participant, stimulus, item).""" from ..app.present import load_responses rows = [] for r in load_responses(str(project.dir)): data = r.get("data", {}) if not isinstance(data, dict) or not data: rows.append({"participant": r.get("participant", ""), "stimulus": r["stimulus"], "kind": r["kind"], "item": "", "answer": ""}) continue for item, ans in data.items(): rows.append({"participant": r.get("participant", ""), "stimulus": r["stimulus"], "kind": r["kind"], "item": item, "answer": ", ".join(ans) if isinstance(ans, list) else ans}) return pd.DataFrame(rows) def group_summary(project) -> pd.DataFrame: """Per-group means of the participant-level metrics.""" df = participant_table(project) if df.empty: return df num = df.select_dtypes("number").columns return df.groupby("group")[list(num)].mean().reset_index() # -------------------------------------------------------------------------- # writers # -------------------------------------------------------------------------- def _tables(project, aois) -> dict: out = {"participants": participant_table(project), "events_long": events_long(project), "responses_long": responses_long(project), "group_summary": group_summary(project)} if aois: out["aoi_long"] = aoi_long(project, aois) return out def to_csv_bundle(project, out_dir: str, aois=None) -> list[str]: """Write one tidy CSV per table. Returns the file paths.""" out = Path(out_dir) out.mkdir(parents=True, exist_ok=True) paths = [] for name, df in _tables(project, aois).items(): p = out / f"{name}.csv" df.to_csv(p, index=False) paths.append(str(p)) return paths def to_csv_zip(project, zip_path: str, aois=None) -> str: """Bundle the tidy CSVs into a single .zip (handy for an app download).""" import io with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z: for name, df in _tables(project, aois).items(): buf = io.StringIO() df.to_csv(buf, index=False) z.writestr(f"{name}.csv", buf.getvalue()) z.writestr("README.txt", _R_HINT) return zip_path def to_excel(project, xlsx_path: str, aois=None) -> str: """Write a single multi-sheet workbook (one sheet per table).""" with pd.ExcelWriter(xlsx_path, engine="openpyxl") as xl: for name, df in _tables(project, aois).items(): df.to_excel(xl, sheet_name=name[:31], index=False) return xlsx_path _R_HINT = ( "biosync statistical export — tidy CSVs.\n\n" "Long tables (aoi_long, events_long) are one observation per row, ready for R:\n" " library(tidyverse); library(lme4)\n" " aoi <- read_csv('aoi_long.csv')\n" " lmer(dwell_s ~ stimulus * aoi + (1|participant), data = aoi)\n" " ggplot(aoi, aes(stimulus, dwell_s, fill = aoi)) + geom_boxplot()\n" )