"""
Multi-stimulus timeline replay (Phase 5).
Loads a recorded session and lets you scrub/play through it: a moving playhead
sweeps every synchronized signal at once, the active stimulus is highlighted, and
the per-stimulus segments are listed. This is the iMotions "replay the timeline"
view — proof, visually, that everything shares one clock.
Two entry points:
* `export_replay_html(session_path, out)` — a SELF-CONTAINED interactive HTML
(Plotly + a vanilla-JS scrubber/play button). No server, opens anywhere.
* `serve_replay(session_path)` — a Dash app version with a live slider.
"""
from __future__ import annotations
import json
import numpy as np
from ..analysis import load as A
PALETTE = ["#4C78A8", "#F58518", "#54A24B", "#E45756", "#72B7B2", "#B279A2"]
# --------------------------------------------------------------------------
def _downsample(t, x, max_points=1500):
"""Uniform decimation to keep the HTML small but faithful."""
n = len(t)
if n <= max_points:
return t, x
idx = np.linspace(0, n - 1, max_points).astype(int)
return t[idx], x[idx]
def replay_data(session: dict, max_points=1500) -> dict:
"""Build a compact, JSON-serializable replay payload from a session."""
t0, t_end = A.common_window(session)
streams = []
for name, s in session["streams"].items():
if s["type"] == "Markers" or not len(s["t"]):
continue
t = np.asarray(s["t"], float) - t0
x = np.asarray(s["x"], float)
x = x[:, 0] if x.ndim > 1 else x
td, xd = _downsample(t, x)
streams.append({"name": name, "type": s["type"], "srate": s["srate"],
"t": [round(float(v), 4) for v in td],
"y": [round(float(v), 4) for v in xd]})
# stimulus segments from on/off marker pairs
segments, open_on = [], {}
for t, label in A.markers(session):
label = str(label)
rel = float(t) - t0
if label.startswith("stim_on"):
open_on[label.split(":", 1)[-1]] = rel
elif label.startswith("stim_off"):
name = label.split(":", 1)[-1]
if name in open_on:
segments.append({"name": name, "start": open_on.pop(name), "end": rel})
return {"duration": float(t_end - t0), "streams": streams, "segments": segments}
# --------------------------------------------------------------------------
def export_replay_html(session_path: str, out_path: str, max_points=1500, *,
back_href: str | None = None, subtitle: str | None = None,
screen=None) -> str:
"""Write a self-contained interactive replay HTML.
`back_href` adds the shared Back/Home bar; `screen=(url, offset)` embeds the
screen recording synced to the playhead (video.currentTime = offset + playhead).
"""
from ..app.ui import THEME_CSS, topbar
data = replay_data(A.load_session(session_path), max_points)
if screen:
data["screen"] = {"url": screen[0], "offset": float(screen[1])}
payload = json.dumps(data)
html = (_REPLAY_TEMPLATE
.replace("/*__DATA__*/", payload)
.replace("/*__THEME__*/", THEME_CSS)
.replace("", topbar(subtitle=subtitle or "Timeline replay",
back_href=back_href)))
with open(out_path, "w") as f:
f.write(html)
return out_path
def serve_replay(session_path: str, host="127.0.0.1", port=8051): # pragma: no cover
"""Dash version with a live time slider."""
from dash import Dash, dcc, html, Input, Output
import plotly.graph_objects as go
from plotly.subplots import make_subplots
data = replay_data(A.load_session(session_path))
app = Dash(__name__)
app.layout = html.Div([
html.H1(f"Replay — {session_path}"),
dcc.Slider(0, data["duration"], value=0, id="playhead",
tooltip={"placement": "bottom"}),
dcc.Graph(id="replay"),
])
@app.callback(Output("replay", "figure"), Input("playhead", "value"))
def _draw(t):
fig = make_subplots(rows=len(data["streams"]), cols=1, shared_xaxes=True,
subplot_titles=[s["name"] for s in data["streams"]])
for i, s in enumerate(data["streams"], 1):
fig.add_trace(go.Scattergl(x=s["t"], y=s["y"], mode="lines",
line=dict(width=1, color=PALETTE[i % len(PALETTE)])),
row=i, col=1)
for seg in data["segments"]:
fig.add_vrect(x0=seg["start"], x1=seg["end"], line_width=0,
fillcolor="rgba(76,120,168,.08)", row="all", col=1)
fig.add_vline(x=t, line=dict(color="#E45756", width=2))
fig.update_layout(height=160 * len(data["streams"]), showlegend=False)
return fig
app.run(host=host, port=port)
# --------------------------------------------------------------------------
_REPLAY_TEMPLATE = r"""
biosync — timeline replay