Utilities and Supporting Types¶

Processing History¶

processing_history.py

Lightweight provenance tracking for data processing steps.

Design goals¶

JSON-first persistence (portable, diffable)
Minimal intrusion into existing processing code
Captures ordered operations (B) and key state snapshots (C)
Lightweight dataset metadata (A-lite): name, sampling rate, channels, file stats

class tad.processing_history.DatasetInfo(fname, basename, sampling_frequency=None, stream_id=None, channel_ids=None, electrode_labels=None, file_size_bytes=None, file_mtime_utc=None)[source]¶

Bases: object

Lightweight dataset identity/metadata (A-lite).

Parameters:

fname (str)
basename (str)
sampling_frequency (float | None)
stream_id (int | None)
channel_ids (List[Any] | None)
electrode_labels (List[Any] | None)
file_size_bytes (int | None)
file_mtime_utc (str | None)

fname¶

Path (as provided) to the source file.

Type:: str

basename¶

Base filename.

Type:: str

sampling_frequency¶

Sampling frequency in Hz.

Type:: float, optional

stream_id¶

Stream identifier used by the reader (if applicable).

Type:: int, optional

channel_ids¶

Channel identifiers.

Type:: list, optional

electrode_labels¶

Electrode labels (if present).

Type:: list, optional

file_size_bytes¶

File size in bytes (if path exists).

Type:: int, optional

file_mtime_utc¶

File modification time in UTC ISO string.

Type:: str, optional

basename: str¶

channel_ids: List[Any] | None = None¶

electrode_labels: List[Any] | None = None¶

file_mtime_utc: str | None = None¶

file_size_bytes: int | None = None¶

fname: str¶

classmethod from_path(fname, sampling_frequency=None, stream_id=None, channel_ids=None, electrode_labels=None)[source]¶

Parameters:

fname (str)
sampling_frequency (float | None)
stream_id (int | None)
channel_ids (List[Any] | None)
electrode_labels (List[Any] | None)

Return type:

DatasetInfo

sampling_frequency: float | None = None¶

stream_id: int | None = None¶

to_dict()[source]¶

Return type:: Dict[str, Any]

class tad.processing_history.OperationRecord(name, timestamp_utc, params=<factory>, state_before=None, state_after=None, summary=<factory>, artifacts=<factory>)[source]¶

Bases: object

A single processing operation record (B).

Parameters:

name (str)
timestamp_utc (str)
params (Dict[str, Any])
state_before (str | None)
state_after (str | None)
summary (Dict[str, Any])
artifacts (Dict[str, Any])

name¶

Operation name (e.g., “apply_filter”).

Type:: str

timestamp_utc¶

UTC timestamp ISO.

Type:: str

params¶

JSON-friendly parameters.

Type:: dict

state_before¶

Hash of state snapshot before operation.

Type:: str, optional

state_after¶

Hash of state snapshot after operation.

Type:: str, optional

summary¶

Small human-friendly summary (counts, etc.).

Type:: dict, optional

artifacts¶

Any extra “result metadata” (e.g., raster t-range, n_spikes).

Type:: dict, optional

artifacts: Dict[str, Any]¶

name: str¶

params: Dict[str, Any]¶

state_after: str | None = None¶

state_before: str | None = None¶

summary: Dict[str, Any]¶

timestamp_utc: str¶

to_dict()[source]¶

Return type:: Dict[str, Any]

class tad.processing_history.ProcessingHistory(dataset, operations=<factory>, created_utc=<factory>, session_id=<factory>, _state_getter=None)[source]¶

Bases: object

Provenance container for a processing session.

Parameters:

dataset (DatasetInfo) – Lightweight dataset metadata.
session_id (str, optional) – Stable id for this history instance. If None, a hash is generated.
operations (List[OperationRecord])
created_utc (str)
_state_getter (Callable[[], Dict[str, Any]] | None)

created_utc: str¶

dataset: DatasetInfo¶

classmethod from_dict(d)[source]¶

Parameters:: d (Dict[str, Any])
Return type:: ProcessingHistory

classmethod load_json(path)[source]¶

Load history JSON from disk.

Parameters:: path (str)
Return type:: ProcessingHistory

operations: List[OperationRecord]¶

record(name, params=None, *, state_before=None, state_after=None, summary=None, artifacts=None)[source]¶

Append an operation record.

Parameters:

name (str) – Operation name.
params (dict, optional) – Operation parameters (JSON-friendly).
state_before (str, optional) – State hash before op.
state_after (str, optional) – State hash after op.
summary (dict, optional) – Human-friendly summary.
artifacts (dict, optional) – Output-related metadata.

Return type:

None

save_json(path, indent=2)[source]¶

Write history JSON to disk.

Parameters:

path (str)
indent (int)

Return type:

None

session_id: str¶

set_state_getter(getter)[source]¶

Attach a callback that returns a JSON-friendly state snapshot.

Parameters:: getter (Callable[[], Dict[str, Any]])
Return type:: None

snapshot_state()[source]¶

Return a state snapshot using the attached getter, or empty dict.

Return type:: Dict[str, Any]

state_hash(snapshot=None)[source]¶

Compute a stable hash for a given snapshot (or current snapshot if None).

Parameters:: snapshot (Dict[str, Any] | None)
Return type:: str

to_dict()[source]¶

Serialize to a JSON-friendly dict.

Return type:: Dict[str, Any]

to_json(indent=2)[source]¶

Serialize to JSON string.

Parameters:: indent (int)
Return type:: str

tad.processing_history.decode_bool_mask(encoded)[source]¶

Decode a base64-encoded boolean mask payload.

Parameters:: encoded (str) – Base64-encoded mask string.
Returns:: Decoded bytes.
Return type:: bytes

tad.processing_history.encode_bool_mask(mask_bytes)[source]¶

Base64 encode a boolean mask serialized as raw bytes.

Parameters:: mask_bytes (bytes) – Raw bytes representation of a boolean mask (e.g., numpy packbits output).
Returns:: Base64-encoded string.
Return type:: str

tad.processing_history.tracked_operation(name=None, *, track=True, include_result_artifacts=None)[source]¶

Decorator to record a method call into provenance.

It supports two targets (if present on self):

self.processing_history : ProcessingHistory-like object with snapshot_state/state_hash/record
self.history : list[dict] (lightweight generic log)

Parameters:

name (str, optional) – Operation name. Defaults to function name.
track (bool, default=True) – Whether to track this operation.
include_result_artifacts (callable, optional) – Function called with the method return value; must return JSON-friendly dict to attach as artifacts.

Return type:

Callable

Probe and MEA¶

class tad.mea_probe.MEAProbe(positions, contact_shapes=None, shape_params=None, name='MEA_probe')[source]¶

Bases: object

Wrapper around probeinterface.Probe to handle MEA probes and allow filtering/removal of channels, setting shapes, etc.

Parameters:

positions (List[Tuple[float, float]])
contact_shapes (List[str] | None)
shape_params (List[Dict] | None)
name (str)

__init__(positions, contact_shapes=None, shape_params=None, name='MEA_probe')[source]¶

Initialize a RecordingProbe.

Parameters:

positions (list of (x, y)) – Coordinates of each contact.
contact_shapes (list of str, optional) – Shape of each contact (‘circle’, ‘square’, etc.)
shape_params (list of dict, optional) – Parameters for each contact shape (radius, width, etc.)
name (str) – Name of the probe

apply_to_recording(recording)[source]¶: Apply the probe settings to a recording (mask channels, etc.)

get_active_channels()[source]¶

Return a list of channel indices that are not masked

Return type:: List[int]

load(filepath)[source]¶

Load probe information from a saved file

Parameters:: filepath (str)

mask_channels(channels)[source]¶

Mark channels to be ignored in analysis (dead, noisy, stim, reference, etc.)

Parameters:: channels (List[int])
Return type:: None

plot(show_mask=True)[source]¶

Plot the probe. Optionally indicate masked channels.

Parameters:: show_mask (bool)

save(filepath)[source]¶

Save probe information to a JSON or other format

Parameters:: filepath (str)

unmask_channels(channels)[source]¶

Remove channels from the masked list

Parameters:: channels (List[int])
Return type:: None