TaskBase

class TaskBase

Bases: object

Base class for defining behavioral tasks.

Subclass TaskBase to implement your own task. You must override four methods and may optionally call helper methods and read certain attributes during execution.

────────────────────────────────────────────────────────── METHODS YOU MUST OVERRIDE ──────────────────────────────────────────────────────────

start(self) -> None

Called once before the trial loop begins. Use it to configure hardware, pre-compute stimuli, open files, etc.

create_trial(self) -> None

Called at the beginning of every trial. Build and send the Bpod state machine here (or set up stimuli for non-Bpod controllers). The state machine runs to completion before after_trial is called.

after_trial(self) -> None

Called immediately after each trial finishes. Use it to score the animal’s performance, update adaptive parameters, and call register_value to save custom columns to the data file. self.trial_data is already populated at this point.

close(self) -> None

Called once after the trial loop ends (session finished, forced stop, or error). Use it to close files, stop hardware, release resources.

────────────────────────────────────────────────────────── METHODS YOU CAN CALL (do not override) ──────────────────────────────────────────────────────────

register_value(name: str, value: Any) -> None

Saves a custom value to the current trial row in the CSV. Call this inside after_trial.

Example:

self.register_value("correct", 1)
self.register_value("response_time", 0.432)

execute_function(self, i: int) -> None:

Executes a registered function.

Args:

i (int): The function index (1-99).

Example::

self.execute_function(1) # executes the function registered at index 1

────────────────────────────────────────────────────────── ATTRIBUTES YOU CAN READ INSIDE YOUR TASK ──────────────────────────────────────────────────────────

self.trial_datadict

Dictionary populated automatically after each trial with Bpod event timestamps and any values previously registered via register_value. Available inside after_trial to drive adaptive logic.

Typical keys: "trial", "subject", "task", "date", "system_name", plus every key you have registered with register_value.

Example:

def after_trial(self):
    if self.trial_data.get("correct") == 1:
        self.settings.difficulty += 1

self.current_trialint

1-based index of the trial that just ran (available in both create_trial and after_trial).

self.subjectstr

Name of the subject running the session.

self.settingsSettings

Object that holds all the session parameters defined in the training protocol. Read and write its attributes to implement adaptive training.

self.calibrationsCalibrations

Object that holds the task’s calibrations. Call its methods to convert between hardware values and real-world units.

self.bpodBpodController

Low-level Bpod interface (state machine construction, sending, etc.). Primarily used inside create_trial.

self.arduinoArduinoController

Arduino interface for tasks that use an Arduino instead of Bpod.

self.cam_boxCamera | NullCamera

Camera attached to the behaviour box (NullCamera if not configured).

__init__() → None

Methods

start() → None

Starts the task. Must be overridden by subclasses.

create_trial() → None

Creates the state machine for the current trial. Must be overridden.

after_trial() → None

Executed after each trial completes. Must be overridden.

close() → None

Closes the task and releases resources. Must be overridden.

register_start_trial(raspberry_timestamp: float, controller_timestamp: float) → None

Registers the start of a trial. We use 2 timestamps because we use the start of the trial to synchronize the clocks between the raspberry and the controller. If you are not using a controller, (i.e. the controller is the same raspberry), the controller_timestamp is exactly the same as the raspberry timestamp. To get the raspberry timestamp you can use time_utils.now_timestamp().

Parameters:

• raspberry_timestamp (float) – Raspberry time.

• controller_timestamp (float) – Controller clock timestamp.

register_end_trial(controller_timestamp: float = 0.0) → None

Registers the end of a trial. If you are not using a controller, (i.e. the controller is the same raspberry), the controller_timestamp is exactly the same as the raspberry timestamp. :param controller_timestamp: Controller clock timestamp. :type controller_timestamp: float

register_enter_state(state_name: str, controller_timestamp: float) → None

Registers the entry into a Bpod state.

Parameters:

• state_name (str) – The name of the state entered.

• controller_timestamp (float) – Controller clock timestamp.

register_controller_event(name: str, controller_timestamp: float) → None

Registers a custom event using a controller clock timestamp.

Parameters:

• name (str) – The name of the event (column header).

• controller_timestamp (float) – Controller clock timestamp.

register_raspberry_event(name: str, raspberry_timestamp: float) → None

Registers a custom event using a raspberry timestamp. If you are using bpod, you may want to call this for events that are generated asynchronously from the bpod state machine, for example when executing direct functions or when using a camera to detect or trigger events.

Parameters:

• name (str) – The name of the event (column header).

• raspberry_timestamp (float) – Raspberry time.

register_value(name: str, value: Any) → None

Registers a custom value to be saved with the trial data.

Parameters:

• name (str) – The name of the value (column header).

• value (Any) – The value to store.

execute_function(i: int) → None

Executes a registered function.

Parameters:

i (int) – The function index (1-99).

run_in_thread(daemon: bool = True) → None

Runs the task in a separate background thread.

Parameters:

• daemon (bool, optional) – Whether to run as a daemon thread.

• True. (Defaults to)

run() → None

Runs the task in the main thread until completion or forced stop.

do_trial() → None

Executes a single trial.

Initializes the state machine, runs it, collects data, and performs post-trial updates.

concatenate_trial_data() → None

Appends the current trial’s data to the session DataFrame.

disconnect_and_save(run_mode: str) → Tuple[Save, float, int, int, str]

Stops the task, disconnects devices, and saves session data.

Parameters:

run_mode (str) – The mode in which the task was run (e.g., “Manual”).

Returns:

A tuple containing the save status, session duration, number of trials, water consumed, and settings string.

Return type:

Tuple[Save, float, int, int, str]

save_json(run_mode: str) → str

Saves the session settings to a JSON file.

Parameters:

run_mode (str) – The execution mode string.

Returns:

The JSON string containing the settings.

Return type:

str

save_csv(run_mode: str) → Tuple[float, int, int, bool]

Saves the session data to CSV files.

Processes raw data, saves raw and clean session files, and updates the subject’s cumulative data file.

Parameters:

run_mode (str) – The execution mode string.

Returns:

Duration, trial count, water consumed, and success status.

Return type:

Tuple[float, int, int, bool]

classmethod get_name() → str

Returns the name of the task class.

create_paths() → None

Sets up file and directory paths for the session.