poulet_py package

class Arduino(ports=None)

Bases: object

__init__(ports=None)

Initialize the Arduino class with the given ports.

Args:

ports (list): List of serial port addresses.

close_port()

Close all serial connections.

log_error(error_message)

read_data(data_parser=<class 'float'>)

Read data from the Arduino and parse it.

Args:

data_parser (callable): Function to parse the data read from the Arduino.

Defaults to float.

Sets:

self.parsed_data: Parsed data from Arduino.

save_data(data)

Save the data to a CSV file.

Args:

timestamp: The timestamp to be saved.

set_error_log_path(folder_path, error_file_name)

Set the error log path for the Arduino.

Args:

folder_path (str): Path to the folder where the error log will be stored. error_file_name (str): Name of the error log file.

set_output_file(path, extra_name, data_columns=['data'], base_file_name='arduino')

Set the output file for the Arduino.

Args:

path (str): Path to the folder where the output file will be saved. extra_name (str): An additional name to be added to the base file name. base_file_name (str): Base name of the output file. Defaults to ‘arduino’.

set_timer(start_time)

Sets the timer for the camera.

Args:

start_time (float): The time at which the camera recording started.

class BaseTrigger(**data)

Bases: BaseModel, ABC

Abstract base class for trigger devices.

abstractmethod cleanup()

Cleanup trigger resources.

Return type:

None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

abstractmethod wait()

Wait for trigger event.

Return type:

bool

name: str

timeout: float | None

class BaslerCamera(max_cameras=2)

Bases: object

A class to interact with multiple Basler cameras using pypylon and OpenCV. Each camera will record to its own video file and log timestamps to a CSV.

__init__(max_cameras=2)

Initializes the BaslerCamera object by enumerating devices and attaching up to max_cameras.

Args:

max_cameras (int): The maximum number of cameras to use.

capture_frame()

Captures a single frame from whichever camera has a frame ready. The frame is written to its corresponding video file and timestamp logged.

log_error(error_message)

recording(data_save_folder, cage_id, n_mouse, condition, mouse_ids=[], duration_s=10, buffer_s=10, total_rec=4, fps=30, video_format='mp4')

save_metadata(base_file_name='basler-camera', extra_name='')

Saves metadata about the recording for each camera to a JSON file.

Args:

path (str): Directory to save the metadata files. base_file_name (str, optional): Base name for the metadata files. extra_name (str, optional): Extra name to add to the file names.

save_timestamp(camera_index, timestamp)

Save a timestamp to the CSV file for the specified camera.

Args:

camera_index (int): Index of the camera. timestamp (float): Timestamp to record.

set_error_log_path(path, file_name)

Sets the error log file.

Args:

path (str): Directory for the error log. file_name (str): Name of the error log file.

set_frames_per_second(frames_per_second)

Sets the frame rate for each camera.

Args:

frames_per_second (float): Desired frame rate in frames per second.

set_output_file(path, extra_name, base_file_name='basler-camera')

Sets up output video files and timestamp CSV files for all cameras.

Args:

path (str): Directory to save the output files. extra_name (str): Extra name to add to the file names. base_file_name (str): Base name for the files.

set_timer(start_time)

Sets the timer for the camera.

Args:

start_time (float): The time at which the camera recording started.

start_streaming()

Starts the grabbing (streaming) for all cameras.

stop_streaming()

Stops the streaming and closes all cameras and video writers.

stream_video(window_width=None, window_height=None)

Streams the live video feed from all cameras. Each camera is shown in its own window.

Args:

window_width (int, optional): Width to resize the window. window_height (int, optional): Height to resize the window.

class GPIOTrigger(*, name: str = '', timeout: float | None = None, pin: int, pull_up: bool = False, edge: Literal['rising', 'falling', 'both'] = 'rising')

Bases: BaseTrigger

GPIO-based trigger using gpiozero.

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

cleanup()

Cleanup GPIO resources.

Return type:

None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Return type:

None

Args:

self: The BaseModel instance. context: The context.

wait()

Wait for GPIO event.

Return type:

bool

pin: int

pull_up: bool

edge: Literal['rising', 'falling', 'both']

class JulaboChiller(port=None, baudrate=9600, timeout=1)

Bases: object

Class to interact with a Julabo water chiller via serial port.

__init__(port=None, baudrate=9600, timeout=1)

Initialize the JulaboChiller with the given serial port configuration.

Args:

port (str): The serial port (e.g., ‘COM12’). baudrate (int, optional): The baud rate for the serial communication. Default is 9600. timeout (int or float, optional): The read timeout value. Default is 1 second.

check_started()

Check whether the chiller has started.

Returns:

str: The started status.

check_status()

Check the status of the chiller.

Returns:

str: The status information.

check_version()

Check the version of the chiller.

Returns:

str: The version information.

close_port()

Close the serial port connection.

get_target_temperature()

Get the target temperature set point.

Returns:

str: The target temperature.

get_temperature()

Get the current temperature from the chiller.

Returns:

str: The current temperature reported by the chiller.

read()

Read data from the chiller.

Returns:

str: The response from the chiller, or None if no data is available.

save_temperature(timestamp, temperature)

Save the temperature to a CSV file.

Args:

timestamp: The timestamp to be saved. temperature: The temperature to be saved.

set_error_log_path(path, file_name)

Sets the path for the error log file.

Args:

path (str): The directory where the error log file will be saved.

set_output_file(path, extra_name, base_file_name='julabo_chiller')

Sets the output file for recording the video.

Args:

path (str): The directory where the output file will be saved.

set_temperature(temperature)

Set the temperature of the chiller.

Args:

temperature (float): The temperature to set (in Celsius).

set_timer(start_time)

Set the timer for the chiller.

Args:

start_time (float): The start time of the experiment.

start()

Turn on the chiller.

stop()

Turn off the chiller.

write(command)

Write a command to the chiller.

Args:

command (str): The command to send to the chiller.

class KeyboardTrigger(**data)

Bases: BaseTrigger

Keyboard-based trigger (press Enter key).

cleanup()

No cleanup needed for keyboard.

Return type:

None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

wait()

Wait for keyboard input with optional timeout.

Return type:

bool

key: str

class Oscilloscope(max_samples=1000, max_points=100, title='Real-time Data', xlabel='X', ylabel='Y', xlim='auto', ylim='auto', xpadding=0.1, ypadding=0.1, animation_interval=33)

Bases: object

A real-time data visualization tool similar to an oscilloscope.

This class provides a thread-safe, animated plotting interface for visualizing streaming data with automatic downsampling and view adjustment.

Parameters:

• max_samples (int, optional) – Maximum number of samples to keep in memory (default: 1000).

• max_points (int, optional) – Maximum number of points to display (downsampling threshold, default: 100).

• title (str, optional) – Title of the plot (default: “Real-time Data”).

• xlabel (str, optional) – Label for the x-axis (default: “X”).

• ylabel (str, optional) – Label for the y-axis (default: “Y”).

• xlim (tuple[float, float] or "auto", optional) – Fixed x-axis limits or “auto” for automatic adjustment (default: “auto”).

• ylim (tuple[float, float] or "auto", optional) – Fixed y-axis limits or “auto” for automatic adjustment (default: “auto”).

• xpadding (float, optional) – Padding factor for x-axis when in auto mode (default: 0.1).

• ypadding (float, optional) – Padding factor for y-axis when in auto mode (default: 0.1).

• animation_interval (int, optional) – Refresh interval in milliseconds (default: 33 ~30fps).

fig

The figure instance containing the plot.

Type:

matplotlib.figure.Figure

ax

The axes instance for the plot.

Type:

matplotlib.axes.Axes

Examples

>>> osc = Oscilloscope(max_samples=500, title="Sensor Data")
>>> osc.start()
>>> osc.add_data({"temp": 25.3, "humidity": 45.2})

__init__(max_samples=1000, max_points=100, title='Real-time Data', xlabel='X', ylabel='Y', xlim='auto', ylim='auto', xpadding=0.1, ypadding=0.1, animation_interval=33)

add_data(y, x=None)

Add new data points to the oscilloscope.

This method is thread-safe and can be called from multiple threads.

Parameters:

• y (dict) – Dictionary of y-values where keys are series names and values are the data points. Each call adds one sample per series.

• x (Any, optional) – Corresponding x-value for the y samples. If None, uses an auto-incremented index (default: None).

Return type:

None

Notes

The data is stored in internal buffers and will be displayed during the next animation frame.

force_redraw()

Force an immediate redraw of the plot.

Useful when you need to update the display outside the normal animation interval.

Return type:

None

start()

Start the real-time plotting animation.

Enables interactive mode and begins periodic updates of the display. The plot window will be non-blocking.

See also

stop

Stop the animation and close the window.

Return type:

None

stop()

Stop the animation and clean up resources.

Closes the plot window and clears all internal data buffers.

Return type:

None

class Seq(input_file, *, height=None, width=None, raw=False)

Bases: Seq

A sequence handler for FLIR thermal data with export capabilities.

Extends flirpy.io.seq.Seq to add NumPy, HDF5, and pickle serialization.

Parameters:

• input_file (str | Path) – Path to the input SEQ file.

• height (int, optional) – Force a specific image height. If None, uses the file’s metadata.

• width (int, optional) – Force a specific image width. If None, uses the file’s metadata.

• raw (bool, default=False) – If True, returns the raw bytes not Fff class.

__init__(input_file, *, height=None, width=None, raw=False)

Load a FLIR SEQ file. Currently this must be a SEQ file containing FFF files. The resulting object can be indexed as a normal array and will return the

to_hdf(path, *, key, mode='w', dtype=None, complib=None, complevel=None, meta=None)

Save the sequence to an HDF5 file.

Parameters:

• path (str | Path) – Output file path.

• key (str) – HDF5 dataset name.

• mode (str, default="w") – File mode (‘w’, ‘a’, ‘r+’, etc.).

• dtype (type, optional) – Data type for the stored array.

• complib (str, optional) – Compression library (e.g., ‘gzip’, ‘lzf’).

• complevel (int, optional) – Compression level (1-9 for gzip).

• meta (dict, optional) – Metadata to store as HDF5 attributes.

to_list()

Convert the sequence to a list of frames.

Returns:

List of frames (as ndarray or bytes if unprocessed).

Return type:

list

to_numpy(*, dtype=None)

Convert the sequence to a NumPy array.

Parameters:

dtype (type, optional) – Desired data type for the output array (e.g., np.float32).

Returns:

Stacked frames as a 3D array (n_frames, height, width).

Return type:

ndarray

to_pickle(path, *, dtype=None, compression='infer', protocol=5)

Save the sequence to a pickle file using pandas’ I/O handler.

Parameters:

• path (str | Path) – Output file path.

• dtype (type, optional) – Data type for the array.

• compression (str or dict, default 'infer') – For on-the-fly compression of the output data. If ‘infer’ and ‘path’ is path-like, then detect compression from the following extensions: ‘.gz’, ‘.bz2’, ‘.zip’, ‘.xz’, ‘.zst’, ‘.tar’, ‘.tar.gz’, ‘.tar.xz’ or ‘.tar.bz2’ (otherwise no compression). Set to None for no compression. Can also be a dict with key ‘method’ set to one of {‘zip’, ‘gzip’, ‘bz2’, ‘zstd’, ‘xz’, ‘tar’} and other key-value pairs are forwarded to zipfile.ZipFile, gzip.GzipFile, bz2.BZ2File, zstandard.ZstdCompressor, lzma.LZMAFile or tarfile.TarFile, respectively. As an example, the following could be passed for faster compression and to create a reproducible gzip archive: compression={‘method’: ‘gzip’, ‘compresslevel’: 1, ‘mtime’: 1}

• protocol (int, default=HIGHEST_PROTOCOL) – Pickle protocol version.

class Settings(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, **values)

Bases: BaseSettings

This class loads configuration from environment variables and a .env file. It supports nested configuration using the env_nested_delimiter (e.g., LOG__LEVEL).

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

log: Log

class TCS(port, *, maximum_temperature=40.0, beep=False, trigger_out_channel=255, read_timeout=2.0, response_timeout=2.0, stimulus_trigger=None)

Bases: object

Interface for communicating with a TCS thermal stimulator.

Parameters:

• port (str) – Serial port to which the device is connected.

• maximum_temperature (float, optional) – Maximum allowed temperature in °C (default: 40).

• beep (bool, optional) – Whether to enable audible beeps (default: False).

• trigger_out_channel (int, optional) – Output channel for trigger signals (default: 255).

• read_timeout (float, optional) – Timeout for read operations in seconds (default: 2).

• response_timeout (float, optional) – Timeout for device responses in seconds (default: 2).

• stimulus_trigger (BaseTrigger, optional) – A Trigger found in poulet_py/hardware/triggers to trigger the next stimulus.

init() → None

Initialize the TCS connection and verify communication.

close() → None

Close the connection and clean up resources.

info() → str

Get device information including firmware version and probe details.

reset() → None

Reset the TCS device to its default state.

trigger() → None

Execute the configured stimulation.

get_readings() → dict

Get current temperature readings from all sensors.

Examples

>>> with TCS(port="/dev/ttyUSB0") as tcs:
>>>     tcs.init()
>>>     stimulus = TCSStimulus(surface=1, target=35.0)
>>>     tcs.stimulus = stimulus
>>>     tcs.trigger()
>>>     readings = tcs.get_readings()
>>>     print(readings)

__init__(port, *, maximum_temperature=40.0, beep=False, trigger_out_channel=255, read_timeout=2.0, response_timeout=2.0, stimulus_trigger=None)

close()

Close the connection and clean up resources.

execute_command(command, *args, expected_pattern=None)

Execute a command and optionally wait for a response.

Parameters:

• command (TCSCommand) – The command to execute

• *args – Arguments to format into the command

• expected_pattern (Pattern | None) – Regex pattern to match against the response

Returns:

If expected_pattern provided, returns (timestamp, match) tuple

Return type:

tuple[int, Match[str]]] | None

Examples

>>> tcs.execute_command(TCSCommand.READ_INFO, expected_pattern=compile(r"Firmware:(.*)"))

get_readings()

Get current temperature readings from all sensors.

Returns:

Dictionary containing temperatures for neutral and all surfaces, plus a timestamp key.

Return type:

Dict[str, float]

Raises:

RuntimeError – If reading temperatures fails

info()

Get device information including firmware version and probe details.

Returns:

Device information string

Return type:

str

Raises:

RuntimeError – If the info command fails or times out

init()

Initialize the TCS connection and verify communication.

Raises:

RuntimeError – If initialization fails

reset()

Reset the TCS device to its default state.

property stimulus: TCSStimulus

trigger()

Execute the configured stimulation.

Raises:

RuntimeError – If stimulation fails to trigger

write(command)

Write a command to the TCS device.

Parameters:

command (bytes) – The command to send

Returns:

Number of bytes written

Return type:

int

Raises:

RuntimeError – If the write operation fails

class TCSCommand(*values)

Bases: bytes, Enum

Enumeration of all available TCS commands with their byte representations.

Each command includes formatting capability for parameterized commands.

Examples

>>> TCSCommand.READ_TEMPERATURES
<TCSCommand.READ_TEMPERATURES: b'E'>
>>> TCSCommand.BASELINE_TEMPERATURE.format(300)
b'N300'

format(*args)

Format the command with the given arguments.

Parameters:

*args (int, float) – Arguments to format into the command string

Returns:

Formatted command string

Return type:

bytes

Raises:

ValueError – If arguments don’t match the command’s format requirements

Examples

>>> TCSCommand.TARGET_TEMPERATURE.format(1, 350)
b'C1350'

READ_INFO = b'H'

READ_TEMPERATURES = b'E'

READ_STIMULATION_VALUES = b'P'

READ_BUTTON_STATUS = b'K'

READ_BATTERY = b'B'

READ_ERRORS = b'Q'

DISPLAY_TEMPERATURES_BETWEEN_STIMULATION = b'Oa'

DISPLAY_TEMPERATURES_DURING_STIMULATION = b'Ob'

RESET = b'Oc'

SET_MAX_TEMPERATURE = b'Om%03d'

AUTOMATIC_CALIBRATION = b'G'

DEACTIVATE_DISPLAY = b'F'

TRIGGER_STIMULATION = b'L'

HALT_STIMULATION = b'A'

BASELINE_TEMPERATURE = b'N%03d'

SURFACE_SELECTION = b'S%05d'

TARGET_TEMPERATURE = b'C%d%03d'

STIMULATION_RATE = b'V%d%04d'

RETURN_SPEED = b'R%d%04d'

STIMULATION_DURATION = b'D%d%05d'

TRIGGER_CHANNEL_DURATION = b'T%03d%03d'

BUZZER = b'Z%03d%03d'

class TCSInterface(port, *, maximum_temperature=40, beep=False, trigger_out_channel=255, read_timeout=2, response_timeout=2, n_trials=1, stimuli=None, stimulus_trigger=None, mode='random', interstimulus_period=0)

Bases: TCS

Interface class for the Thermal Cutaneous Stimulator (TCS) device.

Handles serial communication with the device and provides methods to configure and execute thermal stimulus experiments.

Parameters:

• port (str) – Serial port to which the device is connected.s

• maximum_temperature (float, optional) – Maximum allowed temperature in °C (default: 40).

• beep (bool, optional) – Whether to enable audible beeps (default: False).

• trigger_out_channel (int, optional) – Output channel for trigger signals (default: 255).

• read_timeout (float, optional) – Timeout for read operations in seconds (default: 2).

• response_timeout (float, optional) – Timeout for device responses in seconds (default: 2).

• n_trials (int, optional) – Number of trials to run (default: 1).

• stimuli (list[TCSStimulus], optional) – List of stimulus configurations (default: None).

• stimulus_trigger (BaseTrigger, optional) – A Trigger found in poulet_py/hardware/triggers to trigger the next stimulus

• mode ({'random', 'fixed'}, optional) – Stimulus presentation mode (default: ‘random’).

• interstimulus_period (int or list[int], optional) – Time between stimuli in milliseconds (default: 0).

stimuli

List of configured stimulus sequences.

Type:

list[TCSStimulus]

n_trials

Number of trials to be executed.

Type:

int

mode

Stimulus presentation mode (‘random’ or ‘fixed’).

Type:

str

interstimulus_period

Time between stimuli in milliseconds. assign a list of integers for random selection or a single integer for fixed time.

Type:

int or list[int]

run(plot=False, max_plot_samples=1000)

Execute the configured experiment.

to_df()

Convert collected data to a pandas DataFrame.

__init__(port, *, maximum_temperature=40, beep=False, trigger_out_channel=255, read_timeout=2, response_timeout=2, n_trials=1, stimuli=None, stimulus_trigger=None, mode='random', interstimulus_period=0)

run(*, plot=False, max_plot_samples=1000)

Execute the configured experiment.

Parameters:

• plot (bool, optional) – Whether to show real-time plotting (default: False).

• max_plot_samples (int, optional) – Maximum number of samples to show in the plot (default: 1000).

Returns:

List of readings collected during the experiment.

Return type:

list[dict]

Raises:

ValueError – If no stimuli are configured before running.

Notes

Each reading contains: - timestamp: Time of reading - temperature: Current temperature - trial: Trial number - Other relevant parameters

property stimuli: list[TCSStimulus]

Get or set the stimulus sequence.

Returns:

Currently configured stimulus sequence.

Return type:

list[TCSStimulus]

Notes

When setting the stimuli, they are automatically validated and a sequence is generated according to the configured mode (random/fixed) and number of trials.

to_df()

Convert collected data to a pandas DataFrame.

Returns:

Formatted DataFrame with timestamp as index and trial data.

Return type:

pandas.DataFrame

Examples

>>> df = device.to_df()
>>> df.plot()

class TCSStimulus(**data)

Bases: BaseModel

Configuration for thermal stimulation parameters.

surface

Target surface (0-5, where 0 means all surfaces).

Type:

int

baseline

Baseline temperature in °C (20-45).

Type:

float

target

Target temperature in °C (0-60).

Type:

float

rise_rate

Temperature rise rate in °C/s (0.1-999.9).

Type:

float

return_speed

Temperature return speed in °C/s (0.1-999.9).

Type:

float

duration

Stimulation duration in ms (10-99999).

Type:

int

commands() → list

Generate the sequence of commands needed to configure this stimulus.

Examples

>>> stimulus = TCSStimulus(surface=1)
>>> stimulus.commands()
[b'S10000', b'N300', b'C1000', b'V10010', b'D100100', b'R10010']

commands()

Generate the sequence of commands needed to configure this stimulus.

Returns:

Sequence of formatted command strings

Return type:

list

Examples

>>> stimulus = TCSStimulus(surface=1)
>>> stimulus.commands()
[b'S10000', b'N300', b'C1000', b'V10010', b'D100100', b'R10010']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

surface: int

baseline: float

target: float

rise_rate: float

return_speed: float

duration: int

check_or_create(path)

Function to check whether a folder exists. If it does NOT exist, it is created

define_folder_name(name, *, add_date=True)

Return type:

str

generate_stimulus_sequence(n, *, stimuli_options, mode='random')

Generate a list of trials with specified stimuli distribution.

Parameters:

• n (int) – Number of trials to generate. Must be divisible by the number of stimuli options when mode is ‘random’ or when multiple stimuli are provided in ‘fixed’ mode.

• stimuli_options (List[Any]) – List of possible stimulus values. For a single stimulus, all trials will use it. For multiple stimuli, distribution depends on mode.

• mode ({'random', 'fixed'}, optional) – Distribution mode: - ‘random’: Shuffled trials with equal representation of each stimulus - ‘fixed’: Trials use stimuli in sequence (or single stimulus repeated) (default: ‘random’)

Returns:

Generated list of stimuli for each trial

Return type:

List[Any]

Raises:

ValueError – If n is not divisible by number of stimuli options (for relevant modes), or if mode is invalid.

Notes

• For ‘random’ mode with multiple stimuli, each appears exactly

  n//len(stimuli_options) times.

• For ‘fixed’ mode with multiple stimuli, stimuli are repeated in sequence

  until n is reached.

• For ‘fixed’ mode with single stimulus, that stimulus is repeated n times.

Examples

>>> generate_trials(4, stimuli_options=[1, 2], mode="random")
[2, 1, 2, 1]  # Random order with equal representation

>>> generate_trials(3, stimuli_options=[5], mode="fixed")
[5, 5, 5]

go_to(key, *, path=None)

Change the current working directory to the level containing a specified key in a path.

Parameters:

• key (str) – The directory name to search for in the path (e.g., “neuropixels”).

• path (Path or str, optional) – The input path to analyze. Defaults to the caller script’s location (__file__).

Returns:

True if the directory was changed successfully, False otherwise.

Return type:

bool

Examples

>>> change_dir_to_key("neuropixels", path="/project/neuropixels/src/file.py")
True  # Changes CWD to "/project/neuropixels"

json_serializer(data, file=None)

Serialize data to JSON.

Features: - Efficient serialization using orjson (faster than standard json module) - Built-in numpy array support via orjson’s native serialization - Returns bytes if no file provided, writes to file otherwise

Return type:

bytes | None

Args:

data: Dictionary containing data to serialize.

Numpy arrays are automatically supported.

file: Output file

If None, returns serialized bytes instead of writing to file. Must end with ‘.json’ extension if provided.

Returns:

bytes | None: Serialized JSON as bytes if file is None, otherwise None.

Raises:

ValueError: If provided file doesn’t end with ‘.json’ extension TypeError: If data contains non-serializable types JSONEncodeError: If serialization fails for other reasons

Example:

>>> data = {"array": np.array([1, 2, 3])}
>>> # Write to file
>>> json_serializer(data, "output.json")
>>> # Get bytes
>>> json_bytes = json_serializer(data)

sanitize_path(path, *, add_timestamp=False)

Sanitize a full path by making all components filesystem-friendly and optionally adding a timestamp to the last component (file or folder).

Parameters:

• path (str) – The input path to sanitize (can include folders and filename)

• add_timestamp (bool, optional) – If True, prepend timestamp in YYYYMMDD format to the last component

Returns:

Sanitized path with all special characters replaced by underscores and optional timestamp on the last component

Return type:

str

Examples

>>> sanitize_path("/s$ss!on1/d%ta/345.pkl", add_timestamp=True)
'/s_ss_on1/d_ta/20250503_345.pkl'

>>> sanitize_path("my@project/data#files")
'my_project/data_files'

save_metadata_exp(metadata, path, name)

setup_logging(logger, *, terminal_width=None, show_time=False, show_path=True, markup=True, rich_tracebacks=True, tracebacks_extra_lines=4, tracebacks_word_wrap=True, tracebacks_show_locals=True, level='warning', file=None)

Configure logging for the provided logger with optional rich formatting and file logging.

Parameters:

• logger (Logger) – The logger instance to configure.

• terminal_width (int, optional) – The width of the terminal for rich console output. If None, the default width is used.

• show_time (bool, optional) – Whether to show the time in the log output. Default is False.

• show_path (bool, optional) – Whether to show the path in the log output. Default is True.

• markup (bool, optional) – Whether to enable markup in the log output. Default is True.

• rich_tracebacks (bool, optional) – Whether to enable rich tracebacks. Default is True.

• tracebacks_extra_lines (int, optional) – Number of extra lines to show in tracebacks. Default is 4.

• tracebacks_word_wrap (bool, optional) – Whether to enable word wrap in tracebacks. Default is True.

• tracebacks_show_locals (bool, optional) – Whether to show local variables in tracebacks. Default is True.

• level (int or str, optional) – The logging level to set for the logger. Default is warning.

• file (str, optional) – The file to log to. If None, logs are output to the console.

Return type:

None

Subpackages

• poulet_py.config package
  • Settings
    • Settings.model_config
    • Settings.log
  • setup_logging()
  • Submodules
    • poulet_py.config.logging module
      • setup_logging()
    • poulet_py.config.settings module
      • Usage Examples
      • Log
      • Settings
      • SETTINGS
• poulet_py.converters package
  • Seq
    • Seq.__init__()
    • Seq.to_hdf()
    • Seq.to_list()
    • Seq.to_numpy()
    • Seq.to_pickle()
  • Submodules
    • poulet_py.converters.seq module
      • Seq
• poulet_py.hardware package
  • Arduino
    • Arduino.__init__()
    • Arduino.close_port()
    • Arduino.log_error()
    • Arduino.read_data()
    • Arduino.save_data()
    • Arduino.set_error_log_path()
    • Arduino.set_output_file()
    • Arduino.set_timer()
  • BaseTrigger
    • BaseTrigger.cleanup()
    • BaseTrigger.model_config
    • BaseTrigger.wait()
    • BaseTrigger.name
    • BaseTrigger.timeout
  • BaslerCamera
    • BaslerCamera.__init__()
    • BaslerCamera.capture_frame()
    • BaslerCamera.log_error()
    • BaslerCamera.recording()
    • BaslerCamera.save_metadata()
    • BaslerCamera.save_timestamp()
    • BaslerCamera.set_error_log_path()
    • BaslerCamera.set_frames_per_second()
    • BaslerCamera.set_output_file()
    • BaslerCamera.set_timer()
    • BaslerCamera.start_streaming()
    • BaslerCamera.stop_streaming()
    • BaslerCamera.stream_video()
  • GPIOTrigger
    • GPIOTrigger.__init__()
    • GPIOTrigger.cleanup()
    • GPIOTrigger.model_config
    • GPIOTrigger.model_post_init()
    • GPIOTrigger.wait()
    • GPIOTrigger.pin
    • GPIOTrigger.pull_up
    • GPIOTrigger.edge
    • GPIOTrigger.name
    • GPIOTrigger.timeout
  • JulaboChiller
    • JulaboChiller.__init__()
    • JulaboChiller.check_started()
    • JulaboChiller.check_status()
    • JulaboChiller.check_version()
    • JulaboChiller.close_port()
    • JulaboChiller.get_target_temperature()
    • JulaboChiller.get_temperature()
    • JulaboChiller.read()
    • JulaboChiller.save_temperature()
    • JulaboChiller.set_error_log_path()
    • JulaboChiller.set_output_file()
    • JulaboChiller.set_temperature()
    • JulaboChiller.set_timer()
    • JulaboChiller.start()
    • JulaboChiller.stop()
    • JulaboChiller.write()
  • KeyboardTrigger
    • KeyboardTrigger.cleanup()
    • KeyboardTrigger.model_config
    • KeyboardTrigger.wait()
    • KeyboardTrigger.key
    • KeyboardTrigger.name
    • KeyboardTrigger.timeout
  • TCS
    • TCS.init()
    • TCS.close()
    • TCS.info()
    • TCS.reset()
    • TCS.trigger()
    • TCS.get_readings()
    • TCS.__init__()
    • TCS.close()
    • TCS.execute_command()
    • TCS.get_readings()
    • TCS.info()
    • TCS.init()
    • TCS.reset()
    • TCS.stimulus
    • TCS.trigger()
    • TCS.write()
  • TCSCommand
    • TCSCommand.format()
    • TCSCommand.READ_INFO
    • TCSCommand.READ_TEMPERATURES
    • TCSCommand.READ_STIMULATION_VALUES
    • TCSCommand.READ_BUTTON_STATUS
    • TCSCommand.READ_BATTERY
    • TCSCommand.READ_ERRORS
    • TCSCommand.DISPLAY_TEMPERATURES_BETWEEN_STIMULATION
    • TCSCommand.DISPLAY_TEMPERATURES_DURING_STIMULATION
    • TCSCommand.RESET
    • TCSCommand.SET_MAX_TEMPERATURE
    • TCSCommand.AUTOMATIC_CALIBRATION
    • TCSCommand.DEACTIVATE_DISPLAY
    • TCSCommand.TRIGGER_STIMULATION
    • TCSCommand.HALT_STIMULATION
    • TCSCommand.BASELINE_TEMPERATURE
    • TCSCommand.SURFACE_SELECTION
    • TCSCommand.TARGET_TEMPERATURE
    • TCSCommand.STIMULATION_RATE
    • TCSCommand.RETURN_SPEED
    • TCSCommand.STIMULATION_DURATION
    • TCSCommand.TRIGGER_CHANNEL_DURATION
    • TCSCommand.BUZZER
  • TCSStimulus
    • TCSStimulus.surface
    • TCSStimulus.baseline
    • TCSStimulus.target
    • TCSStimulus.rise_rate
    • TCSStimulus.return_speed
    • TCSStimulus.duration
    • TCSStimulus.commands()
    • TCSStimulus.commands()
    • TCSStimulus.model_config
    • TCSStimulus.surface
    • TCSStimulus.baseline
    • TCSStimulus.target
    • TCSStimulus.rise_rate
    • TCSStimulus.return_speed
    • TCSStimulus.duration
  • Subpackages
    • poulet_py.hardware.camera package
      • BaslerCamera
      • Submodules
    • poulet_py.hardware.stimulator package
      • Arduino
      • JulaboChiller
      • TCS
      • TCSCommand
      • TCSStimulus
      • Submodules
    • poulet_py.hardware.triggers package
      • BaseTrigger
      • GPIOTrigger
      • KeyboardTrigger
      • Submodules
• poulet_py.tools package
  • check_or_create()
  • define_folder_name()
  • generate_stimulus_sequence()
  • go_to()
  • json_serializer()
  • sanitize_path()
  • save_metadata_exp()
  • Submodules
    • poulet_py.tools.generators module
      • generate_stimulus_sequence()
    • poulet_py.tools.organizational module
      • check_or_create()
      • define_folder_name()
      • sanitize_path()
      • go_to()
    • poulet_py.tools.serializers module
      • json_serializer()
      • save_metadata_exp()
• poulet_py.utils package
  • Oscilloscope
    • Oscilloscope.fig
    • Oscilloscope.ax
    • Oscilloscope.__init__()
    • Oscilloscope.add_data()
    • Oscilloscope.force_redraw()
    • Oscilloscope.start()
    • Oscilloscope.stop()
  • TCSInterface
    • TCSInterface.stimuli
    • TCSInterface.n_trials
    • TCSInterface.mode
    • TCSInterface.interstimulus_period
    • TCSInterface.run()
    • TCSInterface.to_df()
    • TCSInterface.__init__()
    • TCSInterface.run()
    • TCSInterface.stimuli
    • TCSInterface.to_df()
  • Submodules
    • poulet_py.utils.oscilloscope module
      • Oscilloscope
    • poulet_py.utils.qst module
      • TCSInterface