""" Unified biosync app shell (Phase 5 packaging). One Flask server that ties the whole tool together into a single application: / landing page — record panel, session list, navigation /builder/ the Study Builder (Dash, mounted here) /view/dashboard the analysis dashboard for a chosen session (rendered report) /view/replay the interactive timeline replay for a chosen session /api/record POST start a synchronized recording (background job) /api/status GET poll the running/last recording /api/sessions GET list recorded sessions in the data directory This is what gets wrapped in a native window (see app/desktop.py) and frozen into a single .app/.exe (see packaging/). It reuses the already-tested report and replay generators, so the shell stays thin. """ from __future__ import annotations import os import threading import time from pathlib import Path from flask import Flask, request, jsonify, send_file, abort # built-in synthetic "test signals" so the app is usable with no hardware. _TEST_SIGNALS = { "gaze": ("sources.synthetic", "SyntheticGaze"), "eda": ("sources.synthetic", "SyntheticEDA"), "ecg": ("sources.synthetic", "SyntheticECG"), "eeg": ("sources.synthetic", "SyntheticEEG"), } class _Recorder: """Runs one recording at a time in a background thread, tracks status.""" def __init__(self, data_dir: Path): self.data_dir = data_dir self.lock = threading.Lock() self.status = {"state": "idle", "session": None, "message": ""} def start(self, seconds, signals, study, participant=None) -> bool: if not self.lock.acquire(blocking=False): return False threading.Thread(target=self._run, args=(seconds, signals, study, participant), daemon=True).start() return True def _run(self, seconds, signals, study, participant=None): import importlib from ..recorder import Recorder # which stream names we expect, so we can flag any that fail to resolve _EXPECT = {"gaze": "EyeTracker", "eda": "GSR", "ecg": "ECG", "eeg": "EEG"} try: self.status = {"state": "recording", "session": None, "message": f"recording {seconds:.0f}s"} sources = [] expect = set() for key in signals: mod, cls = _TEST_SIGNALS[key] Cls = getattr(importlib.import_module(f"biosync.{mod}"), cls) sources.append(Cls()) if key in _EXPECT: expect.add(_EXPECT[key]) if study: from ..studybuilder import StudyPlan from ..stimulus.study import StudySource plan_path = self.data_dir / "study.json" plan = (StudyPlan.from_json(str(plan_path)) if plan_path.exists() else __import__("biosync.studybuilder", fromlist=["example_plan"]).example_plan()) sources.append(StudySource(plan.to_trials(), loop=True)) expect.add("Markers") for s in sources: s.start() time.sleep(0.4) fname = time.strftime("session_%Y%m%d_%H%M%S.h5") rec = Recorder(autosave_every=5.0) rec.record(seconds, str(self.data_dir / fname), expect=expect, settle=2.0) self.last_summary = rec.summary for s in sources: s.stop() # tag the session to a participant in the project manifest if participant and participant.get("id"): from ..participants import Project Project(str(self.data_dir)).attach_session( participant["id"], fname, name=participant.get("name", ""), group=participant.get("group", "")) summ = getattr(self, "last_summary", {}) or {} warn = "" if summ.get("missing"): warn = " ⚠ missing: " + ", ".join(summ["missing"]) elif summ.get("empty_streams"): warn = " ⚠ no data: " + ", ".join(summ["empty_streams"]) self.status = {"state": "done", "session": fname, "message": f"saved {fname}{warn}", "summary": summ} except Exception as e: # surface errors to the UI self.status = {"state": "error", "session": None, "message": str(e)} finally: self.lock.release() def create_app(data_dir: str = "sessions") -> Flask: data = Path(data_dir) data.mkdir(parents=True, exist_ok=True) app = Flask(__name__) rec = _Recorder(data) from ..licensing import LicenseManager lic = LicenseManager(str(data)) @app.route("/license") def license_page(): return _license_html() @app.route("/api/license/status") def license_status(): return jsonify(lic.status()) @app.route("/api/license/activate", methods=["POST"]) def license_activate(): body = request.get_json(force=True, silent=True) or {} return jsonify(lic.activate(body.get("key", ""))) @app.route("/api/license/deactivate", methods=["POST"]) def license_deactivate(): lic.deactivate() return jsonify({"ok": True}) # mount the Study Builder Dash app at /builder/ from ..studybuilder.builder_app import build_app as build_builder build_builder(server=app, url_base_pathname="/builder/") @app.route("/") def index(): return _landing_html() @app.route("/api/sessions") def sessions(): files = sorted((f.name for f in data.glob("*.h5")), reverse=True) return jsonify(files) @app.route("/api/record", methods=["POST"]) def record(): body = request.get_json(force=True, silent=True) or {} seconds = float(body.get("seconds", 20)) signals = [s for s in body.get("signals", ["gaze", "eda", "ecg"]) if s in _TEST_SIGNALS] study = bool(body.get("study", True)) if not lic.allowed(): return jsonify({"started": False, "licensed": False, "error": "Your trial has ended. Enter a license key to keep recording.", "license": lic.status()}), 402 ok = rec.start(seconds, signals, study, participant=body.get("participant")) return jsonify({"started": ok, "status": rec.status}) @app.route("/api/status") def status(): return jsonify(rec.status) @app.route("/participants") def participants_page(): return _participants_html() @app.route("/api/participants") def api_participants(): from ..participants import Project return jsonify(Project(str(data)).to_dict()) @app.route("/api/export.") def api_export(fmt): from ..participants import Project from ..analysis import export as EX proj = Project(str(data)) if not proj.all(): abort(404, "no participants to export yet") if fmt == "xlsx": out = data / "biosync_export.xlsx" EX.to_excel(proj, str(out)) return send_file(str(out), as_attachment=True, download_name="biosync_export.xlsx") if fmt == "zip": out = data / "biosync_export.zip" EX.to_csv_zip(proj, str(out)) return send_file(str(out), as_attachment=True, download_name="biosync_export_csv.zip") abort(404, "use export.xlsx or export.zip") @app.route("/view/aggregate") def view_aggregate(): from ..participants import Project from ..dashboard import export_group_html group = request.args.get("group") sess = Project(str(data)).sessions(group=group if group != "(all)" else None) if not sess: abort(404, "no sessions for this group yet") out = data / "_aggregate_view.html" export_group_html(sess, str(out), back_href="/participants", subtitle=f"Group: {group or 'all'} · n={len(sess)}") return send_file(str(out)) from .live import LiveGaze live = LiveGaze() from .monitor import LiveMonitor monitor = LiveMonitor() @app.route("/monitor") def monitor_page(): return _monitor_html() @app.route("/api/monitor/start", methods=["POST"]) def monitor_start(): return jsonify(monitor.start()) @app.route("/api/monitor/status") def monitor_status(): return jsonify(monitor.quality()) @app.route("/api/monitor/stop", methods=["POST"]) def monitor_stop(): return jsonify(monitor.stop()) @app.route("/validate") def validate_page(): return _validate_html() @app.route("/api/validate", methods=["POST"]) def api_validate(): import json as _json from .. import validation as V from ..analysis import load as A body = request.get_json(force=True, silent=True) or {} sess_path = _safe_session(data, body.get("session")) device = body.get("device", "") session = A.load_session(str(sess_path)) type_mod = {"Gaze": "gaze", "ECG": "ecg", "EEG": "eeg", "GSR": "eda", "EMG": "emg", "Respiration": "rsp", "NIRS": "fnirs"} reports = [] for name, s in session["streams"].items(): if s["type"] == "Markers": continue mod = type_mod.get(s["type"], "generic") rep = V.validate_stream(session, device=device or name, stream=name, modality=mod, min_channels=1) reports.append(rep.dict()) overall = bool(reports) and all(r["passed"] for r in reports) out = _jsonable({"session": sess_path.name, "overall": overall, "reports": reports}) _json.dump(out, open(data / "validation_report.json", "w"), indent=2) return jsonify(out) from ..sources.webcam_eye import WebcamEyeSource webcam = {"src": None} @app.route("/webcam") def webcam_page(): return _webcam_html() @app.route("/api/webcam/start", methods=["POST"]) def webcam_start(): if webcam["src"] is None: webcam["src"] = WebcamEyeSource() webcam["src"].start() return jsonify({"running": True}) @app.route("/api/webcam/gaze", methods=["POST"]) def webcam_gaze(): body = request.get_json(force=True, silent=True) or {} src = webcam["src"] if src is None: return jsonify({"ok": False, "error": "webcam source not started"}), 400 if "samples" in body: n = src.push_batch(body["samples"]) else: src.push(body.get("x", 0.5), body.get("y", 0.5), body.get("conf", 1.0)) n = 1 return jsonify({"ok": True, "received": n}) @app.route("/api/webcam/stop", methods=["POST"]) def webcam_stop(): if webcam["src"] is not None: webcam["src"].stop() webcam["src"] = None return jsonify({"running": False}) @app.route("/screen") def screen_page(): return _screen_html() @app.route("/api/screen/start", methods=["POST"]) def screen_start(): from pylsl import local_clock return jsonify({"t0": local_clock()}) # video start on the shared LSL clock @app.route("/api/screen/upload", methods=["POST"]) def screen_upload(): import json as _json f = request.files.get("video") if f is None: return jsonify({"saved": False, "error": "no video"}), 400 t0 = float(request.form.get("t0", 0)) stem = _latest_session_stem(data, request.form.get("session")) if not stem: return jsonify({"saved": False, "error": "no session to attach to"}), 400 sdir = data / "screens"; sdir.mkdir(exist_ok=True) f.save(str(sdir / f"{stem}.webm")) _json.dump({"t0_lsl": t0}, open(sdir / f"{stem}.screen.json", "w")) return jsonify({"saved": True, "session": stem + ".h5"}) @app.route("/media/screen/") def media_screen(name): p = (data / "screens" / name).resolve() if p.parent != (data / "screens").resolve() or not p.exists(): abort(404) return send_file(str(p)) @app.route("/calibrate") def calibrate_page(): return _calibrate_html() @app.route("/gaze") def gaze_page(): return _gaze_html() @app.route("/api/gaze/start", methods=["POST"]) def gaze_start(): body = request.get_json(force=True, silent=True) or {} slug = body.get("slug") or None try: return jsonify(live.start(slug)) except Exception as e: return jsonify({"running": False, "error": str(e)}), 500 @app.route("/api/gaze/stop", methods=["POST"]) def gaze_stop(): return jsonify(live.stop()) @app.route("/api/gaze/latest") def gaze_latest(): return jsonify(live.latest) @app.route("/api/gaze/stream") def gaze_stream(): from flask import Response def gen(): import json as _json while live.running: yield f"data: {_json.dumps(live.latest)}\n\n" time.sleep(1 / 60) return Response(gen(), mimetype="text/event-stream") @app.route("/api/calibration/evaluate", methods=["POST"]) def calibration_eval(): from ..analysis.calibration import evaluate from ..analysis.eyetracking import Screen import numpy as np body = request.get_json(force=True, silent=True) or {} pts = {} for key, samples in (body.get("points") or {}).items(): tx, ty = (float(v) for v in key.split(",")) pts[(tx, ty)] = np.array(samples, dtype=float) screen = Screen(**(body.get("screen") or {})) res = evaluate(pts, screen=screen, threshold_deg=float(body.get("threshold", 1.0))) return jsonify({**res.summary(), "per_point": [{"target": list(p.target), "accuracy_deg": round(p.accuracy_deg, 3) if p.accuracy_deg == p.accuracy_deg else None, "precision_deg": round(p.precision_deg, 3) if p.precision_deg == p.precision_deg else None, "n": p.n_samples, "valid": p.valid} for p in res.points]}) from .present import Presenter, save_response, load_responses, flatten_trials presenter = Presenter() @app.route("/present") def present_page(): return _present_html() @app.route("/api/present/start", methods=["POST"]) def present_start(): from ..studybuilder import StudyPlan, example_plan body = request.get_json(force=True, silent=True) or {} presenter.start(body.get("participant")) p = _study_path(data) plan = StudyPlan.from_json(str(p)) if p.exists() else example_plan() return jsonify({"running": True, "trials": flatten_trials(plan)}) @app.route("/api/present/marker", methods=["POST"]) def present_marker(): body = request.get_json(force=True, silent=True) or {} presenter.marker(body.get("label", "")) return jsonify({"ok": True}) @app.route("/api/present/respond", methods=["POST"]) def present_respond(): body = request.get_json(force=True, silent=True) or {} save_response(str(data), (body.get("participant") or {}).get("id", ""), body.get("stimulus", ""), body.get("kind", "survey"), body.get("data", {})) return jsonify({"saved": True}) @app.route("/api/present/stop", methods=["POST"]) def present_stop(): return jsonify(presenter.stop()) @app.route("/api/responses") def api_responses(): return jsonify(load_responses(str(data))) @app.route("/flow") def flow_page(): return _flow_html() @app.route("/api/stimuli/upload", methods=["POST"]) def stimuli_upload(): import re as _re f = request.files.get("file") if f is None: return jsonify({"ok": False, "error": "no file"}), 400 sdir = data / "stimuli"; sdir.mkdir(exist_ok=True) safe = _re.sub(r"[^A-Za-z0-9._-]", "_", f.filename or "stimulus") f.save(str(sdir / safe)) return jsonify({"ok": True, "name": safe, "url": f"/media/stimuli/{safe}"}) @app.route("/media/stimuli/") def media_stimuli(name): p = (data / "stimuli" / name).resolve() if p.parent != (data / "stimuli").resolve() or not p.exists(): abort(404) return send_file(str(p)) @app.route("/aoi-editor") def aoi_editor_page(): return _aoi_editor_html() @app.route("/api/aoi", methods=["GET", "POST"]) def api_aoi(): import json as _json p = data / "aois.json" if request.method == "POST": body = request.get_json(force=True, silent=True) or {} _json.dump(body, open(p, "w"), indent=2) n = sum(len(v) for v in body.values()) return jsonify({"saved": True, "aois": n}) return jsonify(_json.load(open(p)) if p.exists() else {}) @app.route("/api/video-stimuli") def api_video_stimuli(): # list video stimuli in the current study (name -> media url) from ..studybuilder import StudyPlan, example_plan p = _study_path(data) plan = StudyPlan.from_json(str(p)) if p.exists() else example_plan() out = [] for b in plan.blocks: for leaf in b.leaves(): if leaf.kind == "video" and leaf.image: out.append({"name": leaf.name, "url": leaf.image, "duration": leaf.duration}) return jsonify(out) @app.route("/api/study", methods=["GET", "POST"]) def api_study(): from ..studybuilder import StudyPlan, example_plan p = _study_path(data) if request.method == "POST": body = request.get_json(force=True, silent=True) or {} plan = StudyPlan.from_dict(body) # validate by round-trip errs = plan.validate() plan.to_json(str(p)) return jsonify({"saved": True, "errors": errs, "trials": len(plan.to_trials()), "duration": round(plan.estimated_duration(), 1)}) plan = StudyPlan.from_json(str(p)) if p.exists() else example_plan() return jsonify(plan.to_dict()) @app.route("/sensors") def sensors_page(): return _sensors_html(_current_sensors(data)) @app.route("/api/sensors", methods=["GET", "POST"]) def api_sensors(): if request.method == "POST": body = request.get_json(force=True, silent=True) or {} sel = [s for s in body.get("sensors", []) if s in _ALL_SLUGS] _save_sensors(data, sel) return jsonify({"saved": True, "sensors": sel}) return jsonify(_current_sensors(data)) @app.route("/view/dashboard") def view_dashboard(): sess = _safe_session(data, request.args.get("session")) from ..dashboard import export_html out = data / "_dashboard_view.html" export_html(str(sess), str(out), back_href="/", subtitle=sess.name) return send_file(str(out)) @app.route("/view/overlay") def view_overlay(): from ..studybuilder import StudyPlan, example_plan from ..analysis import load as A, overlay as OV from .ui import THEME_CSS, topbar sess = _safe_session(data, request.args.get("session")) p = _study_path(data) plan = StudyPlan.from_json(str(p)) if p.exists() else example_plan() imgs = OV.stimulus_images(plan, str(data / "stimuli")) session = A.load_session(str(sess)) cards, first = [], True for name, ipath in imgs.items(): gx, gy = OV.gaze_for_stimulus(session, name) if not len(gx): continue for fig in (OV.heatmap_over_image(session, name, ipath), OV.scanpath_over_image(session, name, ipath)): cards.append('
' + fig.to_html( full_html=False, include_plotlyjs=("cdn" if first else False), config={"displayModeBar": False}) + "
") first = False if not cards: cards = ['

No gaze-on-image data yet. ' 'Upload image stimuli in the Flow Designer, then record a session ' 'so gaze is captured while those images are shown.

'] head = (f'' f'' f'biosync — gaze maps' f'' f'') return (head + topbar(subtitle=f"Gaze maps · {sess.name}", back_href="/") + '
' + "\n".join(cards) + "
") @app.route("/view/replay") def view_replay(): sess = _safe_session(data, request.args.get("session")) from ..studybuilder.replay import export_replay_html out = data / "_replay_view.html" screen = _screen_for(data, sess) export_replay_html(str(sess), str(out), back_href="/", subtitle=sess.name, screen=screen) return send_file(str(out)) return app def _study_path(data: Path) -> Path: return data / "study.json" def _current_sensors(data: Path) -> list: p = _study_path(data) if p.exists(): import json try: return list(json.load(open(p)).get("sensors", [])) except Exception: return [] return [] def _save_sensors(data: Path, sensors: list): from ..studybuilder import StudyPlan p = _study_path(data) plan = StudyPlan.from_json(str(p)) if p.exists() else StudyPlan() plan.sensors = sensors plan.to_json(str(p)) def _sensor_groups() -> dict: """Group the device catalog by modality for the Select-Sensors screen.""" from ..drivers.catalog import DEVICES groups: dict = {} for d in DEVICES.values(): groups.setdefault(d.modality, []).append(d) for v in groups.values(): v.sort(key=lambda d: (d.brand, d.model)) return groups from ..drivers.catalog import DEVICES as _CATALOG _ALL_SLUGS = set(_CATALOG) _MOD_LABEL = {"gaze": "Eye tracking", "eeg": "EEG", "eda": "EDA / GSR", "ecg": "ECG", "emg": "EMG", "rsp": "Respiration", "fnirs": "fNIRS", "fea": "Facial expression (AFFDEX → py-feat)", "voice": "Voice"} def _sensors_html(selected: list) -> str: from .ui import THEME_CSS, topbar groups = _sensor_groups() sel = set(selected) # eye trackers render as a dropdown (like iMotions); everything else as cards gaze = groups.pop("gaze", []) opts = "".join( f'' for d in gaze) eye = (f'

Eye tracking

' f'' f'

' f'{len(gaze)} supported trackers. LSL-tier models need no extra driver.

') cards = [] order = ["fea", "voice", "eeg", "eda", "ecg", "emg", "rsp", "fnirs"] for mod in order: ds = groups.get(mod) if not ds: continue items = "".join( f'' for d in ds) cards.append(f'

{_MOD_LABEL.get(mod, mod)}

' f'
{items}
') return f""" biosync — sensors {topbar(subtitle="Select sensors", back_href="/")}

Select sensors

Which sensors collect data for this study? Powered by the {len(_ALL_SLUGS)}-device catalog.

{eye} {''.join(cards)}
""" def _flow_html() -> str: from .ui import THEME_CSS, topbar from .onboarding import tour, help_button flow_tour = tour("flow", [ {"sel": "#palettecard", "title": "Add stimuli", "body": "Click a type to add it, or drag it onto the flow. You can also drag an image/video file from your computer right onto the canvas."}, {"sel": "#root", "title": "Build the flow", "body": "Reorder by dragging. Drop a stimulus onto a Block to nest it. Use the gear on each item to set duration, advance, webcam, mouse, or upload a file."}, {"sel": "#save", "title": "Save & run", "body": "Save your flow, then Save & record to run it. Sensors, calibration and recording all read this same study."}, ]) return (r""" biosync — Flow Designer """ + topbar(subtitle="Flow Designer", back_href="/", extra=help_button()) + r"""

Add a stimulus — click a type, or drag it onto the flow

▣Block
▲Image
►Video
☰Text
☑Survey
⚛Website
ⓘInstructions

Tip: you can also drag an image or video file from your computer straight onto the flow to add it as a stimulus.

__TOUR__ """).replace("__THEME__", THEME_CSS).replace("__TOUR__", flow_tour) def _present_html() -> str: from .ui import THEME_CSS return r""" biosync — present ← Exit

Present study

Walks through the current study's stimuli and pushes markers. Run a recording at the same time to capture event-locked data.

""".replace("__THEME__", THEME_CSS) def _screen_html() -> str: from .ui import THEME_CSS, topbar return (r""" biosync — screen recording """ + topbar(subtitle="Screen recording", back_href="/") + r"""

Screen recording

Capture the screen during a session. The video is timestamped on the shared clock and attached to your latest recording, so the Timeline Replay shows what the participant saw, synced to the gaze and signals.

""").replace("__THEME__", THEME_CSS) def _webcam_html() -> str: from .ui import THEME_CSS return r""" biosync — webcam eye tracking ← Home
webcam 0/s · idle

Webcam eye tracking

Uses your laptop camera (WebGazer). Calibrate by clicking each dot while looking at it; gaze then streams into biosync as a Gaze source — record it, build heatmaps/AOIs, exactly like a hardware tracker.

""".replace("__THEME__", THEME_CSS) def _validate_html() -> str: from .ui import THEME_CSS, topbar return (r""" biosync — hardware validation """ + topbar(subtitle="Hardware validation", back_href="/") + r"""

Validate a device

Connect a device so it streams (built-in test signals work too), give it a name, then run a short recording and an automatic check: stream present, channels, sample rate, signal quality, and a known-signal test per modality (gaze validity/accuracy, plausible HR, EEG band power, EDA range). The result is saved as a validation report.

""").replace("__THEME__", THEME_CSS) def _aoi_editor_html() -> str: from .ui import THEME_CSS, topbar return (r""" biosync — AOI editor """ + topbar(subtitle="Dynamic AOI editor", back_href="/") + r"""

Dynamic AOI editor

Pick a video stimulus, scrub to a moment, drag a box over the object, and Add keyframe. Repeat at later moments — the AOI moves between keyframes automatically. Scrub to preview. Save to use it in AOI metrics.

0.00s

AOIs on this stimulus

""").replace("__THEME__", THEME_CSS) def _license_html() -> str: from .ui import THEME_CSS, topbar portal = os.environ.get("BIOSYNC_PORTAL", "https://account.biosync-lab.co.za") return (r""" biosync — license """ + topbar(subtitle="License", back_href="/") + r"""

License status

checking…

Activate a license key

Paste the key from your account. It is verified securely on this computer — no internet required.

Need a license?

Licenses are issued with an annual subscription. Create an account, subscribe, and your key appears in your account dashboard.

Get a license →
""").replace("__THEME__", THEME_CSS).replace("__PORTAL__", portal) def _monitor_html() -> str: from .ui import THEME_CSS, topbar return (r""" biosync — signal quality """ + topbar(subtitle="Signal quality", back_href="/") + r"""

Live signal quality

Start your sensors (or a recording), then watch each stream's health here. Green = good, amber = warning, red = bad (flatline, low rate, clipping, or poor gaze validity).

""").replace("__THEME__", THEME_CSS) def _participants_html() -> str: from .ui import THEME_CSS, topbar return (r""" biosync — participants """ + topbar(subtitle="Participants", back_href="/") + r"""

Groups

Participants

IDNameGroupSessions

Statistical export

Tidy tables for R / Python / SPSS / Excel: per-participant (wide), AOI + events (long), group summary.

""").replace("__THEME__", THEME_CSS) def _calibrate_html() -> str: from .ui import THEME_CSS return r""" biosync — calibration ← Home

12-point calibration

Fix your eyes on each dot as it appears. Uses the live gaze source (synthetic by default; pick a tracker on the Sensors screen).

ready
""".replace("__THEME__", THEME_CSS) def _gaze_html() -> str: from .ui import THEME_CSS return r""" biosync — live gaze ← Home
gaze 0 Hz · —
gaze lost
""".replace("__THEME__", THEME_CSS) def _jsonable(obj): """Recursively coerce numpy scalars/arrays to native types for jsonify.""" import numpy as np if isinstance(obj, dict): return {k: _jsonable(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_jsonable(v) for v in obj] if isinstance(obj, np.integer): return int(obj) if isinstance(obj, np.floating): return float(obj) if isinstance(obj, (np.bool_,)): return bool(obj) if isinstance(obj, np.ndarray): return _jsonable(obj.tolist()) return obj def _latest_session_stem(data: Path, name: str | None) -> str | None: if name: return name[:-3] if name.endswith(".h5") else name files = sorted(data.glob("*.h5"), key=lambda p: p.stat().st_mtime) return files[-1].stem if files else None def _screen_for(data: Path, session_path: Path): """Return (video_url, offset_s) if a screen recording exists for this session. offset maps the replay time axis (0 = first overlapping sample) to video time: video.currentTime = offset + playhead. """ import json as _json from ..analysis import load as A stem = session_path.stem side = data / "screens" / f"{stem}.screen.json" vid = data / "screens" / f"{stem}.webm" if not (side.exists() and vid.exists()): return None try: t0_video = float(_json.load(open(side))["t0_lsl"]) t0_cw = A.common_window(A.load_session(str(session_path)))[0] return (f"/media/screen/{stem}.webm", t0_cw - t0_video) except Exception: return None def _safe_session(data: Path, name: str | None) -> Path: if not name: files = sorted(data.glob("*.h5")) if not files: abort(404, "no sessions recorded yet") return files[-1] p = (data / name).resolve() if p.parent != data.resolve() or not p.exists(): abort(404, "session not found") return p def _landing_html() -> str: from .ui import THEME_CSS, topbar from .onboarding import tour, help_button landing_tour = tour("home", [ {"sel": "#getstarted", "title": "Welcome to biosync", "body": "Your home base. These six numbered steps are the full study flow — from choosing sensors to exporting stats. This tour points out every feature; you can replay it anytime with the '?' top-right."}, {"sel": "#reccard", "title": "Record a session", "body": "Pick signals, enter a Participant ID + group, set a duration, and hit Record. Built-in test signals work with NO hardware, so you can try the whole app today."}, {"sel": 'a[href="/sensors"]', "title": "1 · Select Sensors", "body": "Choose which devices collect data — eye trackers, EEG, GSR, ECG and more, from a 42-device catalog. Your choice is saved with the study."}, {"sel": 'a[href="/calibrate"]', "title": "2 · Calibrate", "body": "Run a 12-point eye-tracker calibration and get accuracy + precision in degrees, with a PASS/FAIL gate — before you record."}, {"sel": 'a[href="/gaze"]', "title": "Live Gaze", "body": "A real-time gaze cursor so you can confirm the eye tracker is following the eyes before a study."}, {"sel": 'a[href="/webcam"]', "title": "Webcam Eye Tracking", "body": "No eye tracker? Estimate gaze from the laptop camera. It streams into biosync exactly like a hardware tracker."}, {"sel": 'a[href="/flow"]', "title": "3 · Flow Designer", "body": "Build the study by dragging (or clicking) blocks and stimuli — images, video, text, surveys, websites. Upload your own files, nest blocks, set timing."}, {"sel": 'a[href="/present"]', "title": "4 · Present & record", "body": "Run the study for a participant: it shows each stimulus and pushes markers so everything is event-locked to the sensor data."}, {"sel": 'a[href="/monitor"]', "title": "Signal Quality", "body": "Watch each sensor's health live while recording — green/amber/red for rate, gaze validity, flatline and clipping. Catch a bad electrode before the session is wasted."}, {"sel": 'a[href="/validate"]', "title": "Validate Hardware", "body": "Before a study, run an automatic device check: stream present, sample rate, signal quality, and a known-signal test per modality. Saves a pass/fail report."}, {"sel": 'a[href="/screen"]', "title": "Screen Recording", "body": "Capture the screen during a session; it's timestamped and shown in the Replay, synced to the gaze and signals."}, {"sel": "#dash", "title": "5 · Dashboard", "body": "Per-session analysis: physiology (HR/HRV, EDA), EEG/ERP, eye-movement metrics, gaze heatmap & scanpath, AOI, and an affect circumplex."}, {"sel": "#replay", "title": "Timeline Replay", "body": "Scrub or play back the whole session — every signal moves together with the stimulus markers and the screen recording."}, {"sel": "#gazemaps", "title": "Gaze Maps", "body": "Heatmap and scanpath drawn on the ACTUAL stimulus image the participant saw — the credibility view for eye tracking."}, {"sel": 'a[href="/aoi-editor"]', "title": "AOI Editor", "body": "Draw Areas of Interest on a video and keyframe them so they move with the object. Dwell, TTFF and coverage then track the moving AOI."}, {"sel": 'a[href="/participants"]', "title": "5–6 · Participants, groups & export", "body": "Every recording is tagged to a participant and group. Open a group for an aggregate dashboard, and download tidy CSV (R-ready) or Excel for stats."}, {"sel": "#sesscard", "title": "Your sessions", "body": "Recordings appear here — open the Dashboard, Replay or Gaze Maps for any of them. That's the whole platform. Press '?' anytime to see this tour again."}, ]) return f""" biosync {topbar(subtitle="acquisition · sync · analysis", extra=help_button())}

Human-behavior research platform

Synchronize eye tracking, physiology, EEG/fNIRS, facial & voice on one clock — record, analyze event-locked, replay. An open iMotions-class lab tool.

Record a session

ready

Built-in test signals run with no hardware. Real devices are added via drivers.make(slug) — see HARDWARE.md.

Open

Sessions

No sessions yet — record one above.
{landing_tour} """