""" Study-builder UI (Phase 5) — author a StudyPlan, save it, and launch it. The state transitions are pure functions on the plan dict (`apply_action`) so the builder logic is unit-tested without a browser; the Dash layer is a thin shell on top. `launch_study` runs the authored plan through the runtime stimulus engine (and can record at the same time), closing the loop builder -> live markers. A Qt or React/Ionic front-end can reuse `apply_action` + `launch_study` verbatim; only the rendering differs. """ from __future__ import annotations import time from typing import Callable from .plan import StudyPlan, Block, example_plan # -------------------------------------------------------------------------- # Pure state core (UI-agnostic, fully testable) # -------------------------------------------------------------------------- def apply_action(plan_dict: dict, action: str, payload: dict | None = None) -> dict: """Apply one builder action to a plan dict and return the new plan dict.""" plan = StudyPlan.from_dict(plan_dict) p = payload or {} if action == "add": plan.add(Block(name=p.get("name", f"stim_{len(plan.blocks)+1}"), image=p.get("image"), duration=float(p.get("duration", 2.0)), isi=float(p.get("isi", 1.0)), kind=p.get("kind", "image"), text=p.get("text"))) elif action == "remove": plan.remove(int(p["index"])) elif action == "move_up": plan.move(int(p["index"]), -1) elif action == "move_down": plan.move(int(p["index"]), +1) elif action == "edit": b = plan.blocks[int(p["index"])] for k in ("name", "image", "duration", "isi", "kind", "text"): if k in p: setattr(b, k, float(p[k]) if k in ("duration", "isi") else p[k]) elif action == "settings": for k in ("title", "randomize", "repeats", "seed", "fixation_isi"): if k in p: setattr(plan, k, p[k]) else: raise ValueError(f"unknown action {action!r}") return plan.to_dict() # -------------------------------------------------------------------------- # Launch the authored plan through the live stimulus engine # -------------------------------------------------------------------------- def launch_study(plan: StudyPlan, *, record_seconds: float | None = None, out: str = "session.h5", present: Callable | None = None, extra_sources: list | None = None): """ Run a plan. If `record_seconds` is given, also start a Recorder + any `extra_sources` (real device Sources) so you capture synchronized data while the study presents. Returns the session path (if recording) or the marker log. `present` defaults to headless (no display); pass a PsychoPy presenter for real visuals — see stimulus.study.psychopy_present_factory. """ from ..stimulus.study import StudySource, run_study, headless_present errs = plan.validate() if errs: raise ValueError("invalid study plan: " + "; ".join(errs)) trials = plan.to_trials() if record_seconds is None: return run_study(trials, present=present or headless_present) from ..recorder import Recorder study = StudySource(trials, present=present or headless_present, loop=False) sources = [study] + list(extra_sources or []) for s in sources: s.start() time.sleep(0.3) path = Recorder().record(record_seconds, out) for s in sources: s.stop() return path # -------------------------------------------------------------------------- # Dash UI (thin shell over the pure core) # -------------------------------------------------------------------------- def build_app(initial: StudyPlan | None = None, *, server=None, url_base_pathname=None): # pragma: no cover - needs browser from dash import Dash, dcc, html, Input, Output, State, ALL, ctx plan0 = (initial or example_plan()).to_dict() dash_kw = {} if server is not None: # mount on a shared Flask server dash_kw["server"] = server dash_kw["url_base_pathname"] = url_base_pathname or "/builder/" app = Dash(__name__, **dash_kw) from ..app.ui import THEME_CSS app.index_string = ( '{%metas%}biosync — Study Builder' '{%favicon%}{%css%}' '' f'' '{%app_entry%}') def block_row(i, b): return html.Div([ dcc.Input(value=b["name"], id={"t": "name", "i": i}, style={"width": "16%"}), dcc.Input(value=b.get("image") or "", id={"t": "image", "i": i}, placeholder="image path", style={"width": "34%"}), dcc.Input(value=b["duration"], id={"t": "duration", "i": i}, type="number", style={"width": "12%"}), dcc.Input(value=b["isi"], id={"t": "isi", "i": i}, type="number", style={"width": "12%"}), html.Button("↑", id={"a": "move_up", "i": i}, className="minibtn"), html.Button("↓", id={"a": "move_down", "i": i}, className="minibtn"), html.Button("✕", id={"a": "remove", "i": i}, className="minibtn"), ], className="blk") def render(plan): return [block_row(i, b) for i, b in enumerate(plan["blocks"])] app.layout = html.Div([ # sticky brand bar with Home button html.Div(className="topbar", children=[ html.A("← Home", href="/", className="iconbtn"), html.Span(className="brand", children=[ html.Span("biosync", className="name"), html.Span("Study Builder", className="sub")]), ]), html.Div(className="wrap", children=[ html.Div(className="card", children=[ html.H2("Study settings"), dcc.Store(id="plan", data=plan0), html.Div(style={"display": "flex", "gap": "10px", "alignItems": "center", "flexWrap": "wrap"}, children=[ dcc.Input(id="title", value=plan0["title"], style={"flex": "1", "minWidth": "200px"}), html.Label(className="fld", children=[ dcc.Checklist(["on"], id="rand", value=["on"] if plan0["randomize"] else [], style={"display": "inline-block"}), " randomize"]), html.Label("repeats", className="fld"), dcc.Input(id="repeats", type="number", value=plan0["repeats"], style={"width": "64px"}), ]), ]), html.Div(className="card", children=[ html.H2("Blocks"), html.Div(id="rows", children=render(plan0)), html.Div(style={"marginTop": "10px", "display": "flex", "gap": "10px", "alignItems": "center"}, children=[ html.Button("+ Add block", id="add", className="btn"), html.Button("Save JSON", id="save", className="iconbtn"), html.Span(id="summary", className="statline"), ]), dcc.Download(id="download"), ]), ]), ]) @app.callback(Output("plan", "data"), Output("rows", "children"), Output("summary", "children"), Input("add", "n_clicks"), Input({"a": ALL, "i": ALL}, "n_clicks"), Input("rand", "value"), Input("repeats", "value"), Input("title", "value"), State("plan", "data"), prevent_initial_call=True) def _update(_add, _btns, rand, repeats, title, plan): trig = ctx.triggered_id if trig == "add": plan = apply_action(plan, "add") elif isinstance(trig, dict) and "a" in trig: plan = apply_action(plan, trig["a"], {"index": trig["i"]}) plan = apply_action(plan, "settings", { "title": title, "randomize": bool(rand), "repeats": int(repeats or 1)}) sp = StudyPlan.from_dict(plan) summary = f"{len(sp.blocks)} blocks · ~{sp.estimated_duration():.0f}s per run" return plan, render(plan), summary @app.callback(Output("download", "data"), Input("save", "n_clicks"), State("plan", "data"), prevent_initial_call=True) def _save(_, plan): import json return dict(content=json.dumps(plan, indent=2), filename="study.json") return app def serve(plan_path: str | None = None, host="127.0.0.1", port=8052): # pragma: no cover plan = StudyPlan.from_json(plan_path) if plan_path else example_plan() build_app(plan).run(host=host, port=port)