rtcog.utils package

Submodules

rtcog.utils.core module

show-inheritance:

class rtcog.utils.core.SharedClock

Bases: object

Minimal clock for tracking elapsed time across processes.

This clock is initialized at instantiation and provides a method for retrieving the current time relative to that start point.

_start_time

The absolute time (in seconds) when the clock was created, based on time.perf_counter().

Type:

float

now()

Get the current time relative to the clock’s start time.

Returns:

Elapsed time in seconds since the clock was initialized.

Return type:

float

rtcog.utils.core.create_sync_events()

Create multiprocessing infrastructure.

rtcog.utils.core.file_exists(path)

Check if file exists.

rtcog.utils.core.setup_afni()

rtcog.utils.exceptions module

exception rtcog.utils.exceptions.VolumeOverflowError

Bases: Exception

Raised when the scanner sends more volumes than expected.

rtcog.utils.fMRI module

rtcog.utils.fMRI.load_fMRI_file(path, verbose=False)

Load fMRI Dataset

Parameters:

path: str

full path to the dataset of interest

Returns:

img: nib.Nifti2Image

Image structure with data, affine, etc.

rtcog.utils.fMRI.mask_fMRI_img(data_img, mask_img, verbose=False)

Converts NiftiImage into vectorized numpy array [Nv,Nt] It is an equivalent to nilearn mask_img, but I needed to implement this becuase of the issues with ordering going from 3D/4D to vector. This way is consistent throughout the code.

Parameters:

data_img: Nifti2Image

Nifti Image with the 3D or 4D to be vectorized

mask_img: Nifti2Image

Nifti Image with the 3D mask

Returns:

data_v: np.array [Nacquisitions, Nvoxels in mask]

rtcog.utils.fMRI.unmask_fMRI_img(data, mask_img, out_path=None)

Convert Nv,Nt array of fMRI data into an actual Nifti Image, and possibly write to disk

Parameters:

datanp.array [Nv,Nt]

data to be reshaped into an image

mask_img: Nifti2Image

mask used to obtain the affine, and where to put the data in space

out_path: str

path to write nifti file. None means do not write file

Returns:

out: Nifti2Image

rtcog.utils.log module

rtcog.utils.log.get_logger()

Configure and return logger with both console and file handlers.

Returns:

The configured logger instance.

Return type:

logging.Logger

rtcog.utils.log.set_logger(debug=False, silent=False)

Set the logging level.

Parameters:

• debug (bool) – Whether to set the logging level to DEBUG.

• silent (bool) – Whether to silence all but CRITICAL messages.

rtcog.utils.options module

class rtcog.utils.options.Options(config)

Bases: object

Configuration object for the real-time fMRI pipeline.

Loads configuration from a YAML file and/or command-line arguments, and exposes them as attributes for easy access throughout the experiment code.

(Dynamically assigned)

All configuration keys are stored as object attributes.

classmethod from_cli(argv=None)

Create an Options object from CLI arguments (with YAML config as base).

Parameters:

argv (list or None) – Command-line arguments.

Returns:

An instance with merged YAML and CLI arguments.

Return type:

Options

classmethod from_yaml(path)

Create an Options object from a YAML file.

Useful for testing purposes.

Parameters:

path (str) – Path to the YAML configuration file.

Returns:

An instance populated with the file contents.

Return type:

Options

static load_yaml(path)

Load configuration from a YAML file.

Parameters:

path (str) – Path to the YAML configuration file.

Returns:

Parsed YAML content as a dictionary.

Return type:

dict

classmethod parse_cli_args(argv=None)

Parse command-line arguments to load and optionally override config options.

Parameters:

argv (list or None) – Command-line arguments.

Returns:

Merged configuration from YAML and CLI overrides.

Return type:

dict

Raises:

• FileNotFoundError – If the specified YAML config file cannot be found.

• SystemExit – If required arguments are missing.

save_config()

Save the current configuration to a YAML file in the output directory. The filename will include the out_prefix (e.g., Run01_Options.yaml).

rtcog.utils.recorder module

recorder.py Provides WAV recording functionality via two approaches: Blocking mode (record for a set duration): >>> rec = Recorder(channels=2) >>> with rec.open(‘blocking.wav’, ‘wb’) as recfile: … recfile.record(duration=5.0) Non-blocking mode (start and stop recording): >>> rec = Recorder(channels=2) >>> with rec.open(‘nonblocking.wav’, ‘wb’) as recfile2: … recfile2.start_recording() … time.sleep(5.0) … recfile2.stop_recording() Original Code from: https://gist.github.com/sloria/5693955

class rtcog.utils.recorder.Recorder(channels=1, rate=44100, frames_per_buffer=1024)

Bases: object

A recorder class for recording audio to a WAV file. Records in mono by default.

open(fname, mode='wb')

class rtcog.utils.recorder.RecordingFile(fname, mode, channels, rate, frames_per_buffer)

Bases: object

close()

get_callback()

record(duration)

start_recording()

stop_recording()

rtcog.utils.shared_memory_manager module

class rtcog.utils.shared_memory_manager.SharedMemoryManager(name: str, create: bool = False, size: int | None = None)

Bases: object

Manager for SharedMemory with open and cleanup logic.

This class wraps multiprocessing.shared_memory.SharedMemory to provide automatic cleanup.

cleanup()

Clean up the shared memory segment.

open() → SharedMemory

Create/attach to shared memory.

rtcog.utils.sync module

class rtcog.utils.sync.ActionState(action_onsets: List[int], action_offsets: List[int], in_action: bool, in_cooldown: bool, cooldown_end: int | None, hit: bool)

Bases: object

Represents the current state of an action block during the experiment.

action_offsets: List[int]

TR indices at which action blocks end.

action_onsets: List[int]

TR indices at which action blocks start.

cooldown_end: int | None

TR index at which the cooldown period ends, or None if no cooldown has occurred.

hit: bool

Whether a hit occurred.

in_action: bool

Whether the experiment is currently within an action block.

in_cooldown: bool

Whether the experiment is currently within a cooldown period.

class rtcog.utils.sync.SyncEvents(hit: Event, action_end: Event, end: Event, new_tr: Event, shm_ready: Event, server_ready: Event, tr_index: Synchronized)

Bases: object

Container for multiprocessing synchronization primitives used in experiment.

action_end: Event

Set when the current action block ends.

end: Event

Set when the entire experiment should terminate.

hit: Event

Set when a hit event occurs.

new_tr: Event

Set when a new TR is received.

server_ready: Event

Set when the server process is ready.

shm_ready: Event

Set when shared memory is ready for access.

tr_index: Synchronized

Shared integer tracking the current TR index.

Module contents