Buckets:
| #!/usr/bin/env python3 | |
| """parse_questionnaire_pdfs.py -- fielded short questionnaire (user_study_v1, 2026-09-22) PDFs -> data_schema long table. | |
| Fielded form (2 pages): per drive D1/D2 five 7-point items APP1, APP2, APP3, TIM1, ANN1 (+ "cannot_judge"), END1 preference | |
| (first / second / same), END5 free text. Owner rule (2026-09-23): D1 = CockpitWM (C), D2 = stock openpilot (S) for EVERY respondent. | |
| Sources, in order of preference per PDF: | |
| 1. AcroForm field values (raw /V of the radio groups and the END5 text field), read with PyMuPDF at the xref level | |
| (widget.field_value does not return radio-group values for these files); | |
| 2. --manual JSON {name: {"D1.APP1": "5", ..., "END1": "second", "END5": "..."}} for flattened PDFs (hand-drawn circles read | |
| from the rendered page by the analyst; the JSON lives OUTSIDE the repo with the id table). | |
| Names never enter the repo: --ids CSV (outside the repo) maps name -> participant_id (P01, P02, ... assigned in sorted-name order | |
| when missing; existing rows are kept). Outputs (repo, P-ids only): | |
| <out>/questionnaire.csv long table: participant_id, session_index, condition_code, form, item_id, raw_value, scored_value, free_text, entry_date, entered_by | |
| <out>/sessions.csv one row per session: participant_id, session_index, condition_code (X=session 1, Y=session 2), condition (C/S) | |
| <out>/questionnaire_wide.csv one row per participant (P-id, source, D1.*, D2.*, END1, END5) | |
| raw_value: Likert 1-7 as int, empty for cannot_judge / unanswered (free_text = "cannot_judge" or "unanswered"); | |
| END1: first->1, second->2, same->none. scored_value = raw_value (no reverse-scored items in the fielded form; APP3 is positively worded). | |
| """ | |
| import argparse, csv, datetime, glob, json, os, re, sys | |
| ITEMS = ["APP1", "APP2", "APP3", "TIM1", "ANN1"] | |
| FIELDS = [f"D{d}.{i}" for d in (1, 2) for i in ITEMS] + ["END1", "END5"] | |
| END1_MAP = {"first": "1", "second": "2", "same": "none", "none": "none"} | |
| def raw_fields(path): | |
| import fitz # PyMuPDF | |
| doc = fitz.open(path) | |
| out = {} | |
| for xref in range(1, doc.xref_length()): | |
| try: | |
| ft = doc.xref_get_key(xref, "FT") | |
| except Exception: | |
| continue | |
| if ft[0] == "null": | |
| continue | |
| name = doc.xref_get_key(xref, "T")[1].strip("()") | |
| if name == "null" or name not in FIELDS: | |
| continue | |
| v = doc.xref_get_key(xref, "V") | |
| if v[0] == "null": | |
| out[name] = None | |
| elif name == "END5": | |
| s = v[1] | |
| if s.startswith("(") and s.endswith(")"): | |
| s = s[1:-1] | |
| out[name] = s.replace("\\r", " ").replace("\\n", " ").strip() | |
| else: | |
| out[name] = v[1].lstrip("/") | |
| has_widgets = any(k in out for k in FIELDS[:-1]) | |
| filled = any(out.get(k) not in (None, "", "Off") for k in FIELDS[:-1]) | |
| return out, has_widgets, filled | |
| def load_ids(path, names): | |
| rows = {} | |
| if os.path.exists(path): | |
| with open(path, newline="") as f: | |
| for r in csv.DictReader(f): | |
| rows[r["name"]] = r["participant_id"] | |
| used = set(rows.values()) | |
| nxt = 1 | |
| for n in sorted(names): | |
| if n in rows: | |
| continue | |
| while f"P{nxt:02d}" in used: | |
| nxt += 1 | |
| rows[n] = f"P{nxt:02d}"; used.add(rows[n]) | |
| os.makedirs(os.path.dirname(path), exist_ok=True) | |
| with open(path, "w", newline="") as f: | |
| w = csv.writer(f); w.writerow(["name", "participant_id"]) | |
| for n in sorted(rows, key=lambda k: rows[k]): | |
| w.writerow([n, rows[n]]) | |
| return rows | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| ap.add_argument("--pdf-dir", required=True) | |
| ap.add_argument("--ids", required=True, help="name -> participant_id CSV OUTSIDE the repo (created / extended)") | |
| ap.add_argument("--manual", default=None, help="JSON {name: {field: value}} for flattened PDFs (outside the repo)") | |
| ap.add_argument("--out", required=True, help="repo data dir for questionnaire.csv / sessions.csv / questionnaire_wide.csv") | |
| ap.add_argument("--entered-by", default="parse_questionnaire_pdfs.py") | |
| ap.add_argument("--exclude", default="origin", help="comma-separated PDF stems to ignore (blank templates etc.)") | |
| ap.add_argument("--overrides", default=None, help="JSON {name: {field: value, \"_reason\": ...}} applied AFTER parsing (owner corrections from the video record; outside the repo)") | |
| a = ap.parse_args(argv) | |
| manual = json.load(open(a.manual)) if a.manual else {} | |
| overrides = json.load(open(a.overrides)) if a.overrides else {} | |
| excl = {x.strip() for x in a.exclude.split(",") if x.strip()} | |
| pdfs = sorted(p for p in glob.glob(os.path.join(a.pdf_dir, "*.pdf")) if os.path.splitext(os.path.basename(p))[0] not in excl) | |
| names = [os.path.splitext(os.path.basename(p))[0] for p in pdfs] | |
| ids = load_ids(a.ids, names) | |
| today = datetime.date.today().isoformat() | |
| os.makedirs(a.out, exist_ok=True) | |
| long_rows, wide_rows, sess_rows, report = [], [], [], [] | |
| for p, name in zip(pdfs, names): | |
| fields, has_widgets, filled = raw_fields(p) | |
| if name in manual and (not has_widgets or not filled): | |
| fields = dict(manual[name]); source = "manual_visual" | |
| elif has_widgets and filled: | |
| source = "acroform" | |
| elif has_widgets and not filled: | |
| source = "EMPTY_FORM" | |
| else: | |
| source = "NO_FIELDS_NO_MANUAL" | |
| pid = ids[name] | |
| if name in overrides: | |
| for k, v in overrides[name].items(): | |
| if k in FIELDS: | |
| fields[k] = v | |
| source += "+override(" + ",".join(k for k in overrides[name] if k in FIELDS) + ")" | |
| report.append((pid, source, sum(1 for k in FIELDS[:-1] if fields.get(k) not in (None, "", "Off")))) | |
| wide_rows.append([pid, source] + [fields.get(k) or "" for k in FIELDS[:-1]] + [(fields.get("END5") or "").replace("\n", " ")]) | |
| if source in ("EMPTY_FORM", "NO_FIELDS_NO_MANUAL"): | |
| continue | |
| for d, cond in ((1, "C"), (2, "S")): | |
| code = "X" if d == 1 else "Y" | |
| sess_rows.append([pid, d, code, cond]) | |
| for it in ITEMS: | |
| v = fields.get(f"D{d}.{it}") | |
| if v in (None, "", "Off"): | |
| raw, ft = "", "unanswered" | |
| elif v == "cannot_judge": | |
| raw, ft = "", "cannot_judge" | |
| else: | |
| raw, ft = int(v), "" | |
| long_rows.append([pid, d, code, "S", it, "", raw, raw, ft, today, a.entered_by]) | |
| e1 = fields.get("END1") | |
| long_rows.append([pid, "", "", "E", "END1", "", END1_MAP.get(e1, "") if e1 else "", "", "" if e1 else "unanswered", today, a.entered_by]) | |
| long_rows.append([pid, "", "", "E", "END5", "", "", "", (fields.get("END5") or "").replace("\n", " "), today, a.entered_by]) | |
| with open(os.path.join(a.out, "questionnaire.csv"), "w", newline="") as f: | |
| w = csv.writer(f); w.writerow(["participant_id", "session_index", "condition_code", "form", "item_id", "alert_id", "raw_value", "scored_value", "free_text", "entry_date", "entered_by"]) | |
| w.writerows(long_rows) | |
| with open(os.path.join(a.out, "sessions.csv"), "w", newline="") as f: | |
| w = csv.writer(f); w.writerow(["participant_id", "session_index", "condition_code", "condition"]); w.writerows(sess_rows) | |
| with open(os.path.join(a.out, "questionnaire_wide.csv"), "w", newline="") as f: | |
| w = csv.writer(f); w.writerow(["participant_id", "source"] + FIELDS[:-1] + ["END5"]); w.writerows(sorted(wide_rows)) | |
| for pid, src, n in sorted(report): | |
| print(f"{pid} {src:20s} answered_items {n}/11") | |
| print(f"participants with data: {sum(1 for r in report if r[1].startswith(('acroform','manual_visual')))} / {len(report)}; long rows {len(long_rows)}; ids -> {a.ids}") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 8.04 kB
- Xet hash:
- cc70217411b7df903fef62a65c803a79d373099b460146000f6a125b186e281e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.