""" Study plan — the hierarchical experiment model behind the study-builder. Mirrors the iMotions Flow Designer: a study is a tree of `Block`s. A block is either a STIMULUS leaf (image / video / text / survey / website / instructions) or a GROUP that contains child blocks. Each stimulus carries the settings the Stimuli Overview shows — advance mode, duration, record-webcam, track-mouse — and groups carry randomize/repeats so you can shuffle trials within a block. author tree -> save JSON -> to_trials() (flatten) -> run_study / record Plus the per-study `sensors` selection (catalog slugs) and `screen` geometry used by eye-tracking analysis & calibration. Pure data + stdlib; fully testable. Backward compatible: old flat `Block(name,image,duration,isi)` plans still load. """ from __future__ import annotations import json import random from dataclasses import dataclass, asdict, field from ..stimulus.study import Trial @dataclass class Block: """A node in the study flow. Leaf = stimulus; with `children` = group.""" name: str # --- stimulus (leaf) fields --- image: str | None = None # image/video/website source (path or URL) kind: str = "image" # image | video | text | survey | website | instructions text: str | None = None # for kind="text"/"instructions" duration: float = 2.0 # seconds on screen (0 = until advance) isi: float = 1.0 # blank inter-stimulus interval after offset advance: str = "automatic" # automatic | manual | key record_webcam: bool = False # capture FEA webcam during this stimulus track_mouse: bool = False # log mouse position during this stimulus # --- kind-specific payload (survey questions, website url, dwell, ...) --- props: dict = field(default_factory=dict) # --- group fields --- children: list = field(default_factory=list) # list[Block]; non-empty => group randomize: bool = False # shuffle children order repeats: int = 1 # repeat children N times @property def is_group(self) -> bool: return bool(self.children) def validate(self) -> list[str]: errs = [] if not self.name: errs.append("a block has no name") if self.is_group: if self.repeats < 1: errs.append(f"{self.name}: repeats must be >= 1") for c in self.children: errs += c.validate() else: if self.duration < 0: errs.append(f"{self.name}: duration must be >= 0") if self.isi < 0: errs.append(f"{self.name}: isi must be >= 0") if self.advance not in ("automatic", "manual", "key"): errs.append(f"{self.name}: bad advance {self.advance!r}") return errs def leaves(self) -> list["Block"]: """Depth-first list of stimulus leaves under this block.""" if not self.is_group: return [self] out = [] for c in self.children: out += c.leaves() return out def expand(self, rng: random.Random) -> list["Block"]: """Flatten to an ordered list of stimulus leaves, honoring group repeats/randomize.""" if not self.is_group: return [self] out = [] for _ in range(self.repeats): seq = list(self.children) if self.randomize: rng.shuffle(seq) for c in seq: out += c.expand(rng) return out @dataclass class StudyPlan: title: str = "Untitled study" blocks: list = field(default_factory=list) # root-level list[Block] randomize: bool = False # shuffle root order per run repeats: int = 1 # repeat the whole study set seed: int | None = None fixation_isi: float = 0.0 sensors: list = field(default_factory=list) # catalog slugs (Select Sensors) screen: dict = field(default_factory=dict) # {width_px,height_px,width_mm,...} calibration_points: int = 9 # 5 | 9 | 13 # --- editing helpers (pure) ----------------------------------------- def add(self, block: Block, index: int | None = None) -> "StudyPlan": self.blocks.insert(len(self.blocks) if index is None else index, block) return self def remove(self, index: int) -> "StudyPlan": if 0 <= index < len(self.blocks): self.blocks.pop(index) return self def move(self, index: int, delta: int) -> "StudyPlan": j = index + delta if 0 <= index < len(self.blocks) and 0 <= j < len(self.blocks): self.blocks[index], self.blocks[j] = self.blocks[j], self.blocks[index] return self # --- introspection -------------------------------------------------- def validate(self) -> list[str]: errs = [] if not self.blocks: errs.append("study has no blocks") if self.repeats < 1: errs.append("repeats must be >= 1") for b in self.blocks: errs += b.validate() return errs def stimuli_overview(self) -> list[dict]: """Flat table of every stimulus leaf — the Stimuli Overview view.""" rows = [] for b in self.blocks: for leaf in b.leaves(): rows.append({"name": leaf.name, "kind": leaf.kind, "advance": leaf.advance, "duration": leaf.duration, "record_webcam": leaf.record_webcam, "track_mouse": leaf.track_mouse}) return rows def estimated_duration(self) -> float: trials = self.to_trials() return sum(t.duration + t.isi for t in trials) + \ self.fixation_isi * len(trials) # --- bridge to the runtime stimulus engine -------------------------- def to_trials(self) -> list[Trial]: rng = random.Random(self.seed) trials: list[Trial] = [] for _ in range(self.repeats): order = list(self.blocks) if self.randomize: rng.shuffle(order) for b in order: for leaf in b.expand(rng): trials.append(Trial(name=leaf.name, image=leaf.image, duration=leaf.duration, isi=leaf.isi)) return trials # --- persistence ----------------------------------------------------- def to_dict(self) -> dict: return asdict(self) def to_json(self, path: str) -> str: with open(path, "w") as f: json.dump(self.to_dict(), f, indent=2) return path @classmethod def from_dict(cls, d: dict) -> "StudyPlan": return cls( title=d.get("title", "Untitled study"), blocks=[_block_from_dict(b) for b in d.get("blocks", [])], randomize=d.get("randomize", False), repeats=d.get("repeats", 1), seed=d.get("seed"), fixation_isi=d.get("fixation_isi", 0.0), sensors=list(d.get("sensors", [])), screen=dict(d.get("screen", {})), calibration_points=d.get("calibration_points", 9), ) @classmethod def from_json(cls, path: str) -> "StudyPlan": with open(path) as f: return cls.from_dict(json.load(f)) def _block_from_dict(b: dict) -> Block: kids = [_block_from_dict(c) for c in b.get("children", [])] fields = {k: b[k] for k in ("name", "image", "kind", "text", "duration", "isi", "advance", "record_webcam", "track_mouse", "randomize", "repeats", "props") if k in b} return Block(children=kids, **fields) def example_plan() -> StudyPlan: """A nested sample study (groups + trials), like a real Flow Designer study.""" return StudyPlan( title="Ad reactions (sample)", blocks=[ Block("instructions", kind="instructions", text="Look naturally at each image.", duration=4.0, isi=0.5, advance="key"), Block("passiveViewing", randomize=True, repeats=2, children=[ Block("face_A", image="stimuli/face_A.png", duration=2.0, isi=1.0, record_webcam=True), Block("face_B", image="stimuli/face_B.png", duration=2.0, isi=1.0, record_webcam=True), Block("scene_1", image="stimuli/scene_1.png", duration=3.0, isi=1.0), ]), Block("adClip", kind="video", image="stimuli/ad_clip.mp4", duration=5.0, isi=1.5, track_mouse=True), Block("thankYou", kind="text", text="Thank you!", duration=2.0, isi=0.0), ], sensors=["smarteye_aurora", "shimmer_gsr", "fea_webcam"], randomize=False, repeats=1, seed=42, calibration_points=9, )