magscope.tracking_data

Contents

magscope.tracking_data#

Attributes#

logger

_NANOSECONDS_PER_SECOND

_MAX_UINT16

_BATCH_CHUNK

_FRAME_CHUNK

_RECORD_CHUNK

TRACKING_DATA_WRITER_FAILURE_WARNING

Classes#

TrackingDataBatch

CloseTrackingDataFiles

TrackingHDF5File

TrackingDataWriter

Process objects represent activity that is run in a separate process

Functions#

timestamps_to_epoch_ns(→ numpy.ndarray)

build_tracking_data_batch(→ TrackingDataBatch)

tracking_data_path(→ pathlib.Path)

_to_uint16_array(→ numpy.ndarray)

_tracking_file_bucket_start_ns(→ tuple[int, int])

_file_key(→ tuple[int, str, bool, int])

_format_timestamp_for_filename(→ str)

Module Contents#

magscope.tracking_data.logger[source]#
magscope.tracking_data._NANOSECONDS_PER_SECOND = 1000000000[source]#
magscope.tracking_data._MAX_UINT16[source]#
magscope.tracking_data._BATCH_CHUNK = 1024[source]#
magscope.tracking_data._FRAME_CHUNK = 4096[source]#
magscope.tracking_data._RECORD_CHUNK = 65536[source]#
magscope.tracking_data.TRACKING_DATA_WRITER_FAILURE_WARNING = 'tracking_data_writer_failure'[source]#
class magscope.tracking_data.TrackingDataBatch[source]#
recording_id: int[source]#
recording_start_ns: int[source]#
batch_sequence: int[source]#
acquisition_dir: str[source]#
include_roi_positions: bool[source]#
file_start_ns: int[source]#
max_file_duration_ns: int | None[source]#
frame_timestamps_ns: numpy.ndarray[source]#
frame_offsets: numpy.ndarray[source]#
bead_ids: numpy.ndarray[source]#
positions_nm: numpy.ndarray[source]#
roi_positions_px: numpy.ndarray | None = None[source]#
class magscope.tracking_data.CloseTrackingDataFiles[source]#
recording_id: int | None = None[source]#
magscope.tracking_data.timestamps_to_epoch_ns(timestamps: numpy.ndarray) → numpy.ndarray[source]#
magscope.tracking_data.build_tracking_data_batch(*, recording_id: int, acquisition_dir: str, timestamps: numpy.ndarray, tracks: numpy.ndarray, n_rois: int, include_roi_positions: bool, recording_start_ns: int | None = None, max_file_duration_ns: int | None = None, batch_sequence: int = 0) → TrackingDataBatch[source]#
magscope.tracking_data.tracking_data_path(directory: str | pathlib.Path, first_timestamp_ns: int) → pathlib.Path[source]#
class magscope.tracking_data.TrackingHDF5File(path: str | pathlib.Path, *, include_roi_positions: bool, recording_id: int | None = None, recording_start_ns: int | None = None, file_start_ns: int | None = None, rotation_interval_ns: int | None = None)[source]#
path[source]#
include_roi_positions[source]#
_file[source]#
_group[source]#
classmethod open_existing(path: str | pathlib.Path, *, include_roi_positions: bool, recording_id: int | None = None, recording_start_ns: int | None = None, file_start_ns: int | None = None, rotation_interval_ns: int | None = None) → TrackingHDF5File[source]#
_create_datasets(group) → None[source]#
_bind_datasets(group) → None[source]#
_validate_existing_group(group, *, recording_id: int | None, recording_start_ns: int | None, file_start_ns: int | None, rotation_interval_ns: int | None) → None[source]#
static _validate_existing_attr(group, name: str, expected: int | None) → None[source]#
append(batch: TrackingDataBatch) → None[source]#
close() → None[source]#
_update_timestamp_range(frame_timestamps_ns: numpy.ndarray) → None[source]#
_append_rollback_state() → dict[str, object][source]#
_rollback_append(state: dict[str, object]) → None[source]#
class magscope.tracking_data.TrackingDataWriter(queue: multiprocessing.queues.Queue, warning_queue: QueueType | None = None)[source]#

Bases: multiprocessing.Process

Process objects represent activity that is run in a separate process

The class is analogous to threading.Thread

_queue[source]#
_warning_queue = None[source]#
run() → None[source]#

Method to be run in sub-process; can be overridden in sub-class

_handle_close_request(request: CloseTrackingDataFiles, *, open_files: dict[tuple[int, str, bool, int], TrackingHDF5File], closed_keys: set[tuple[int, str, bool, int]], closed_recording_ids: set[int]) → None[source]#
_close_rotated_files(batch: TrackingDataBatch, *, open_files: dict[tuple[int, str, bool, int], TrackingHDF5File], closed_keys: set[tuple[int, str, bool, int]]) → None[source]#
_open_tracking_file(path: pathlib.Path, batch: TrackingDataBatch, *, append_existing: bool) → TrackingHDF5File[source]#
static _should_close_after_append(key: tuple[int, str, bool, int], batch: TrackingDataBatch, *, closed_keys: set[tuple[int, str, bool, int]], closed_recording_ids: set[int]) → bool[source]#
_report_writer_failure(batch: TrackingDataBatch, path: pathlib.Path | None, exc: Exception) → None[source]#
magscope.tracking_data._to_uint16_array(values: numpy.ndarray, field_name: str) → numpy.ndarray[source]#
magscope.tracking_data._tracking_file_bucket_start_ns(first_frame_timestamp_ns: int, *, recording_start_ns: int | None, max_file_duration_ns: int | None) → tuple[int, int][source]#
magscope.tracking_data._file_key(batch: TrackingDataBatch) → tuple[int, str, bool, int][source]#
magscope.tracking_data._format_timestamp_for_filename(timestamp_ns: int) → str[source]#
Contents