TaskBase

class TaskBase[source]

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.

The task starts by calling start(), then enters a loop where it calls create_trial() followed by after_trial(). The loop continues until the maximum number of trials is reached, the maximum duration is exceeded, or force_stop is set to True. Finally, close() is called to clean up resources.

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

start(self): Called once before the trial loop begins. Use it to configure hardware, pre-compute stimuli, open files, etc.
create_trial(self): Called at the beginning of every trial. If using Bpod: each call starts with an empty state machine. Add states with bpod.add_state and, when the method returns, the state machine is sent and run automatically. after_trial is called once it finishes. If not using Bpod: you are responsible for creating and running the trial logic yourself.
after_trial(self): Called immediately after each trial finishes. Use it to score the animal’s performance, update adaptive parameters, and call register_value to save whatever values you need for the trial. You must always register a water value with the amount of water consumed during the trial; the system sums it across the session to get the total water consumption. If that total falls below a configurable threshold, an alarm is triggered. The threshold can be adjusted in the Settings tab of the GUI.
close(self): 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_start_trial(self, raspberry_timestamp: float, controller_timestamp: float)

No need to call this if you are using a Bpod controller, it is called automatically. For other controllers, call this at the beginning of each trial to register the start of the 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().

register_end_trial(self, controller_timestamp: float)

No need to call this if you are using a Bpod controller, it is called automatically. For other controllers, call this at the end of each trial to register the end of the trial.

register_enter_state(self, state_name: str, controller_timestamp: float)

No need to call this if you are using a Bpod controller, it is called automatically. For other controllers, call this when entering a state to register the entry into that state.

register_controller_event(self, name: str, controller_timestamp: float)

No need to call this if you are using a Bpod controller, it is called automatically. For other controllers, call this to register an event from the controller.

register_raspberry_event(self, name: str, raspberry_timestamp: float)

Call this to register an event from the raspberry, for example when executing direct functions or when using a camera to detect or trigger actions.

register_value(self, name: str, value: Any)

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)

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.name: str

Name of the task.

self.infostr

Human-readable description of the task, set once (usually in __init__). Shown to the user when selecting the task in the GUI.

Example:

self.info = """
My Task
-------
Describe what the task does and how it progresses here.
"""

self.subjectstr

Name of the subject running the session.

self.system_namestr

Name of the system as defined in the settings tab of the GUI.

self.bpodBpodController

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

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.

Example:

gain = self.calibrations.sound_calibration.get_sound_gain(
    speaker=1, dB=70.0, sound_name="white_noise"
)

self.cam_boxCamera | NullCamera

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

self.gpioGpio | NullGpio

Output pin control (set_on() / set_off()). See the Custom GPIO Interaction docs for the input-pin trigger hook.

self.custom_areasdict[int, CameraAreaBase]

BOX area index (1-4) -> the CameraAreaBase overriding that area’s shape, if any (see the Custom Detection Area docs). Empty for any area not overridden.

self.current_trialint

The current trial number starting from 1

self.datestr

Date string of the current session, set once when the session starts.

self.run_modestr

"Manual" or "Auto".

self.trial_datadict

Dictionary populated automatically at the end of each trial. Available inside after_trial.

Keys always present:

"date" (str): date string of the session.
"trial" (int): current trial number.
"subject" (str): subject name.
"task" (str): task name.
"system_name" (str): system name from settings.
"TRIAL_START" (float): absolute timestamp (seconds, UNIX epoch) when the trial started.
"TRIAL_END" (float): absolute timestamp when the trial ended.
"ordered_list_of_events" (list[str]): event names in the order they occurred during the trial (e.g. ["Port1In", "Port1Out", "Port1In"]).

Keys added per state visited (Bpod or manual):

"STATE_<name>_START" (list[float]): timestamps of every entry into that state. It is a list because the same state can be visited more than once in a single trial.
"STATE_<name>_END" (list[float]): timestamps of every exit from that state, paired with the corresponding START list.

Keys added per event type:

"<EventName>" (list[float]): timestamps of every occurrence of that event (e.g. "Port1In", "Port1Out").

When using Bpod all state and event keys are filled automatically. When not using Bpod, call register_enter_state and register_controller_event / register_raspberry_event yourself to populate them.

Keys added by you:

Any name passed to register_value inside after_trial.

Example:

def after_trial(self):
    if self.trial_data.get("correct") == 1:
        self.settings.difficulty += 1
    t_start = self.trial_data["STATE_stimulus_START"][0]
    t_end = self.trial_data["STATE_stimulus_END"][0]
    response_time = t_end - t_start

self.force_stop: bool

Set this to True from inside your own task logic (e.g. in after_trial, once some in-task condition is met) to make the task stop.

self.stop_button_pressed: bool

True once something external (e.g. the STOP TASK button) has asked the task to stop. Unlike force_stop, you should not set this yourself – just read it if you want to react to it.

self.should_stop: bool (read-only)

True once ANY stopping condition is met: the trial/time limit (in Manual mode), force_stop, or stop_button_pressed. If your task does not use Bpod and create_trial contains its own waiting loop, check this property on every iteration so the loop actually exits when the task is asked to stop, instead of running forever:

Example:

def create_trial(self):
    while not self.should_stop:
        ...  # wait for your own condition

self.chrono: Chrono

Using self.chrono.get_seconds() you can get the time in seconds since the task started.

__init__() → None[source]

Methods

Attributes

start() → None[source]: Starts the task. Must be overridden by subclasses.

create_trial() → None[source]: Creates the state machine for the current trial. Must be overridden.

after_trial() → None[source]: Executed after each trial completes. Must be overridden.

close() → None[source]: Closes the task and releases resources. Must be overridden.

register_start_trial(raspberry_timestamp: float, controller_timestamp: float) → None[source]

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 obtain the raspberry_timestamp, use time_utils.now_timestamp().

Parameters:

raspberry_timestamp (float) – Raspberry time.
controller_timestamp (float) – Controller clock timestamp.

register_end_trial(controller_timestamp: float) → None[source]

Registers the end of a trial. If you are not using a controller, (i.e. the controller is the same raspberry), pass time_utils.now_timestamp() as controller_timestamp.

Parameters:: controller_timestamp (float) – Controller clock timestamp.

register_enter_state(state_name: str, controller_timestamp: float) → None[source]

Registers the entry into a state machine state. If you are not using a controller, (i.e. the controller is the same raspberry), pass time_utils.now_timestamp() as controller_timestamp.

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[source]

Registers a custom event using a controller clock timestamp. If you are not using a controller, (i.e. the controller is the same raspberry), this method does exactly the same as register_raspberry_event().

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[source]

Registers a custom event using a raspberry timestamp. You may want to call this for events that are generated asynchronously from the controller state machine, for example when executing direct functions or when using a camera to detect or trigger events. Pass time_utils.now_timestamp() as raspberry_timestamp.

Parameters:

name (str) – The name of the event (column header).
raspberry_timestamp (float) – Raspberry time.

register_value(name: str, value: Any) → None[source]

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[source]

Executes a registered function.

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

property should_stop: bool

True once any stopping condition is met.

Trial/time limits and force_stop are only checked BETWEEN trials, by run()’s own while loop – they cannot interrupt a create_trial() that is currently blocked in its own waiting loop. Tasks that do not use Bpod and wait on their own condition inside create_trial should poll this property instead, so they actually exit when asked to.

The trial limit only applies in Manual mode: Auto mode runs until the subject leaves the corridor, not for a fixed number of trials.

run() → None[source]: Runs the task in the main thread until completion or forced stop.

do_trial() → None[source]

Executes a single trial.

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

concatenate_trial_data() → None[source]: Appends the current trial’s data to the session DataFrame.

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

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[source]

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][source]

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[source]: Returns the name of the task class.

create_paths() → None[source]: Sets up file and directory paths for the session.