import os.path as osp
from itertools import cycle
import numpy as np
import pandas as pd
import holoviews as hv
import hvplot.pandas
import panel as pn
from holoviews.streams import Stream
from bokeh.palettes import Category10
from rtcog.utils.sync import ActionState
from rtcog.viz.streaming_config import StreamingConfig
from rtcog.viz.plotter import Plotter
[docs]
class ScorePlotter(Plotter):
"""
Live and post-hoc visualization of template matching scores.
This plotter receives template matching scores over time and visualizes them
as streaming line plots. It overlays action-related annotations including hit
thresholds, action windows, cooldown periods, and hit markers indicating the
strongest-matching template at action onset times.
The plot can be rendered dynamically during acquisition or saved as a static
HTML report at the end of the experiment.
"""
data_key = 'scores'
def __init__(self, config: StreamingConfig, streaming=True):
"""
Initialize the ScorePlotter.
Parameters
----------
config : StreamingConfig
Configuration object containing plotting, streaming, and action parameters.
streaming : bool, optional
If True, enables live updating via a HoloViews DynamicMap.
If False, the plotter is used only for static rendering.
"""
super().__init__(config)
self._hit_thr = config.hit_thr
# DataFrame storing scores for each template across time
self._df = pd.DataFrame(np.nan, index=np.arange(self._Nt), columns=self._template_labels)
# Dynamic map for live updates
if streaming:
self.dmap = hv.DynamicMap(self._plot, streams=[Stream.define('Next', t=int)()])
else:
self.dmap = None
# Static overlays accumulated over time
self._polys_static = []
# Gray box before matching starts
self._no_match_poly = self._draw_poly(
0, config.matching_opts.match_start
).opts(color='gray', line_color=None, alpha=0.2)
self._action_state = None
self._last_cooldown_shown = None
self._colors = self._get_template_colors()
self._out_prefix = config.out_prefix
self._out_dir = config.out_dir
[docs]
def update(self, t: int, data: np.ndarray, action_state: ActionState) -> None:
"""
Update the plot with new score data at a given time point.
Parameters
----------
t : int
TR corresponding to the incoming scores.
data : np.ndarray
Array of template matching scores at time `t`.
Must align with the template label order.
action_state : ActionState
Current action state containing action onsets, offsets, and cooldown info.
"""
self._df.iloc[t] = data
self._action_state = action_state
# Trigger redraw only if data is valid
if not np.isnan(self._df.iloc[t]).all():
self.dmap.event(t=t)
def _plot(self, t: int) -> hv.Overlay:
"""
Construct the full plot overlay for a given time point.
This includes:
- Line plots of template scores
- Hit threshold line
- action hit markers
- action, cooldown, and pre-matching shaded regions
Parameters
----------
t : int
Current time index used to update dynamic elements.
Returns
-------
hv.Overlay
Combined HoloViews overlay for rendering.
"""
line_plot = self._df.hvplot.line(
legend='top',
width=1200,
cmap=self._colors,
group_label='Template',
value_label='Score',
)
overlays = [line_plot, self._no_match_poly]
if self._action_state is None:
return hv.Overlay(overlays)
# Threshold line
overlays.append(
hv.HLine(self._hit_thr).opts(
color='black', line_dash='dashed', line_width=1
)
)
# Hit markers
overlays.append(self._draw_hit_markers())
# Action state-dependent dynamic shaded regions
if self._action_state.in_action:
overlays.append(self._draw_dynamic_box(t))
elif self._action_state.action_offsets and t == self._action_state.action_offsets[-1]:
# Final action box
self._polys_static.append(
self._draw_poly(self._action_state.action_onsets[-1], t
).opts(alpha=0.2, color='blue', line_color=None))
elif self._action_state.in_cooldown:
# Cooldown box
if self._action_state.cooldown_end != self._last_cooldown_shown:
self._polys_static.append(
self._draw_poly(self._action_state.action_offsets[-1], self._action_state.cooldown_end)
.opts(alpha=0.2, color='cyan', line_color=None)
)
self._last_cooldown_shown = self._action_state.cooldown_end
overlays.append(
hv.Overlay(self._polys_static)
if self._polys_static else hv.Overlay([])
)
return hv.Overlay(overlays)
def _draw_dynamic_box(self, t: int) -> hv.Polygons:
"""
Draw a dynamic action window box extending to the current time.
Parameters
----------
t : int
Current time index.
Returns
-------
hv.Polygons
Polygon representing the active action window.
"""
return self._draw_poly(self._action_state.action_onsets[-1], t).opts(alpha=0.2, color='blue', )
def _draw_poly(self, start: int, end: int) -> hv.Polygons:
"""
Draw a rectangular polygon spanning a time interval.
Parameters
----------
start : int
Start time index.
end : int
End time index.
Returns
-------
hv.Polygons
Rectangle covering the interval.
"""
end = min(end, self._Nt)
return hv.Polygons([
[(start, -5), (end, -5), (end, 5), (start, 5)]
])
def _draw_hit_markers(self) -> hv.Scatter:
"""
Draw scatter markers at action onset (hit) times.
Each marker is placed at the score of the highest-scoring template
at that time point and colored according to the template identity.
Returns
-------
hv.Scatter
Scatter plot of action hit markers.
"""
points = []
for hit_time in self._action_state.action_onsets:
row = self._df.iloc[hit_time]
if not row.isna().all():
max_template = row.idxmax() # Template with the highest score
max_score = row[max_template] # Highest score value
points.append((hit_time, max_score, max_template))
if points:
df_points = pd.DataFrame(points, columns=["TR", "score", "template"])
df_points['color'] = df_points['template'].map(self._colors).fillna('gray')
return hv.Scatter(df_points, kdims=["TR"], vdims=["score", "template", "color"]).opts(
marker='circle',
alpha=0.5,
size=12,
tools=['hover'],
color='color'
)
else:
return hv.Scatter([], kdims=["TR"], vdims=["score", "template"])
def _get_template_colors(self):
"""
Assign colors to each template label.
Colors are drawn from the Category10 palette and cycled if the number
of templates exceeds the palette size.
Returns
-------
dict
Mapping from template label to color string.
"""
palette = Category10[10]
# Cycle through palette if more labels than colors
assigned_colors = [
c for _, c in zip(self._template_labels, cycle(palette))
]
return dict(zip(self._template_labels, assigned_colors))
[docs]
def close(self):
"""
Save the final state of the plot to an HTML file.
"""
out_html = osp.join(self._out_dir, self._out_prefix + '.dyn_report')
renderer = hv.renderer('bokeh')
# Get last time index with valid data
last_valid_idx = self._df.dropna(how='all').index.max()
final_plot = self._plot(last_valid_idx)
renderer.save(final_plot, out_html)
print(f'++ Score report written to disk: [{out_html}.html]')
[docs]
def render_static(self, df: pd.DataFrame, action_state: ActionState) -> hv.Overlay:
"""
Render a static score plot after the experiment has completed.
Parameters
----------
df : pd.DataFrame
Full score DataFrame indexed by time and template.
action_state : ActionState
Final action state containing all action intervals.
Returns
-------
hv.Overlay
Rendered plot overlay.
"""
self._df = df
self._action_state = action_state
self.close()