Buckets:
| #!/usr/bin/env python3 | |
| """analysis.py -- pre-specified analysis of the CockpitWM (C) vs stock openpilot (S) within-participant study (user_study_v1). | |
| Inputs (CSV). Two layouts are accepted and auto-detected (tools/schema_adapter.py): the "tools" layout below, and the | |
| ../data_schema.md layout (session_index, condition_code X/Y + `condition` or --condition-key, wide alerts.csv with alert_id / | |
| t_route_s / presented / suppressed, questionnaire item ids APP1..TRU3 / VDL1..9 (-2..+2, scored_value) / AL0..AL4 / END1..END5). | |
| Tools layout: | |
| participants.csv participant_id, order (SC|CS), ... | |
| sessions.csv participant_id, session_id, condition (S|C), order_position (1|2), start_utc, end_utc, count_from_s, route_end_s, ... | |
| count_from_s = C: pred_log t of the first ready=true tick; S: 90 (engine start + 90 s). Driving hours per | |
| session = (route_end_s - count_from_s) / 3600, falling back to (end_utc - start_utc) - count_from_s. | |
| alerts.csv participant_id, session_id, condition, source_device, t_rel_s, kind, detail (JSON) -- from extract_alerts.py | |
| questionnaire.csv participant_id, session_id, condition, order, item_id, value (1-7), reverse (0|1), clip_seq, free_text | |
| item ids: appr_1..3 timeliness time_1..2 annoyance annoy_1..2 trust trust_1..3 Van der Laan vdl_1..9 | |
| (usefulness = 1,3,5,7,9; satisfying = 2,4,6,8; mirrored items carry reverse=1); per-alert pa_appropriate / | |
| pa_timely / pa_want with clip_seq (session_id may be blank = blinded, resolved via --clip-manifest); | |
| end-of-study cmp_pref (value 1|2 = preferred session position, 0 = none), cmp_daily_1 / cmp_daily_2 (7-point, | |
| willingness to use daily, per session position); session_id blank for study-level rows. | |
| --clip-manifest analyst manifest of render_alert_clips.py (seq -> participant_id, session_id, condition) | |
| Outputs: markdown to stdout (and --out FILE). Primary endpoint = paired difference C - S of the per-session mean appropriateness; | |
| participant-level bootstrap CI (2000 resamples, fixed seed) + Wilcoxon signed-rank. Wording rule: a CI containing 0 is | |
| reported as "inconclusive" (never "equivalent"). Only participants with BOTH sessions enter any paired statistic. | |
| --dry-run: writes a SYNTHETIC, clearly labelled dataset (random numbers, null effect by construction) to --dry-run-dir and runs | |
| the full analysis on it so the pipeline is exercised end to end. Nothing in a dry-run output is a study result. | |
| --blind-codes: BLINDED run on the analyst view (data_schema.md section 0): condition code X is labelled "C" and Y "S" internally; every | |
| table is marked BLINDED (X - Y). The owner applies the real key last (sign kept if X = C, flipped if X = S). | |
| --no-prompt-neutral: the pre-specified sensitivity analysis NP-neutral (protocol section 11.1): appropriateness / timeliness / annoyance | |
| of a session whose "no prompt received" box is ticked (sessions.csv no_prompt_box_ticked = 1) are set to the neutral 4 instead of | |
| missing. Default = primary analysis (complete pairs with prompts in both sessions; the number of no-prompt sessions per arm is | |
| reported). Alert rows carrying an in_window flag (data_schema layout / extract_alerts.py --wide-out) are counted by that flag; the | |
| ES7 table adds the exposure breakdown (presented prompts engaged vs manual, engaged hours) of protocol section 9.3. | |
| """ | |
| import argparse | |
| import csv | |
| import datetime as dt | |
| import json | |
| import math | |
| import os | |
| import sys | |
| from collections import OrderedDict | |
| import numpy as np | |
| import pandas as pd | |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) | |
| import schema_adapter as SA # noqa: E402 | |
| SEED = 20260922 | |
| N_BOOT = 2000 | |
| SCALE_MAX = 7 | |
| AGREE_THR = 5 # 7-point: 5..7 = agree side | |
| CONSTRUCT_PREFIX = OrderedDict([("appr", "appropriateness"), ("time", "timeliness"), ("annoy", "annoyance"), ("trust", "trust")]) | |
| VDL_USEFULNESS = (1, 3, 5, 7, 9) | |
| VDL_SATISFYING = (2, 4, 6, 8) | |
| VDL_REVERSED = (3, 6, 8) # standard mirrored items (bad/good, irritating/likeable, undesirable/desirable) | |
| EXPECTED_SESSION_ITEMS = [f"appr_{i}" for i in (1, 2, 3)] + [f"time_{i}" for i in (1, 2)] + [f"annoy_{i}" for i in (1, 2)] + \ | |
| [f"trust_{i}" for i in (1, 2, 3)] + [f"vdl_{i}" for i in range(1, 10)] | |
| PA_ITEMS = ("pa_appropriate", "pa_timely", "pa_want") | |
| EXPECTED_STUDY_ITEMS = ("cmp_pref", "cmp_daily_1", "cmp_daily_2") | |
| SECONDARY = ("timeliness", "annoyance", "trust", "vdl_usefulness", "vdl_satisfying") | |
| EXPECTED_ITEMS_PER_CONSTRUCT = {"appropriateness": 3, "timeliness": 2, "annoyance": 2, "trust": 3, "vdl_usefulness": 5, "vdl_satisfying": 4} | |
| PRIMARY = "appropriateness" | |
| # ------------------------------------------------------------------------------------------------------------------ loading | |
| def construct_of(item_id): | |
| s = str(item_id) | |
| if s.startswith("vdl_"): | |
| try: | |
| k = int(s.split("_", 1)[1]) | |
| except ValueError: | |
| return None | |
| return "vdl_usefulness" if k in VDL_USEFULNESS else ("vdl_satisfying" if k in VDL_SATISFYING else None) | |
| pre = s.split("_", 1)[0] | |
| return CONSTRUCT_PREFIX.get(pre) | |
| def load_tables(participants, sessions, alerts, questionnaire, clip_manifest=None, condition_key=None, blind_codes=False): | |
| T = {} | |
| T["participants"] = pd.read_csv(participants, dtype=str).fillna("") | |
| T["sessions"] = pd.read_csv(sessions, dtype=str).fillna("") | |
| T["alerts"] = pd.read_csv(alerts, dtype=str).fillna("") if alerts and os.path.exists(alerts) else pd.DataFrame(columns=["participant_id", "session_id", "condition", "source_device", "t_rel_s", "kind", "detail"]) | |
| T["questionnaire"] = pd.read_csv(questionnaire, dtype=str).fillna("") | |
| T["manifest"] = pd.read_csv(clip_manifest, dtype=str).fillna("") if clip_manifest else None | |
| key = {"X": "C", "Y": "S"} if blind_codes else SA.read_key(condition_key) # blind run: codes stand in for conditions | |
| layout = dict(participants=SA.detect_participants(T["participants"]), sessions=SA.detect_sessions(T["sessions"]), alerts=SA.detect_alerts(T["alerts"]), | |
| questionnaire=SA.detect_questionnaire(T["questionnaire"])) | |
| T["layout"] = layout | |
| alerts_D = T["alerts"] if layout["alerts"] == "D" else None | |
| T["participants"] = SA.normalise_participants(T["participants"], key) | |
| T["sessions"] = SA.normalise_sessions(T["sessions"], key) | |
| T["alerts"] = SA.normalise_alerts(T["alerts"], T["sessions"]) | |
| T["questionnaire"] = SA.normalise_questionnaire(T["questionnaire"], T["sessions"], alerts_D, key) | |
| if T["manifest"] is None and alerts_D is not None: | |
| T["manifest"] = SA.manifest_from_wide_alerts(alerts_D, T["sessions"]) | |
| T["n_sessions_without_condition"] = int((T["sessions"]["condition"].astype(str).str.strip() == "").sum()) | |
| for name, cols in (("participants", ["participant_id", "order"]), ("sessions", ["participant_id", "session_id", "condition", "order_position", "count_from_s"]), | |
| ("alerts", ["participant_id", "session_id", "condition", "t_rel_s", "kind", "detail"]), ("questionnaire", ["participant_id", "item_id", "value"])): | |
| missing = [c for c in cols if c not in T[name].columns] | |
| if missing: | |
| raise SystemExit(f"{name}.csv: missing columns {missing}") | |
| q = T["questionnaire"] | |
| for c in ("session_id", "condition", "order", "reverse", "clip_seq", "free_text"): | |
| if c not in q.columns: | |
| q[c] = "" | |
| q["value_num"] = pd.to_numeric(q["value"], errors="coerce") | |
| q["reverse_num"] = pd.to_numeric(q["reverse"], errors="coerce").fillna(0).astype(int) | |
| T["questionnaire"] = q | |
| s = T["sessions"] | |
| s["order_position_num"] = pd.to_numeric(s["order_position"], errors="coerce") | |
| s["count_from_s_num"] = pd.to_numeric(s["count_from_s"], errors="coerce") | |
| s["route_end_s_num"] = pd.to_numeric(s.get("route_end_s", pd.Series([""] * len(s))), errors="coerce") | |
| T["sessions"] = s | |
| return T | |
| def session_hours(sess_row): | |
| """driving hours counted from count_from_s: (route_end_s - count_from_s)/3600, else (end_utc - start_utc - count_from_s)/3600.""" | |
| c0 = sess_row.get("count_from_s_num") | |
| c0 = 0.0 if (c0 is None or (isinstance(c0, float) and math.isnan(c0))) else float(c0) | |
| end_s = sess_row.get("route_end_s_num") | |
| if end_s is not None and not (isinstance(end_s, float) and math.isnan(end_s)): | |
| return max(0.0, (float(end_s) - c0) / 3600.0) | |
| try: | |
| t0 = dt.datetime.fromisoformat(str(sess_row.get("start_utc")).replace("Z", "+00:00")) | |
| t1 = dt.datetime.fromisoformat(str(sess_row.get("end_utc")).replace("Z", "+00:00")) | |
| return max(0.0, ((t1 - t0).total_seconds() - c0) / 3600.0) | |
| except Exception: | |
| return float("nan") | |
| def complete_participants(T): | |
| s = T["sessions"] | |
| have = s.groupby("participant_id")["condition"].agg(lambda x: set(x)) | |
| both = sorted(p for p, conds in have.items() if {"S", "C"} <= conds) | |
| only = sorted(p for p, conds in have.items() if not {"S", "C"} <= conds) | |
| return both, only | |
| # ---------------------------------------------------------------------------------------------------------------- statistics | |
| def rankdata_avg(x): | |
| x = np.asarray(x, float) | |
| order = np.argsort(x, kind="mergesort") | |
| ranks = np.empty(len(x), float) | |
| i = 0 | |
| while i < len(x): | |
| j = i | |
| while j + 1 < len(x) and x[order[j + 1]] == x[order[i]]: | |
| j += 1 | |
| ranks[order[i:j + 1]] = (i + j) / 2.0 + 1.0 | |
| i = j + 1 | |
| return ranks | |
| def wilcoxon_signed_rank(d): | |
| """two-sided Wilcoxon signed-rank test on paired differences d (zeros dropped, 'wilcox' zero method). Exact for n <= 50: | |
| the null distribution of W+ over all 2^n sign assignments of the observed (possibly tied, average) ranks, computed by a | |
| dynamic programme over doubled ranks; two-sided p = 2 * P(W+ <= min(W+, W-)) capped at 1. Checked 2026-09-21 against | |
| scipy 1.15: without ties identical to scipy.stats.wilcoxon(method='exact'); with ties identical to the exhaustive permutation | |
| distribution (scipy PermutationMethod with 2^n resamples, max |dp| = 0.0 on 240 random vectors), which scipy's method='auto' | |
| only approximates (up to 0.024 off). For n > 50: normal approximation with tie correction, no continuity correction | |
| (identical to scipy method='approx', correction=False). Returns dict(W_plus, W_minus, n, p, method).""" | |
| d = np.asarray(d, float) | |
| d = d[~np.isnan(d)] | |
| d = d[d != 0] | |
| n = len(d) | |
| if n == 0: | |
| return dict(W_plus=None, W_minus=None, n=0, p=None, method="no non-zero differences") | |
| r = rankdata_avg(np.abs(d)) | |
| wp, wm = float(r[d > 0].sum()), float(r[d < 0].sum()) | |
| _, counts = np.unique(np.abs(d), return_counts=True) | |
| ties = bool((counts > 1).any()) | |
| if n <= 50: | |
| r2 = np.rint(2.0 * r).astype(int) # average ranks are multiples of 1/2 | |
| smax = int(r2.sum()) | |
| cnt = [0] * (smax + 1) | |
| cnt[0] = 1 | |
| for k in r2: | |
| k = int(k) | |
| for sidx in range(smax, k - 1, -1): | |
| if cnt[sidx - k]: | |
| cnt[sidx] += cnt[sidx - k] | |
| w2 = int(np.rint(2.0 * min(wp, wm))) | |
| p = 2.0 * sum(cnt[: w2 + 1]) / float(2 ** n) | |
| return dict(W_plus=wp, W_minus=wm, n=n, p=min(1.0, p), method="exact" + (" (ties)" if ties else "")) | |
| mn = n * (n + 1) / 4.0 | |
| tie_corr = float(((counts ** 3 - counts)).sum()) | |
| var = n * (n + 1) * (2 * n + 1) / 24.0 - tie_corr / 48.0 | |
| z = (wp - mn) / math.sqrt(var) if var > 0 else 0.0 | |
| p = math.erfc(abs(z) / math.sqrt(2.0)) | |
| return dict(W_plus=wp, W_minus=wm, n=n, p=min(1.0, p), method="normal approximation (tie-corrected)") | |
| def bootstrap_mean_ci(x, n_boot=N_BOOT, seed=SEED, alpha=0.05): | |
| x = np.asarray(x, float) | |
| x = x[~np.isnan(x)] | |
| if len(x) == 0: | |
| return dict(n=0, mean=None, lo=None, hi=None) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, len(x), size=(n_boot, len(x))) | |
| means = x[idx].mean(axis=1) | |
| return dict(n=int(len(x)), mean=float(x.mean()), lo=float(np.percentile(means, 100 * alpha / 2)), hi=float(np.percentile(means, 100 * (1 - alpha / 2)))) | |
| def bootstrap_two_group_ci(a, b, n_boot=N_BOOT, seed=SEED, alpha=0.05): | |
| """CI of mean(a) - mean(b), resampling within each group.""" | |
| a, b = np.asarray(a, float), np.asarray(b, float) | |
| a, b = a[~np.isnan(a)], b[~np.isnan(b)] | |
| if len(a) == 0 or len(b) == 0: | |
| return dict(n_a=len(a), n_b=len(b), diff=None, lo=None, hi=None) | |
| rng = np.random.default_rng(seed) | |
| ma = a[rng.integers(0, len(a), size=(n_boot, len(a)))].mean(axis=1) | |
| mb = b[rng.integers(0, len(b), size=(n_boot, len(b)))].mean(axis=1) | |
| d = ma - mb | |
| return dict(n_a=int(len(a)), n_b=int(len(b)), diff=float(a.mean() - b.mean()), lo=float(np.percentile(d, 100 * alpha / 2)), hi=float(np.percentile(d, 100 * (1 - alpha / 2)))) | |
| def verdict(lo, hi, higher="C > S", lower="C < S"): | |
| if lo is None or hi is None: | |
| return "not estimable" | |
| if lo <= 0 <= hi: | |
| return "inconclusive (95% CI contains 0)" | |
| return f"{higher} (95% CI excludes 0)" if lo > 0 else f"{lower} (95% CI excludes 0)" | |
| def fmt(x, nd=2): | |
| if x is None or (isinstance(x, float) and math.isnan(x)): | |
| return "n/a" | |
| return f"{x:.{nd}f}" | |
| # ------------------------------------------------------------------------------------------------------------------ measures | |
| def session_construct_means(T, scale_max=SCALE_MAX): | |
| q = T["questionnaire"] | |
| q = q[(q["session_id"] != "") & q["value_num"].notna()].copy() | |
| q["construct"] = q["item_id"].map(construct_of) | |
| q = q[q["construct"].notna()].copy() | |
| q["v"] = np.where(q["reverse_num"] == 1, scale_max + 1 - q["value_num"], q["value_num"]) | |
| g = q.groupby(["participant_id", "session_id", "construct"], as_index=False).agg(mean=("v", "mean"), n_items=("v", "size")) | |
| # questionnaire_zh_en.md section 4 rule 2: a construct is missing when fewer than half of its items are answered | |
| need = {c: math.ceil(n / 2) for c, n in EXPECTED_ITEMS_PER_CONSTRUCT.items()} | |
| g = g[[n >= need.get(c, 1) for c, n in zip(g["construct"], g["n_items"])]].copy() | |
| sess = T["sessions"][["participant_id", "session_id", "condition", "order_position_num"]] | |
| return g.merge(sess, on=["participant_id", "session_id"], how="left") | |
| def paired_table(scm, construct, both): | |
| d = scm[(scm["construct"] == construct) & scm["participant_id"].isin(both)] | |
| piv = d.pivot_table(index="participant_id", columns="condition", values="mean", aggfunc="mean") | |
| for c in ("C", "S"): | |
| if c not in piv.columns: | |
| piv[c] = np.nan | |
| piv = piv.dropna(subset=["C", "S"]) | |
| piv["diff"] = piv["C"] - piv["S"] | |
| return piv | |
| def paired_block(name, piv, n_boot, seed): | |
| ci = bootstrap_mean_ci(piv["diff"].values, n_boot, seed) | |
| wt = wilcoxon_signed_rank(piv["diff"].values) | |
| return OrderedDict(endpoint=name, n=int(len(piv)), mean_C=float(piv["C"].mean()) if len(piv) else None, mean_S=float(piv["S"].mean()) if len(piv) else None, | |
| diff=ci["mean"], lo=ci["lo"], hi=ci["hi"], W_plus=wt["W_plus"], p=wt["p"], method=wt["method"], verdict=verdict(ci["lo"], ci["hi"])) | |
| def _detail(row): | |
| try: | |
| return json.loads(row["detail"]) if row.get("detail") else {} | |
| except Exception: | |
| return {} | |
| def _in_window_flag(d): | |
| """explicit in_window of the row (data_schema layout / extract_alerts --wide-out): 1 / 0 / None (unknown -> fall back to t >= count_from_s).""" | |
| v = d.get("in_window") | |
| if v in (None, "", "nan"): | |
| return None | |
| try: | |
| return int(float(v)) | |
| except (TypeError, ValueError): | |
| return None | |
| def _engaged_flag(d): | |
| """engaged / manual / unknown at the prompt onset (rlog selfdriveState.enabled via op_engaged / op_engaged_at_t; CockpitWM regime).""" | |
| for k in ("op_engaged", "op_engaged_at_t"): | |
| v = d.get(k) | |
| if v in (True, 1, "1", "True", "true"): | |
| return "engaged" | |
| if v in (False, 0, "0", "False", "false"): | |
| return "manual" | |
| if d.get("regime") in ("engaged", "manual"): | |
| return d["regime"] | |
| return "unknown" | |
| def alert_rates(T, both): | |
| """per-session presented prompts in the counting window (row in_window flag when present, else t >= count_from_s), hours, rates, | |
| and the protocol section 9.3 exposure breakdown (presented prompts engaged vs manual; engaged hours from sessions.csv engaged_time_s).""" | |
| s = T["sessions"] | |
| al = T["alerts"].copy() | |
| al["t"] = pd.to_numeric(al["t_rel_s"], errors="coerce") | |
| al["_d"] = [_detail(r) for _, r in al.iterrows()] | |
| al["_iw"] = [_in_window_flag(d) for d in al["_d"]] | |
| al["_eng"] = [_engaged_flag(d) for d in al["_d"]] | |
| rows = [] | |
| for _, sr in s.iterrows(): | |
| if sr["participant_id"] not in both: | |
| continue | |
| c0 = sr["count_from_s_num"] if not math.isnan(sr["count_from_s_num"]) else 0.0 | |
| h = session_hours(sr) | |
| a = al[al["session_id"] == sr["session_id"]] | |
| a = a[np.array([(iw == 1) if iw is not None else bool(t >= c0) for iw, t in zip(a["_iw"], a["t"])], dtype=bool)] | |
| a = a[np.array([d.get("event") != "tooDistracted" for d in a["_d"]], dtype=bool)] | |
| pres = a[a["kind"] == "stock_alert_presented"] | |
| n_pres = int(len(pres)) | |
| n_cwm = int((a["kind"] == "cockpitwm_alert").sum()) | |
| n_cf = int((a["kind"] == "cockpitwm_counterfactual").sum()) | |
| eng_h = pd.to_numeric(sr.get("engaged_time_s"), errors="coerce") if "engaged_time_s" in sr.index else np.nan | |
| eng_h = float(eng_h) / 3600.0 if eng_h is not None and not (isinstance(eng_h, float) and math.isnan(eng_h)) else np.nan | |
| n_eng, n_man, n_unk = int((pres["_eng"] == "engaged").sum()), int((pres["_eng"] == "manual").sum()), int((pres["_eng"] == "unknown").sum()) | |
| rows.append(dict(participant_id=sr["participant_id"], session_id=sr["session_id"], condition=sr["condition"], hours=h, count_from_s=c0, | |
| presented=n_pres, presented_per_h=(n_pres / h if h and h > 0 else np.nan), cockpitwm_alerts=n_cwm, counterfactual=n_cf, | |
| cockpitwm_per_h=(n_cwm / h if h and h > 0 else np.nan), presented_engaged=n_eng, presented_manual=n_man, presented_unknown=n_unk, | |
| engaged_hours=eng_h, presented_per_engaged_h=(n_eng / eng_h if eng_h and eng_h > 0 else np.nan), | |
| no_prompt_box=pd.to_numeric(sr.get("no_prompt_box_ticked"), errors="coerce") if "no_prompt_box_ticked" in sr.index else np.nan)) | |
| return pd.DataFrame(rows) | |
| def apply_no_prompt_neutral(scm, T, neutral=4.0): | |
| """protocol section 11.1 sensitivity analysis NP-neutral: sessions with no_prompt_box_ticked = 1 get appropriateness / timeliness / | |
| annoyance = the neutral scale midpoint when the construct is missing. Returns (scm, n_filled).""" | |
| s = T["sessions"] | |
| if "no_prompt_box_ticked" not in s.columns: | |
| return scm, 0 | |
| flag = pd.to_numeric(s["no_prompt_box_ticked"], errors="coerce").fillna(0).astype(int) | |
| add = [] | |
| for _, sr in s[flag == 1].iterrows(): | |
| have = set(scm[scm["session_id"] == sr["session_id"]]["construct"]) | |
| for c in ("appropriateness", "timeliness", "annoyance"): | |
| if c not in have: | |
| add.append(dict(participant_id=sr["participant_id"], session_id=sr["session_id"], construct=c, mean=neutral, n_items=0, | |
| condition=sr["condition"], order_position_num=sr["order_position_num"])) | |
| if add: | |
| scm = pd.concat([scm, pd.DataFrame(add)], ignore_index=True) | |
| return scm, len(add) | |
| def suppression(T, both): | |
| """C sessions: stock alertLevel transitions to two/three after count_from_s; share hidden by CockpitWM (level two only can be hidden). | |
| When sessions.csv carries the data_schema.md columns n_dm_level_ge2_moments / n_dm_level_ge2_suppressed (rlog-derived by the | |
| session extractor), those are used directly; otherwise the alerts.csv stock_level_change rows are counted.""" | |
| s = T["sessions"] | |
| if "n_dm_level_ge2_moments" in s.columns: | |
| sc = s[(s["condition"] == "C") & s["participant_id"].isin(both)] | |
| nm = pd.to_numeric(sc["n_dm_level_ge2_moments"], errors="coerce") | |
| ns = pd.to_numeric(sc.get("n_dm_level_ge2_suppressed", pd.Series([np.nan] * len(sc), index=sc.index)), errors="coerce") | |
| if nm.notna().any(): | |
| n, n_h = int(nm.fillna(0).sum()), int(ns.fillna(0).sum()) | |
| return dict(n_level_ge_two=n, n_two=None, n_three=None, n_hidden=n_h, n_unknown=int(nm.isna().sum()), share_hidden=(n_h / n if n else None), source="sessions.csv columns") | |
| al = T["alerts"] | |
| al = al[(al["kind"] == "stock_level_change") & (al["condition"] == "C") & al["participant_id"].isin(both)].copy() | |
| al["t"] = pd.to_numeric(al["t_rel_s"], errors="coerce") | |
| c0 = dict(zip(s["session_id"], s["count_from_s_num"].fillna(0.0))) | |
| n = n_hidden = n_unknown = n_two = n_three = 0 | |
| for _, r in al.iterrows(): | |
| if r["t"] < c0.get(r["session_id"], 0.0): | |
| continue | |
| try: | |
| d = json.loads(r["detail"]) | |
| except Exception: | |
| continue | |
| if d.get("level") not in ("two", "three"): | |
| continue | |
| n += 1 | |
| n_two += d.get("level") == "two" | |
| n_three += d.get("level") == "three" | |
| h = d.get("hidden_by_cockpitwm") | |
| if h is None: | |
| n_unknown += 1 | |
| elif h: | |
| n_hidden += 1 | |
| return dict(n_level_ge_two=n, n_two=n_two, n_three=n_three, n_hidden=n_hidden, n_unknown=n_unknown, share_hidden=(n_hidden / n if n else None)) | |
| def per_alert_shares(T, both, agree_thr, n_boot, seed): | |
| q = T["questionnaire"] | |
| pa = q[q["item_id"].isin(PA_ITEMS) & q["value_num"].notna()].copy() | |
| if len(pa) == 0: | |
| return None | |
| if T["manifest"] is not None and "seq" in T["manifest"].columns: | |
| m = T["manifest"][["participant_id", "seq", "session_id", "condition"]].rename(columns={"session_id": "m_session", "condition": "m_condition"}) | |
| def _k(x): | |
| x = str(x).strip() | |
| return x.lstrip("0") or "0" if x.isdigit() else x | |
| m["seq"] = m["seq"].map(_k) | |
| pa["clip_key"] = pa["clip_seq"].map(_k) | |
| pa = pa.merge(m, left_on=["participant_id", "clip_key"], right_on=["participant_id", "seq"], how="left") | |
| pa["condition"] = np.where(pa["m_condition"].fillna("") != "", pa["m_condition"], pa["condition"]) | |
| pa["session_id"] = np.where(pa["m_session"].fillna("") != "", pa["m_session"], pa["session_id"]) | |
| pa = pa[pa["condition"].isin(["S", "C"]) & pa["participant_id"].isin(both)].copy() | |
| pa["agree"] = (pa["value_num"] >= agree_thr).astype(float) | |
| out = [] | |
| for item in PA_ITEMS: | |
| d = pa[pa["item_id"] == item] | |
| row = OrderedDict(item=item, n_C=int((d["condition"] == "C").sum()), n_S=int((d["condition"] == "S").sum()), | |
| share_C=float(d[d["condition"] == "C"]["agree"].mean()) if (d["condition"] == "C").any() else None, | |
| share_S=float(d[d["condition"] == "S"]["agree"].mean()) if (d["condition"] == "S").any() else None) | |
| # participant-level paired difference of shares (participants with rated alerts in both conditions) | |
| pp = d.groupby(["participant_id", "condition"])["agree"].mean().unstack() | |
| if "C" in pp.columns and "S" in pp.columns: | |
| pp = pp.dropna(subset=["C", "S"]) | |
| ci = bootstrap_mean_ci((pp["C"] - pp["S"]).values, n_boot, seed) | |
| row.update(n_pairs=ci["n"], diff=ci["mean"], lo=ci["lo"], hi=ci["hi"], verdict=verdict(ci["lo"], ci["hi"])) | |
| else: | |
| row.update(n_pairs=0, diff=None, lo=None, hi=None, verdict="not estimable") | |
| out.append(row) | |
| return out | |
| def comparative(T, both, n_boot, seed): | |
| q = T["questionnaire"] | |
| s = T["sessions"] | |
| pos2cond = {} | |
| for _, r in s.iterrows(): | |
| if not math.isnan(r["order_position_num"]): | |
| pos2cond[(r["participant_id"], int(r["order_position_num"]))] = r["condition"] | |
| pref = q[(q["item_id"] == "cmp_pref") & q["participant_id"].isin(both) & q["value_num"].notna()] | |
| counts = {"C": 0, "S": 0, "none": 0, "unmapped": 0} | |
| for _, r in pref.iterrows(): | |
| v = int(r["value_num"]) | |
| if v == 0: | |
| counts["none"] += 1 | |
| else: | |
| c = pos2cond.get((r["participant_id"], v)) | |
| counts[c if c in ("C", "S") else "unmapped"] += 1 | |
| daily = {} | |
| rows = [] | |
| for _, r in q[q["item_id"].isin(["cmp_daily_1", "cmp_daily_2"]) & q["participant_id"].isin(both) & q["value_num"].notna()].iterrows(): | |
| pos = int(r["item_id"].rsplit("_", 1)[1]) | |
| c = pos2cond.get((r["participant_id"], pos)) | |
| if c in ("C", "S"): | |
| rows.append(dict(participant_id=r["participant_id"], condition=c, v=float(r["value_num"]))) | |
| if rows: | |
| piv = pd.DataFrame(rows).pivot_table(index="participant_id", columns="condition", values="v", aggfunc="mean") | |
| for c in ("C", "S"): | |
| if c not in piv.columns: | |
| piv[c] = np.nan | |
| piv = piv.dropna(subset=["C", "S"]) | |
| piv["diff"] = piv["C"] - piv["S"] | |
| daily = paired_block("willingness to use daily (cmp_daily)", piv, n_boot, seed) | |
| n_pref = sum(counts.values()) | |
| # EQ1 blinding check (END3 / blind_guess): did the participant identify the C session? | |
| bg = q[(q["item_id"] == "blind_guess") & q["participant_id"].isin(both)] | |
| blind = {"correct": 0, "wrong": 0, "cannot_tell": 0, "n": 0} | |
| for _, r in bg.iterrows(): | |
| v = r["value_num"] | |
| if math.isnan(v): | |
| continue | |
| blind["n"] += 1 | |
| if int(v) == 0: | |
| blind["cannot_tell"] += 1 | |
| else: | |
| blind["correct" if pos2cond.get((r["participant_id"], int(v))) == "C" else "wrong"] += 1 | |
| return dict(pref_counts=counts, n_pref=n_pref, pref_share_C=(counts["C"] / n_pref if n_pref else None), daily=daily, blind=blind) | |
| def order_check(scm, piv_primary, T, both, n_boot, seed): | |
| parts = T["participants"].set_index("participant_id")["order"].to_dict() | |
| piv = piv_primary.copy() | |
| piv["order"] = [parts.get(p, "") for p in piv.index] | |
| groups = {} | |
| for o in ("SC", "CS"): | |
| d = piv[piv["order"] == o]["diff"].values | |
| groups[o] = dict(n=int(len(d)), mean_diff=(float(np.mean(d)) if len(d) else None)) | |
| between = bootstrap_two_group_ci(piv[piv["order"] == "SC"]["diff"].values, piv[piv["order"] == "CS"]["diff"].values, n_boot, seed) | |
| # position effect: appropriateness in session 1 vs session 2 regardless of condition | |
| d = scm[(scm["construct"] == PRIMARY) & scm["participant_id"].isin(both)].dropna(subset=["order_position_num"]) | |
| pos = d.pivot_table(index="participant_id", columns="order_position_num", values="mean", aggfunc="mean") | |
| pos_ci = None | |
| if 1.0 in pos.columns and 2.0 in pos.columns: | |
| pp = pos.dropna(subset=[1.0, 2.0]) | |
| pos_ci = bootstrap_mean_ci((pp[2.0] - pp[1.0]).values, n_boot, seed) | |
| return dict(groups=groups, between=between, position=pos_ci) | |
| # ------------------------------------------------------------------------------------------------------------------ report | |
| def render_markdown(T, res, label, args): | |
| L = [] | |
| L.append(f"# user_study_v1 analysis -- {label}") | |
| if res.get("synthetic"): | |
| L.append("") | |
| L.append("**SYNTHETIC EXAMPLE DATA.** All numbers below come from random numbers generated by `analysis.py --dry-run` " | |
| "(null effect by construction). They are NOT study results and must not be quoted. Under a null about 1 in 20 endpoints " | |
| "will show a 95% CI excluding 0 by chance; that is expected here, not a finding.") | |
| L.append("") | |
| L.append(f"- inputs: participants={args.participants}, sessions={args.sessions}, alerts={args.alerts}, questionnaire={args.questionnaire}, clip_manifest={args.clip_manifest}") | |
| L.append(f"- CSV layout detected: {T.get('layout')}; sessions without a resolved condition (S/C): {T.get('n_sessions_without_condition', 0)}" | |
| + (" -- give --condition-key or fill sessions.csv `condition`" if T.get('n_sessions_without_condition', 0) else "")) | |
| L.append(f"- participants with both sessions (analysed): **{len(res['both'])}**; excluded (missing a session or an unresolved S/C condition): {len(res['only'])} {res['only'] if res['only'] else ''}") | |
| L.append(f"- bootstrap: {args.n_boot} participant-level resamples, seed {args.seed}, percentile 95% CI; Wilcoxon signed-rank two-sided; scale max {args.scale_max}; agree threshold >= {args.agree_thr}") | |
| L.append("- wording rule: a CI containing 0 is reported as *inconclusive* (never 'equivalent'); differences are C minus S") | |
| if getattr(args, "blind_codes", False): | |
| L.append("- **BLINDED RUN (X - Y): every 'C' below is condition code X and every 'S' is code Y (analyst view, protocol section 4). " | |
| "The owner applies `condition_key.txt` last: the sign of every difference is kept if X = C and flipped if X = S.**") | |
| if getattr(args, "no_prompt_neutral", False): | |
| L.append(f"- **SENSITIVITY ANALYSIS NP-neutral (protocol section 11.1):** appropriateness / timeliness / annoyance of {res.get('n_np_filled', 0)} " | |
| f"no-prompt session-constructs set to the neutral {(args.scale_max + 1) / 2:g}. Not the primary analysis.") | |
| else: | |
| L.append("- primary analysis: EP0 / ES1 / ES2 on complete pairs with prompts in both sessions (no-prompt sessions per arm listed under ES7); " | |
| "run again with --no-prompt-neutral for the pre-specified sensitivity analysis") | |
| L.append("") | |
| L.append("## Primary endpoint") | |
| L.append("") | |
| L.append("| endpoint | n | mean C | mean S | diff C-S | 95% CI | Wilcoxon W+ | p | method | reading |") | |
| L.append("|---|---|---|---|---|---|---|---|---|---|") | |
| b = res["primary"] | |
| L.append(f"| {b['endpoint']} (per-session mean of appr_1..3) | {b['n']} | {fmt(b['mean_C'])} | {fmt(b['mean_S'])} | {fmt(b['diff'])} | [{fmt(b['lo'])}, {fmt(b['hi'])}] | {fmt(b['W_plus'], 1)} | {fmt(b['p'], 4)} | {b['method']} | {b['verdict']} |") | |
| L.append("") | |
| L.append("## Secondary endpoints (questionnaire constructs)") | |
| L.append("") | |
| L.append("| endpoint | n | mean C | mean S | diff C-S | 95% CI | Wilcoxon W+ | p | reading |") | |
| L.append("|---|---|---|---|---|---|---|---|---|") | |
| for b in res["secondary"]: | |
| L.append(f"| {b['endpoint']} | {b['n']} | {fmt(b['mean_C'])} | {fmt(b['mean_S'])} | {fmt(b['diff'])} | [{fmt(b['lo'])}, {fmt(b['hi'])}] | {fmt(b['W_plus'], 1)} | {fmt(b['p'], 4)} | {b['verdict']} |") | |
| L.append("") | |
| L.append("annoyance is scored as asked (higher = more annoying); a negative diff means C was rated less annoying. " | |
| "vdl_* means are on the questionnaire's own scale (-2..+2 when entered per questionnaire_zh_en.md section 4; 1..7 in the tools layout).") | |
| L.append("") | |
| L.append("## Presented alerts per driving hour (rlog selfdriveState.alertType of the device in use)") | |
| L.append("") | |
| r = res["rates"] | |
| if r is not None and len(r): | |
| for c in ("C", "S"): | |
| d = r[r["condition"] == c] | |
| L.append(f"- {c}: sessions {len(d)}, hours {fmt(d['hours'].sum())}, presented alerts {int(d['presented'].sum())}, " | |
| f"pooled rate {fmt(d['presented'].sum() / d['hours'].sum() if d['hours'].sum() > 0 else float('nan'))} /h, per-session median {fmt(d['presented_per_h'].median())} /h") | |
| d = r[r["condition"] == "C"] | |
| L.append(f"- C only: CockpitWM alerts (pred_log onsets, effectiveSource=cockpitwm) {int(d['cockpitwm_alerts'].sum())} " | |
| f"({fmt(d['cockpitwm_alerts'].sum() / d['hours'].sum() if d['hours'].sum() > 0 else float('nan'))} /h); counterfactual onsets (not shown) {int(d['counterfactual'].sum())}") | |
| L.append("- exposure breakdown (protocol section 9.3; S can prompt only while engaged and above the low-speed exemption, C in any regime):") | |
| for c in ("C", "S"): | |
| d = r[r["condition"] == c] | |
| eh = d["engaged_hours"].sum(skipna=True) | |
| npb = d["no_prompt_box"] | |
| n_np = int((pd.to_numeric(npb, errors="coerce") == 1).sum()) if len(npb) else 0 | |
| L.append(f" - {c}: presented while engaged {int(d['presented_engaged'].sum())}, while manual {int(d['presented_manual'].sum())}, unknown {int(d['presented_unknown'].sum())}; " | |
| f"engaged hours {fmt(eh) if eh else 'n/a'}" + (f", engaged-only rate {fmt(d['presented_engaged'].sum() / eh)} /h" if eh else "") | |
| + f"; sessions with no presented prompt (log) {int((d['presented'] == 0).sum())}, 'no prompt' box ticked {n_np}") | |
| b = res["rate_block"] | |
| if b: | |
| L.append(f"- paired difference of per-session rates C-S: n {b['n']}, diff {fmt(b['diff'])} /h, 95% CI [{fmt(b['lo'])}, {fmt(b['hi'])}], Wilcoxon p {fmt(b['p'], 4)} -> {b['verdict']}") | |
| else: | |
| L.append("- no alert rows") | |
| L.append("") | |
| L.append("## Per-alert retrospective ratings (blinded clips; share rated >= agree threshold)") | |
| L.append("") | |
| if res["pa"]: | |
| L.append("| item | n C | n S | share C | share S | participant-level diff C-S | 95% CI | n pairs | reading |") | |
| L.append("|---|---|---|---|---|---|---|---|---|") | |
| for row in res["pa"]: | |
| L.append(f"| {row['item']} | {row['n_C']} | {row['n_S']} | {fmt(row['share_C'])} | {fmt(row['share_S'])} | {fmt(row['diff'])} | [{fmt(row['lo'])}, {fmt(row['hi'])}] | {row['n_pairs']} | {row['verdict']} |") | |
| else: | |
| L.append("- no per-alert ratings") | |
| L.append("") | |
| L.append("## Stock monitor level >= two moments in C sessions (driverMonitoringState.alertLevel)") | |
| L.append("") | |
| sp = res["suppression"] | |
| L.append(f"- transitions to level two/three after count_from_s: {sp['n_level_ge_two']} (two {sp['n_two']}, three {sp['n_three']}); hidden by CockpitWM: {sp['n_hidden']} " | |
| f"(share {fmt(sp['share_hidden'], 3)}); unknown: {sp['n_unknown']}; source: {sp.get('source', 'alerts.csv stock_level_change rows')}. Level three is never hidden by the fork.") | |
| L.append("") | |
| L.append("## End-of-study comparative questions") | |
| L.append("") | |
| cp = res["comparative"] | |
| L.append(f"- preference (cmp_pref mapped through order_position): C {cp['pref_counts']['C']}, S {cp['pref_counts']['S']}, none {cp['pref_counts']['none']}, unmapped {cp['pref_counts']['unmapped']} " | |
| f"(n {cp['n_pref']}, share preferring C {fmt(cp['pref_share_C'], 3)})") | |
| if cp["daily"]: | |
| b = cp["daily"] | |
| L.append(f"- {b['endpoint']}: n {b['n']}, mean C {fmt(b['mean_C'])}, mean S {fmt(b['mean_S'])}, diff {fmt(b['diff'])}, 95% CI [{fmt(b['lo'])}, {fmt(b['hi'])}], Wilcoxon p {fmt(b['p'], 4)} -> {b['verdict']}") | |
| bl = cp.get("blind") or {} | |
| if bl.get("n"): | |
| L.append(f"- blinding check (END3 / blind_guess): n {bl['n']}, identified the C session correctly {bl['correct']}, wrong {bl['wrong']}, cannot tell {bl['cannot_tell']} " | |
| f"(chance for a forced guess = 1/2; report as EQ1, no inference)") | |
| L.append("") | |
| L.append("## Order-effect check (covariate check on the primary endpoint)") | |
| L.append("") | |
| oc = res["order"] | |
| for o, g in oc["groups"].items(): | |
| L.append(f"- order {o}: n {g['n']}, mean diff C-S {fmt(g['mean_diff'])}") | |
| bt = oc["between"] | |
| L.append(f"- difference of the two order groups (SC minus CS): {fmt(bt['diff'])}, 95% CI [{fmt(bt['lo'])}, {fmt(bt['hi'])}] -> {verdict(bt['lo'], bt['hi'], 'SC > CS', 'SC < CS')}") | |
| if oc["position"]: | |
| p = oc["position"] | |
| L.append(f"- session position effect on appropriateness (session 2 minus session 1, any condition): n {p['n']}, diff {fmt(p['mean'])}, 95% CI [{fmt(p['lo'])}, {fmt(p['hi'])}] -> {verdict(p['lo'], p['hi'], 'later > earlier', 'later < earlier')}") | |
| L.append("") | |
| L.append("## Per-session construct means (analysed participants)") | |
| L.append("") | |
| scm = res["scm"] | |
| scm = scm[scm["participant_id"].isin(res["both"])] | |
| piv = scm.pivot_table(index=["participant_id", "session_id", "condition"], columns="construct", values="mean", aggfunc="mean") | |
| cols = [c for c in [PRIMARY] + list(SECONDARY) if c in piv.columns] | |
| L.append("| participant | session | cond | " + " | ".join(cols) + " |") | |
| L.append("|---|---|---|" + "---|" * len(cols)) | |
| for (p, sid, c), row in piv.iterrows(): | |
| L.append(f"| {p} | {sid} | {c} | " + " | ".join(fmt(row.get(k)) for k in cols) + " |") | |
| L.append("") | |
| return "\n".join(L) | |
| def run(T, args, label, synthetic=False): | |
| both, only = complete_participants(T) | |
| scm = session_construct_means(T, args.scale_max) | |
| n_np = 0 | |
| if getattr(args, "no_prompt_neutral", False): | |
| scm, n_np = apply_no_prompt_neutral(scm, T, (args.scale_max + 1) / 2.0) | |
| res = dict(both=both, only=only, scm=scm, synthetic=synthetic, n_np_filled=n_np) | |
| piv = paired_table(scm, PRIMARY, both) | |
| res["primary"] = paired_block(PRIMARY, piv, args.n_boot, args.seed) | |
| res["secondary"] = [paired_block(c, paired_table(scm, c, both), args.n_boot, args.seed) for c in SECONDARY] | |
| rates = alert_rates(T, both) | |
| res["rates"] = rates | |
| res["rate_block"] = None | |
| if len(rates): | |
| rp = rates.pivot_table(index="participant_id", columns="condition", values="presented_per_h", aggfunc="mean") | |
| for c in ("C", "S"): | |
| if c not in rp.columns: | |
| rp[c] = np.nan | |
| rp = rp.dropna(subset=["C", "S"]) | |
| rp["diff"] = rp["C"] - rp["S"] | |
| res["rate_block"] = paired_block("presented alerts per hour", rp, args.n_boot, args.seed) | |
| res["suppression"] = suppression(T, both) | |
| res["pa"] = per_alert_shares(T, both, args.agree_thr, args.n_boot, args.seed) | |
| res["comparative"] = comparative(T, both, args.n_boot, args.seed) | |
| res["order"] = order_check(scm, piv, T, both, args.n_boot, args.seed) | |
| return render_markdown(T, res, label, args), res | |
| # ------------------------------------------------------------------------------------------------------------------ synthetic | |
| def make_synthetic(out_dir, n=30, seed=SEED, effect=0.0, fmt="T"): | |
| """SYNTHETIC dataset for --dry-run (random numbers, condition effect = `effect` on every construct, default 0). File names | |
| are prefixed synthetic_ and participants are SYN01..; nothing here is real.""" | |
| rng = np.random.default_rng(seed) | |
| os.makedirs(out_dir, exist_ok=True) | |
| P, S, A, Q, M = [], [], [], [], [] | |
| seq_by_p = {} | |
| for i in range(1, n + 1): | |
| pid = f"SYN{i:02d}" | |
| order = "SC" if i % 2 == 1 else "CS" # ABAB from the first participant | |
| P.append(dict(participant_id=pid, order=order, enrolled_date="2026-09-23", car="SYNTHETIC", notes="synthetic dry-run participant")) | |
| seq = 1 | |
| for pos, cond in enumerate(order, start=1): | |
| sid = f"{pid}_{cond}" | |
| c0 = float(rng.uniform(60, 110)) if cond == "C" else 90.0 | |
| end_s = float(rng.uniform(900, 1250)) | |
| t0 = dt.datetime(2026, 9, 23, 9, 0, tzinfo=dt.timezone.utc) + dt.timedelta(days=i, hours=pos) | |
| S.append(dict(participant_id=pid, session_id=sid, condition=cond, order_position=pos, date=t0.date().isoformat(), device=f"{cond}_DEV", | |
| start_utc=t0.isoformat(), end_utc=(t0 + dt.timedelta(seconds=end_s + 30)).isoformat(), count_from_s=round(c0, 1), route_end_s=round(end_s, 1), | |
| loggerd_route=f"0000{i:04x}--synthetic", pred_log_dir="", notes="synthetic")) | |
| latent = {} | |
| for item in EXPECTED_SESSION_ITEMS: | |
| cons = construct_of(item) | |
| mu = 4.5 + (effect if cond == "C" and cons != "annoyance" else (-effect if cond == "C" else 0.0)) | |
| v = int(np.clip(round(rng.normal(mu, 1.3)), 1, 7)) | |
| rev = 1 if (item.startswith("vdl_") and int(item.split("_")[1]) in VDL_REVERSED) else 0 | |
| raw = (8 - v) if rev else v | |
| Q.append(dict(participant_id=pid, session_id=sid, condition=cond, order=order, item_id=item, value=raw, reverse=rev, clip_seq="", free_text="")) | |
| n_alerts = int(rng.poisson(4)) | |
| times = np.sort(rng.uniform(c0, end_s, size=n_alerts)) | |
| for t in times: | |
| if cond == "C": | |
| A.append(dict(participant_id=pid, session_id=sid, condition=cond, source_device="C_DEV", t_rel_s=f"{t:.1f}", kind="cockpitwm_alert", | |
| detail=json.dumps({"regime": "manual", "score": round(float(rng.uniform(0.5, 0.8)), 3), "thr": 0.493, "presentationSource": "cockpitwm", | |
| "effectiveSource": "cockpitwm", "ready": True, "counterfactual": False, "synthetic": True}))) | |
| A.append(dict(participant_id=pid, session_id=sid, condition=cond, source_device="C_DEV", t_rel_s=f"{t + 0.3:.1f}", kind="stock_alert_presented", | |
| detail=json.dumps({"alertType": "driverDistracted2/permanent", "event": "driverDistracted2", "level": 2, "presented_by": "cockpitwm", "cwm_active": True, "synthetic": True}))) | |
| else: | |
| A.append(dict(participant_id=pid, session_id=sid, condition=cond, source_device="S_DEV", t_rel_s=f"{t:.1f}", kind="stock_alert_presented", | |
| detail=json.dumps({"alertType": "driverDistracted1/permanent", "event": "driverDistracted1", "level": 1, "presented_by": "stock", "cwm_active": None, "synthetic": True}))) | |
| M.append(dict(seq=seq, clip_file=f"clip_{seq:03d}.mp4", participant_id=pid, session_id=sid, condition=cond, source_device=f"{cond}_DEV", kind="stock_alert_presented", t_rel_s=f"{t:.1f}")) | |
| for item in PA_ITEMS: | |
| Q.append(dict(participant_id=pid, session_id="", condition="", order=order, item_id=item, value=int(np.clip(round(rng.normal(4.5, 1.5)), 1, 7)), reverse=0, clip_seq=seq, free_text="")) | |
| seq += 1 | |
| if cond == "C": | |
| for _ in range(int(rng.poisson(3))): | |
| t = float(rng.uniform(c0, end_s)) | |
| lvl = "two" if rng.uniform() < 0.8 else "three" | |
| hidden = (lvl == "two") and bool(rng.uniform() < 0.9) | |
| A.append(dict(participant_id=pid, session_id=sid, condition=cond, source_device="C_DEV", t_rel_s=f"{t:.1f}", kind="stock_level_change", | |
| detail=json.dumps({"level": lvl, "prev_level": "one", "level_ge_two": True, "cwm_active": hidden or lvl == "three", "hidden_by_cockpitwm": hidden, "synthetic": True}))) | |
| if cond == "C" and rng.uniform() < 0.3: | |
| t = float(rng.uniform(0, c0)) | |
| A.append(dict(participant_id=pid, session_id=sid, condition=cond, source_device="C_DEV", t_rel_s=f"{t:.1f}", kind="cockpitwm_counterfactual", | |
| detail=json.dumps({"effectiveSource": "openpilot", "ready": False, "counterfactual": True, "synthetic": True}))) | |
| Q.append(dict(participant_id=pid, session_id="", condition="", order=order, item_id="cmp_pref", value=int(rng.choice([0, 1, 2])), reverse=0, clip_seq="", free_text="")) | |
| Q.append(dict(participant_id=pid, session_id="", condition="", order=order, item_id="blind_guess", value=int(rng.choice([0, 1, 2])), reverse=0, clip_seq="", free_text="")) | |
| for pos in (1, 2): | |
| Q.append(dict(participant_id=pid, session_id="", condition="", order=order, item_id=f"cmp_daily_{pos}", value=int(np.clip(round(rng.normal(4.5, 1.3)), 1, 7)), reverse=0, clip_seq="", free_text="")) | |
| Q.append(dict(participant_id=pid, session_id="", condition="", order=order, item_id="cmp_comment", value="", reverse=0, clip_seq="", free_text="synthetic free text")) | |
| paths = {} | |
| tables = OrderedDict([("participants", pd.DataFrame(P)), ("sessions", pd.DataFrame(S)), ("alerts", pd.DataFrame(A)), ("questionnaire", pd.DataFrame(Q)), ("manifest_analyst", pd.DataFrame(M))]) | |
| if fmt == "D": # ../data_schema.md layout (round-trip test of schema_adapter) | |
| Tt = dict(tables) | |
| Tt["manifest"] = tables["manifest_analyst"] | |
| D = SA.tools_to_data_schema(Tt) | |
| tables = OrderedDict([("participants", D["participants"]), ("sessions", D["sessions"]), ("alerts", D["alerts"]), ("questionnaire", D["questionnaire"])]) | |
| for name, df in tables.items(): | |
| p = os.path.join(out_dir, f"synthetic_{name}.csv") | |
| df.to_csv(p, index=False) | |
| paths[name] = p | |
| paths.setdefault("manifest_analyst", None) | |
| with open(os.path.join(out_dir, "SYNTHETIC_README.txt"), "w") as f: | |
| f.write("These CSVs were generated by analysis.py --dry-run (random numbers, seed %d, effect %g). They are NOT study data.\n" % (seed, effect)) | |
| return paths | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| ap.add_argument("--participants") | |
| ap.add_argument("--sessions") | |
| ap.add_argument("--alerts") | |
| ap.add_argument("--questionnaire") | |
| ap.add_argument("--clip-manifest", default=None, help="analyst manifest(s) of render_alert_clips.py (concatenate per participant into one CSV)") | |
| ap.add_argument("--out", default=None, help="write the markdown here as well") | |
| ap.add_argument("--seed", type=int, default=SEED) | |
| ap.add_argument("--n-boot", type=int, default=N_BOOT) | |
| ap.add_argument("--scale-max", type=int, default=SCALE_MAX) | |
| ap.add_argument("--agree-thr", type=int, default=AGREE_THR, help="per-alert rating counted as agree when value >= this (default 5 of 7)") | |
| ap.add_argument("--dry-run", action="store_true", help="generate a labelled SYNTHETIC dataset and run end to end") | |
| ap.add_argument("--dry-run-dir", default=None, help="where the synthetic CSVs go (default: a temp dir)") | |
| ap.add_argument("--dry-run-n", type=int, default=30) | |
| ap.add_argument("--dry-run-effect", type=float, default=0.0, help="synthetic condition effect on the 7-point items (default 0 = null)") | |
| ap.add_argument("--dry-run-format", choices=["T", "D"], default="T", help="layout of the synthetic CSVs: T = tools layout, D = ../data_schema.md layout") | |
| ap.add_argument("--condition-key", default=None, help="condition_key.txt (`X=C`) to resolve data_schema condition_code when sessions.csv `condition` is blank") | |
| ap.add_argument("--blind-codes", action="store_true", help="BLINDED run on codes: X labelled C, Y labelled S; tables marked BLINDED (X - Y); key applied last by the owner") | |
| ap.add_argument("--no-prompt-neutral", action="store_true", help="sensitivity analysis NP-neutral (protocol 11.1): no-prompt sessions' APP/TIM/ANN = neutral midpoint") | |
| args = ap.parse_args(argv) | |
| if args.blind_codes and args.condition_key: | |
| ap.error("--blind-codes and --condition-key are mutually exclusive (blind run first, key last)") | |
| synthetic = False | |
| if args.dry_run: | |
| import tempfile | |
| d = args.dry_run_dir or tempfile.mkdtemp(prefix="user_study_v1_synthetic_") | |
| paths = make_synthetic(d, args.dry_run_n, args.seed, args.dry_run_effect, args.dry_run_format) | |
| args.participants, args.sessions, args.alerts, args.questionnaire, args.clip_manifest = paths["participants"], paths["sessions"], paths["alerts"], paths["questionnaire"], paths["manifest_analyst"] | |
| synthetic = True | |
| label = f"DRY RUN on SYNTHETIC data ({args.dry_run_n} synthetic participants, seed {args.seed}, effect {args.dry_run_effect:g}, layout {args.dry_run_format})" | |
| else: | |
| for k in ("participants", "sessions", "alerts", "questionnaire"): | |
| if not getattr(args, k): | |
| ap.error(f"--{k} is required (or use --dry-run)") | |
| label = f"real data, generated {dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}" | |
| T = load_tables(args.participants, args.sessions, args.alerts, args.questionnaire, args.clip_manifest, args.condition_key, args.blind_codes) | |
| md, _ = run(T, args, label, synthetic) | |
| print(md) | |
| if args.out: | |
| with open(args.out, "w") as f: | |
| f.write(md + "\n") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 48.6 kB
- Xet hash:
- 6e3decda17b69471cd6995b0acd60a3a3249abeae699c7edfcd3ee28b258c0c8
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.