rtcog.matching package

Submodules

rtcog.matching.hit_detector module

class rtcog.matching.hit_detector.HitDetector(hit_opts: HitOpts)

Bases: object

Class for deciding if a TR counts as a hit based on Matcher’s scores

calculate_enorm_diff(motion)

Calculate difference in euclidean norm between this and previous TR. Used for motion thresholding.

detect(t, template_labels, scores, motion=None)

Detect if a hit occured, and if so, return the name of the template.

Parameters:

• t (int) – The current TR.

• template_labels (list of str) – List of template labels.

• scores (np.ndarray) – The scores for each template from Matcher.

• motion (list of float) – The 6 motion parameters

Returns:

The label of the template that qualifies as a hit at time t, or None if no hit is found.

Return type:

str or None

is_hit(t, template_labels, scores)

Determines if a specific time point t represents a “hit” for a template based on if match scores exceed a threshold.

A time point is a hit if:

• No more than nonline templates are >= hit_thr at time t.

  • The one with the highest score is selected as the “hit”.

• That same template has also been above hit_thr nconsec_vols (including the current volume)

Parameters:

• t (int) – The current time point.

• template_labels (list of str) – List of template labels corresponding to the rows of scores.

• scores (np.ndarray of shape (n_templates, n_timepoints)) – Match scores for each template across time thus far.

Returns:

The label of the template that qualifies as a hit at time t, or None if no hit is found.

Return type:

str or None

rtcog.matching.hit_opts module

class rtcog.matching.hit_opts.HitOpts(nconsec_vols: int, hit_thr: float, nonline: int, do_mot: bool, mot_thr: float | None = None)

Bases: object

Configuration options controlling hit detection logic.

do_mot: bool

Whether motion thresholding should be applied.

hit_thr: float

Threshold value that must be exceeded to count as a hit.

mot_thr: float | None = None

Motion threshold value. Required if do_mot is True.

nconsec_vols: int

Number of consecutive volumes required to register a hit.

nonline: int

Maximum number of templates allowed to exceed the hit threshold simultaneously.

If more than nonline templates exceed the threshold at the same time, no hit is registered. E.g. a value of 2 means only a two templates may exceed the threshold for a hit to be counted, and the template with the greatest value is selected as the hit.

rtcog.matching.matcher module

class rtcog.matching.matcher.MaskMatcher(match_opts, Nt, sync, match_path)

Bases: Matcher

Match to templates using pretrained Mask Method

class rtcog.matching.matcher.Matcher(match_opts: MatchingOpts, Nt: int, sync: SyncEvents, match_path: str)

Bases: object

Base class for matching processed TR data to given templates.

This class provides the framework for comparing incoming fMRI volumes against predefined brain state templates to detect patterns of interest. Subclasses implement specific matching algorithms (e.g., SVR-based or mask-based).

registry

Class-level registry mapping matcher names to their classes.

Type:

dict

match_start

First volume index to start computing match scores.

Type:

int

Nt

Total number of time points in the experiment.

Type:

int

scores

Array of match scores, shape (Ntemplates, Nt).

Type:

np.ndarray

Ntemplates

Number of templates to match against.

Type:

int

mp_end

Event to signal experiment end.

Type:

multiprocessing.Event

mp_new_tr

Event set when a new TR is processed.

Type:

multiprocessing.Event

mp_shm_ready

Event indicating shared memory is ready.

Type:

multiprocessing.Event

from_name(name)

Factory method to instantiate a matcher by name.

match(t, n, tr_data)

Compute similarity scores for a TR and update shared memory.

setup_shared_memory()

Initialize shared memory for score storage.

cleanup_shared_memory()

Clean up shared memory resources.

_match(tr_data)

Abstract method for computing match scores (implemented by subclasses).

cleanup_shared_memory()

Clean up shared memory resources.

classmethod from_name(name)

match(t, n, tr_data)

Compute similarity scores for a TR and update shared memory.

Parameters:

• t (int) – Time point index.

• n (int) – Processed volume index.

• tr_data (np.ndarray) – Processed TR data.

Returns:

Updated scores array.

Return type:

np.ndarray

registry = {'mask': <class 'rtcog.matching.matcher.MaskMatcher'>, 'svr': <class 'rtcog.matching.matcher.SVRMatcher'>}

setup_shared_memory()

Initialize shared memory for score storage.

Creates a shared memory buffer to pass match scores to the data streaming process.

class rtcog.matching.matcher.SVRMatcher(match_opts, Nt, sync, match_path)

Bases: Matcher

Match to templates using pretrained SVR model.

This matcher uses a support vector regression model trained on previous data to detect activation patterns in incoming TRs.

rtcog.matching.matching_opts module

class rtcog.matching.matching_opts.MatchingOpts(match_method: str, match_start: int, vols_noaction: int)

Bases: object

match_method: str

match_start: int

vols_noaction: int

rtcog.matching.matching_utils module

rtcog.matching.matching_utils.rt_maskscore_vol(data, inputs, labels)

rtcog.matching.matching_utils.rt_svrscore_vol(data, SVRs, caps_labels)

Compute SVR scores using pretrained models.

Parameters:

• data (np.ndarray) – The input data to be used for making predictions.

• SVRs (dict) – A dictionary of trained Support Vector Regressor (SVR) models, where the keys are label names and the values are the corresponding SVR models.

• caps_labels (list of str) – A list of labels corresponding to the SVRs in SVRs. The function will use these labels to predict the values from the respective SVRs.

Returns:

The predicted values from each SVR for each label.

Return type:

np.ndarray

rtcog.matching.transcribe module

rtcog.matching.transcribe.main()

rtcog.matching.transcribe.transcribe(file, model)

Module contents