from abc import ABC, abstractmethod
import numpy as np
from rtcog.utils.sync import ActionState
from rtcog.viz.streaming_config import StreamingConfig
[docs]
class Plotter(ABC):
"""
Abstract base class for real-time fMRI plotters.
Subclasses should implement specific plotting behavior using the `update()` method,
and may optionally override `close()` for cleanup at the end of the experiment.
Attributes
----------
data_key : str
Identifier for the type of data this plotter displays.
"""
data_key: str = None
def __init__(self, config: StreamingConfig):
"""
Initialize the Plotter with basic configuration settings.
Parameters
----------
config : StreamingConfig
Configuration object containing session-level metadata.
"""
self._Nt = config.Nt
self._template_labels = config.template_labels
self._Ntemplates = len(self._template_labels)
[docs]
def should_update(self, t: int, action_state: ActionState) -> bool:
"""
Whether or not the plot should be updated. By default, it will always
update.
Parameters
----------
t : int
Current TR.
action_state : ActionState
The current state of an action block during the experiment.
Returns
----------
bool
Whether the plot should be updated.
"""
return True
[docs]
@abstractmethod
def update(self, t: int, data: np.ndarray, action_state: ActionState) -> None:
"""
Update the plot with new data for the current time repetition (TR).
Parameters
----------
t : int
Current TR.
data : np.ndarray
Data to be visualized (format depends on subclass).
action_state : ActionState
The current state of an action block during the experiment.
"""
pass
[docs]
def close(self):
"""
Optional cleanup at the end of the experiment.
Subclasses can override this method to release resources or finalize output.
"""
pass