I/O and Storage

Persistent storage backends, tabular measurement ingest, serialization, and session snapshot/resume. See Results I/O and Typed Configs and Sessions for the narrative guides.

Data I/O and storage module for pytcl.

Provides persistent storage backends for pytcl data including: - HDF5: Efficient storage of large numerical arrays - SQL: Structured data storage and metadata management

Examples

Store tracking data in HDF5:

>>> from pytcl.io import HDF5Storage
>>> with HDF5Storage() as store:
...     store.open("tracking.h5", mode="w")
...     store.store_array("states", track_states)
...     store.store_scalar("num_tracks", 42)

Query structured data in SQL:

>>> from pytcl.io import SQLStorage
>>> with SQLStorage() as store:
...     store.open("tracking.db", mode="a")
...     store.store_group("mission")
...     store.store_scalar("mission/start_time", 1234567890)
...     keys = store.list_keys("mission")

Storage Interface

Abstract storage interface shared by the SQL and HDF5 backends.

Abstract storage interface for pytcl data persistence.

This module provides interfaces for storing and retrieving pytcl data in different formats (HDF5, SQL, etc.).

class pytcl.io.storage.StorageBackend

Bases: ABC

Abstract base class for storage backends.

Provides a unified interface for storing and retrieving arrays, metadata, and structured data in various formats.

abstractmethod open(path, mode='r')

Open a storage file or database connection.

Parameters:

• path (str) – Path to storage file or database URI

• mode (str, optional) – Open mode: ‘r’ (read), ‘w’ (write), ‘a’ (append). Default is ‘r’.

abstractmethod close()

Close the storage connection.

abstractmethod __enter__()

Context manager entry.

abstractmethod __exit__(exc_type, exc_val, exc_tb)

Context manager exit.

abstractmethod store_array(name, data, metadata=None)

Store a numpy array, replacing any array already under that name.

Parameters:

• name (str) – Dataset name/key for storage

• data (ArrayLike) – Numpy array to store

• metadata (dict, optional) – Associated metadata (e.g., units, description). Replaced wholesale along with the array; metadata from a previous store under the same name does not survive.

Notes

The replace-on-collision rule is stated here because the backends used to disagree and neither said so: SQLStorage replaced, while HDF5Storage let h5py raise ValueError on an existing name (gh-21). Code written against one backend broke on the other, and the base class – the only place the contract could live – was silent.

abstractmethod retrieve_array(name)

Retrieve a stored numpy array.

Parameters:

name (str) – Dataset name/key

Returns:

The stored array

Return type:

ndarray

abstractmethod store_scalar(name, value, metadata=None)

Store a scalar value, replacing any scalar already under that name.

Parameters:

• name (str) – Scalar name/key

• value (int, float, str, bool) – Scalar value to store

• metadata (dict, optional) – Associated metadata. Replaced wholesale along with the value.

Notes

Same replace-on-collision contract as store_array, and stated here for the same reason: the gh-21 fix that unified the array path was never applied to the scalar path, so SQLStorage replaced while HDF5Storage let h5py raise ValueError on an existing name – the identical divergence, one method down.

abstractmethod retrieve_scalar(name)

Retrieve a stored scalar value.

Parameters:

name (str) – Scalar name/key

Return type:

Scalar value

abstractmethod store_group(name, metadata=None)

Create a logical group/container for related data.

Parameters:

• name (str) – Group name

• metadata (dict, optional) – Group-level metadata

abstractmethod list_keys(group='/')

List all stored keys/datasets in a group.

Parameters:

group (str, optional) – Group path. Default is root (“/”)

Returns:

Keys in the group

Return type:

list of str

abstractmethod get_metadata(name)

Get metadata associated with a dataset.

Parameters:

name (str) – Dataset name

Returns:

Metadata dictionary

Return type:

dict

abstractmethod delete(name)

Delete a dataset or group.

Parameters:

name (str) – Dataset/group name to delete

abstractmethod flush()

Ensure all data is written to disk.

SQL Storage

SQLite-backed structured storage for metadata and query-driven access.

SQL storage backend for pytcl data persistence.

Provides structured data storage using SQLite. Good for metadata, tracks, measurements, and searchable structured data.

class pytcl.io.sql_storage.SQLStorage

Bases: StorageBackend

SQL-based storage backend.

Stores structured data, metadata, and searchable information using SQL. SQLite only: open() passes its path straight to sqlite3.connect, so a driver connection string is not accepted. The db_type argument that once selected a backend was removed in v2.0.0 (see the __init__ Notes).

Ideal for: - Track metadata and state - Measurements and detections - Sensor configurations - Any structured, queryable data

Examples

>>> from pytcl.io import SQLStorage
>>> with SQLStorage() as store:
...     store.open("tracking.db", mode="w")
...     store.store_array("tracks", track_states)
...     store.store_scalar("mission/start_time", 1234567890)
...     results = store.retrieve_array("tracks")

__init__()

Initialize SQLite storage.

Notes

This used to take a db_type argument documented as accepting “‘sqlite’ (default) or connection string for other databases”. Any value other than 'sqlite' made open() do nothing at all – no connection was established – after which every method raised RuntimeError("Storage not open"). The argument advertised a capability that did not exist, so it has been removed (gh-21). Adding real support for another backend is a feature, not a parameter.

open(path, mode='r')

Open a SQLite database connection.

Parameters:

• path (str) – Path to the SQLite database file.

• mode (str, optional) – ‘r’ (read), ‘w’ (write), ‘a’ (append). Default is ‘r’.

Raises:

• FileNotFoundError – If mode='r' and the file does not exist.

• ValueError – If mode is not one of ‘r’, ‘w’, ‘a’.

Notes

Opening a nonexistent path for reading used to create an empty database, because sqlite3.connect creates the file whatever the caller intended. Reads against it then failed with sqlite3.OperationalError about a missing table rather than the documented KeyError, and the caller was left holding a stray file they never asked for (gh-21).

close()

Close the database connection.

store_array(name, data, metadata=None)

Store array as blob in SQL table.

Parameters:

• name (str) – Array name/key

• data (ArrayLike) – Array to store

• metadata (dict, optional) – Associated metadata

retrieve_array(name)

Retrieve a stored array.

Parameters:

name (str) – Array name/key

Returns:

The reconstructed array

Return type:

ndarray

store_scalar(name, value, metadata=None)

Store a scalar value as metadata.

Parameters:

• name (str) – Scalar name/key

• value (scalar) – Value to store

• metadata (dict, optional) – Associated metadata

retrieve_scalar(name)

Retrieve a scalar value.

Parameters:

name (str) – Scalar name/key

Return type:

Scalar value

store_group(name, metadata=None)

Mark a group/namespace for organization.

Parameters:

• name (str) – Group name

• metadata (dict, optional) – Group-level metadata

list_keys(group='/')

List all keys in storage or a group.

Parameters:

group (str, optional) – Group path prefix. Default is “/” (all keys).

Returns:

Keys in the group

Return type:

list of str

get_metadata(name)

Get metadata for an entry.

Parameters:

name (str) – Entry name

Returns:

Metadata

Return type:

dict

delete(name)

Delete an entry.

Parameters:

name (str) – Entry name

flush()

Commit any pending transactions.

HDF5 Storage

HDF5 storage for large numerical arrays.

HDF5 storage backend for pytcl data persistence.

Provides efficient storage of large numerical arrays using HDF5. h5py is a core pytcl dependency (see pyproject.toml); no optional install step is needed.

class pytcl.io.hdf5_storage.HDF5Storage

Bases: StorageBackend

HDF5-based storage backend.

Efficiently stores large arrays and structured data using HDF5 format. Ideal for numerical data, model coefficients, and time series.

Examples

>>> from pytcl.io import HDF5Storage
>>> with HDF5Storage() as store:
...     store.open("data.h5", mode="w")
...     store.store_array("gravity/egm96", coefficients)
...     store.store_scalar("header/version", 1)
...     result = store.retrieve_array("gravity/egm96")

__init__()

open(path, mode='r')

Open an HDF5 file.

Parameters:

• path (str) – Path to HDF5 file

• mode (str, optional) – ‘r’ (read), ‘w’ (write), ‘a’ (append). Default is ‘r’.

close()

Close the HDF5 file.

store_array(name, data, metadata=None)

Store a numpy array as an HDF5 dataset.

Parameters:

• name (str) – Dataset path (e.g., “groups/subgroup/data”)

• data (ArrayLike) – Array to store

• metadata (dict, optional) – Metadata stored as HDF5 attributes

Notes

Storing over an existing name replaces it, which is the contract StorageBackend defines and what SQLStorage already did. This used to let h5py raise ValueError instead, so the same code worked against one backend and failed against the other (gh-21).

retrieve_array(name)

Retrieve a stored array.

Parameters:

name (str) – Dataset path

Returns:

The stored array

Return type:

ndarray

store_scalar(name, value, metadata=None)

Store a scalar value, replacing any scalar already under that name.

Parameters:

• name (str) – Scalar name/path

• value (scalar) – Value to store

• metadata (dict, optional) – Associated metadata

retrieve_scalar(name)

Retrieve a scalar value.

Parameters:

name (str) – Scalar name/path

Return type:

Scalar value

store_group(name, metadata=None)

Create a group for organizing related datasets.

Parameters:

• name (str) – Group path

• metadata (dict, optional) – Group-level metadata

list_keys(group='/')

List datasets and groups in a location.

Parameters:

group (str, optional) – Group path. Default is root.

Returns:

Keys in the group

Return type:

list of str

get_metadata(name)

Get metadata for a dataset or group.

Parameters:

name (str) – Dataset/group name

Returns:

Metadata attributes

Return type:

dict

delete(name)

Delete a dataset or group.

Parameters:

name (str) – Dataset/group path

flush()

Ensure all data is written to disk.

HDF5 Track Storage

HDF5-backed archival storage for large-scale tracking datasets.

HDF5-backed storage for large-scale tracking datasets.

Optimized for archival and post-analysis of tracking scenarios with efficient time-series access and compression. h5py is a core pytcl dependency (see pyproject.toml); no optional install step is needed.

class pytcl.io.hdf5_track_storage.TrackHDF5Storage(path, chunk_size=1000, compression='gzip', compression_level=4, dtype='float64', shuffle=True)

Bases: object

HDF5-backed storage for large-scale tracking datasets.

Optimized for archival and post-analysis of tracking scenarios with efficient time-series access and compression.

Parameters:

• path (str) – Path to HDF5 file.

• chunk_size (int) – Chunk size for time-series datasets, in rows along the time axis. Datasets shorter than this are stored as a single chunk spanning their full history (time-aligned, not split across tracks), which is the common case for archival writes. Default is 1000.

• compression (str) – Compression algorithm. Default is ‘gzip’.

• compression_level (int) – Compression level (1-9). Default is 4.

• dtype (str) – Default dtype for stored arrays. Default is ‘float64’.

• shuffle (bool) – Enable HDF5’s byte-shuffle filter before compression. Reorders each chunk’s bytes so that same-significance bytes of adjacent float64 values are contiguous, which measurably improves gzip’s ratio on slowly-varying track data (measured +7% on the benchmark scenario in tests/unit/test_hdf5_compression.py – see that file’s module docstring for the reproduction command and full figures). Default is True.

Examples

>>> from pytcl.io import TrackHDF5Storage
>>> with TrackHDF5Storage("scenario.h5") as store:
...     store.open(mode="w")
...     store.store_track("trk_001", states, covariances, timestamps)
...     traj = store.get_track_trajectory("trk_001", start_time=0.0)

__init__(path, chunk_size=1000, compression='gzip', compression_level=4, dtype='float64', shuffle=True)

open(mode='r')

Open HDF5 file.

Parameters:

mode (str) – ‘r’ (read), ‘w’ (write/create), ‘a’ (append). Default is ‘r’.

close()

Close HDF5 file.

flush()

Ensure all data is written to disk.

store_track(track_id, states, covariances, timestamps, metadata=None, residuals=None, scenario_id=None)

Store a complete track trajectory.

Parameters:

• track_id (str) – Unique track identifier.

• states (ArrayLike) – State history, shape (N, state_dim).

• covariances (ArrayLike) – Covariance history, shape (N, state_dim, state_dim).

• timestamps (ArrayLike) – Timestamps, shape (N,).

• metadata (dict, optional) – Track metadata (status, birth_time, etc.).

• residuals (ArrayLike, optional) – Innovation residuals, shape (N, meas_dim).

• scenario_id (str, optional) – Store under /scenarios/{scenario_id}/tracks/.

append_track_state(track_id, state, covariance, timestamp, residual=None, scenario_id=None)

Append a single state to an existing track’s history.

Parameters:

• track_id (str) – Track identifier.

• state (ArrayLike) – State vector.

• covariance (ArrayLike) – Covariance matrix.

• timestamp (float) – Timestamp.

• residual (ArrayLike, optional) – Innovation residual.

• scenario_id (str, optional) – Scenario identifier.

retrieve_track(track_id, scenario_id=None)

Retrieve a complete track.

Parameters:

• track_id (str) – Track identifier.

• scenario_id (str, optional) – Scenario identifier.

Returns:

Keys: states, covariances, timestamps, residuals (or None), metadata.

Return type:

dict

store_detection(detection_id, measurement, timestamp, sensor_id, covariance=None, metadata=None, scenario_id=None)

Store a single detection.

Parameters:

• detection_id (str) – Detection identifier.

• measurement (ArrayLike) – Measurement vector.

• timestamp (float) – Time of detection.

• sensor_id (str) – Sensor source.

• covariance (ArrayLike, optional) – Measurement covariance.

• metadata (dict, optional) – Additional metadata.

• scenario_id (str, optional) – Scenario identifier.

retrieve_detection(detection_id, scenario_id=None)

Retrieve a single detection.

Parameters:

• detection_id (str) – Detection identifier.

• scenario_id (str, optional) – Scenario identifier.

Returns:

Keys: detection_id, measurement, timestamp, sensor_id, covariance (or None), metadata.

Return type:

dict

get_track_trajectory(track_id, start_time=None, end_time=None, scenario_id=None)

Extract a track segment within a time range.

Selects by boolean mask over the full timestamp array – a linear scan, not the binary search previously claimed here. (get_state_at_time is the method that uses searchsorted.)

Parameters:

• track_id (str) – Track identifier.

• start_time (float, optional) – Minimum timestamp (inclusive).

• end_time (float, optional) – Maximum timestamp (inclusive).

• scenario_id (str, optional) – Scenario identifier.

Returns:

Keys: states, covariances, timestamps.

Return type:

dict

get_state_at_time(track_id, time, interpolate=False, scenario_id=None)

Get state at a specific time.

Parameters:

• track_id (str) – Track identifier.

• time (float) – Query time.

• interpolate (bool) – If True, linearly interpolate between nearest states. If False, return nearest state. Default is False.

• scenario_id (str, optional) – Scenario identifier.

Returns:

Keys: state, covariance, timestamp.

Return type:

dict

get_tracks_in_region(bbox, time_range=None, state_indices=(0, 2), scenario_id=None)

Find track IDs with states inside a bounding box.

Parameters:

• bbox (list of float) – [x_min, y_min, x_max, y_max] bounding box.

• time_range (list of float, optional) – [t_min, t_max] time range filter.

• state_indices (tuple of int) – Indices of x and y components in state vector. Default is (0, 2).

• scenario_id (str, optional) – Scenario identifier.

Returns:

Track IDs with states in the region.

Return type:

list of str

store_tracking_scenario(scenario_id, tracks, detections=None, metadata=None)

Store a complete tracking scenario.

Parameters:

• scenario_id (str) – Unique scenario identifier.

• tracks (dict) –

  {track_id: {“states”: ndarray, “covariances”: ndarray,

  ”timestamps”: ndarray, …}}.

• detections (dict, optional) –

  {detection_id: {“measurement”: ndarray, “timestamp”: float,

  ”sensor_id”: str, …}}.

• metadata (dict, optional) – Scenario-level metadata.

retrieve_tracking_scenario(scenario_id)

Retrieve a complete scenario.

Parameters:

scenario_id (str) – Scenario identifier.

Returns:

Keys: tracks, detections, metadata.

Return type:

dict

list_scenarios()

List all stored scenario IDs.

Returns:

Scenario identifiers.

Return type:

list of str

list_tracks(scenario_id=None)

List all track IDs.

Parameters:

scenario_id (str, optional) – If provided, list tracks within this scenario.

Returns:

Track identifiers.

Return type:

list of str

list_detections(scenario_id=None)

List all detection IDs.

Parameters:

scenario_id (str, optional) – If provided, list detections within this scenario.

Returns:

Detection identifiers.

Return type:

list of str

compare_scenarios(scenario_id1, scenario_id2)

Compare two scenarios.

Parameters:

• scenario_id1 (str) – First scenario.

• scenario_id2 (str) – Second scenario.

Returns:

Keys: common_tracks, unique_to_1, unique_to_2, state_differences (dict of track_id -> RMSE).

Return type:

dict

export_to_sql(db_manager, scenario_id=None)

Export HDF5 track data to a TrackDatabaseManager.

Parameters:

• db_manager (TrackDatabaseManager) – Target SQL database manager (must be open for writing).

• scenario_id (str, optional) – Scenario to export. If None, exports standalone tracks.

import_from_sql(db_manager, scenario_id)

Import tracks and detections from SQL into HDF5 as a scenario.

Parameters:

• db_manager (TrackDatabaseManager) – Source SQL database manager (must be open for reading).

• scenario_id (str) – Scenario ID for the imported data.

Track Database

SQL-backed track lifecycle management: detections, initiation, maintenance.

SQL-backed track lifecycle database manager.

Provides detection management, track initiation, state maintenance, and lifecycle operations using SQLite for real-time tracking scenarios.

class pytcl.io.track_database.TrackDatabaseStatus(value)

Bases: Enum

Track lifecycle status for database persistence.

Extends the runtime TrackStatus with COASTING and DEAD states for full lifecycle management.

TENTATIVE = 'tentative'

CONFIRMED = 'confirmed'

COASTING = 'coasting'

DEAD = 'dead'

class pytcl.io.track_database.TrackDatabaseManager(path)

Bases: object

SQL-backed track lifecycle database manager.

Provides detection management, track initiation, state maintenance, and lifecycle operations using SQLite.

Parameters:

path (str) – Path to SQLite database file.

Examples

>>> import os
>>> import tempfile
>>> from pytcl.io import TrackDatabaseManager
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "tracking.db")
>>> with TrackDatabaseManager(path) as db:
...     db.open(mode="w")
...     db.store_detection("det_001", np.array([1.0, 2.0]), "radar", 0.0)
...     db.initiate_track("trk_001", np.array([1, 2, 0, 0]),
...                       np.eye(4), 0.0)
...     db.update_track_state("trk_001", np.array([1.1, 2.1, 0, 0]),
...                           np.eye(4) * 0.9, 1.0)
>>> tmpdir.cleanup()

__init__(path)

open(mode='a')

Open database connection.

Parameters:

mode (str) – ‘r’ (read), ‘w’ (write/create), ‘a’ (append). Default is ‘a’.

Raises:

• ValueError – If mode is not one of ‘r’, ‘w’, ‘a’.

• FileNotFoundError – If mode='r' and the database does not exist.

Notes

Read mode does not create the database. It used to: sqlite3.connect was called unconditionally, so opening a mistyped path for reading produced an empty file and the first query then failed with no such table: detections – reporting a missing table rather than the missing database the caller actually had. This is the same defect gh-21 fixed for SQLStorage, which was not applied here at the time.

close()

Close database connection.

store_detection(detection_id, measurement, sensor_id, timestamp, covariance=None, metadata=None)

Store a raw detection/measurement.

Parameters:

• detection_id (str) – Unique detection identifier.

• measurement (ArrayLike) – Measurement vector.

• sensor_id (str) – Sensor source identifier.

• timestamp (float) – Time of detection.

• covariance (ArrayLike, optional) – Measurement covariance matrix.

• metadata (dict, optional) – Additional metadata.

retrieve_detections(start_time=None, end_time=None, sensor_id=None, association_status=None, limit=None)

Query detections by time range, sensor, or association status.

Parameters:

• start_time (float, optional) – Minimum timestamp (inclusive).

• end_time (float, optional) – Maximum timestamp (inclusive).

• sensor_id (str, optional) – Filter by sensor.

• association_status (str, optional) – Filter by status (‘unassociated’, ‘associated’, ‘clutter’).

• limit (int, optional) – Maximum number of results.

Returns:

Detection records with keys: detection_id, timestamp, sensor_id, measurement, covariance, association_status, associated_track_id, association_confidence, metadata – everything _row_to_detection builds, not the subset previously listed here.

Return type:

list of dict

retrieve_detection(detection_id)

Retrieve a single detection by ID.

Parameters:

detection_id (str) – Detection identifier.

Returns:

Detection record.

Return type:

dict

Raises:

KeyError – If detection not found.

retrieve_all_detections()

Retrieve all detections.

Returns:

All detection records.

Return type:

list of dict

associate_detection(detection_id, track_id, confidence=1.0)

Link a detection to a track.

Parameters:

• detection_id (str) – Detection identifier.

• track_id (str) – Track identifier.

• confidence (float) – Association confidence score (0-1). Default is 1.0.

initiate_track(track_id, initial_state, initial_covariance, timestamp, metadata=None)

Create a new tentative track with initial state estimate.

Parameters:

• track_id (str) – Unique track identifier.

• initial_state (ArrayLike) – Initial state vector.

• initial_covariance (ArrayLike) – Initial covariance matrix.

• timestamp (float) – Track birth time.

• metadata (dict, optional) – Additional track metadata.

get_initiation_queue(max_age=None)

Get unassociated detections awaiting track initiation.

Parameters:

max_age (float, optional) – Maximum age in seconds. If provided, only returns detections newer than (max_timestamp - max_age).

Returns:

Unassociated detection records.

Return type:

list of dict

confirm_track(track_id)

Promote a tentative track to confirmed status.

Parameters:

track_id (str) – Track identifier.

update_track_state(track_id, state, covariance, timestamp, residual=None, update_type='update')

Record a filter state update.

Parameters:

• track_id (str) – Track identifier.

• state (ArrayLike) – Updated state vector.

• covariance (ArrayLike) – Updated covariance matrix.

• timestamp (float) – Time of update.

• residual (ArrayLike, optional) – Innovation/residual vector.

• update_type (str) – Type of update: ‘prediction’, ‘update’, or ‘smoothed’. Default is ‘update’.

Raises:

KeyError – If track_id is not a track in this database.

Notes

An unknown track_id used to insert the state row anyway and then update zero rows in the tracks table, leaving history that belongs to no track (gh-21). Nothing surfaced: no error, and the state was retrievable by id, so a typo produced a track that existed in every respect except the one that counts.

store_track_history(track_id, states, covariances, timestamps, residuals=None)

Batch-store an entire state timeline for a track.

Parameters:

• track_id (str) – Track identifier.

• states (ArrayLike) – State history, shape (N, state_dim).

• covariances (ArrayLike) – Covariance history, shape (N, state_dim, state_dim).

• timestamps (ArrayLike) – Timestamps, shape (N,).

• residuals (ArrayLike, optional) – Residual history, shape (N, meas_dim).

get_track_state(track_id)

Get the most recent state estimate for a track.

Parameters:

track_id (str) – Track identifier.

Returns:

Keys: track_id, state, covariance, timestamp, status, hits, misses.

Return type:

dict

get_track_history(track_id, start_time=None, end_time=None)

Get state history for a track within a time range.

Parameters:

• track_id (str) – Track identifier.

• start_time (float, optional) – Minimum timestamp (inclusive).

• end_time (float, optional) – Maximum timestamp (inclusive).

Returns:

Keys: states (N, state_dim), covariances (N, state_dim, state_dim), timestamps (N,), residuals (N, meas_dim) or None.

residuals is row-aligned with timestamps. Rows that carry no residual – predictions and initiations – hold NaN, so np.isnan(residuals).any(axis=1) identifies them. It is None only when no row in the range has a residual at all.

Return type:

dict

Raises:

KeyError – If the track has no history in the given range.

Notes

Residuals used to be keyed off the first row. A window beginning with a prediction reported residuals=None even when later rows had them, and a window mixing the two returned an array shorter than timestamps with no indication of which rows it belonged to – breaking the documented (N, meas_dim) shape and silently misaligning every residual with the wrong timestamp (gh-21).

That is the shape a predict-then-update filter produces on every step, which is what KalmanTrackAdapter does, so it was the normal case rather than an edge one.

get_track(track_id)

Get full track metadata.

Parameters:

track_id (str) – Track identifier.

Returns:

Track metadata with keys: track_id, status, birth_time, last_update_time, state_dim, hits, misses, total_misses, confidence_score, metadata.

Return type:

dict

retrieve_all_tracks(status=None)

Retrieve all tracks, optionally filtered by status.

Parameters:

status (TrackDatabaseStatus, optional) – Filter by status.

Returns:

Track metadata records.

Return type:

list of dict

mark_track_tentative(track_id)

Set track status to TENTATIVE.

mark_track_confirmed(track_id)

Set track status to CONFIRMED.

mark_track_coasting(track_id)

Set track status to COASTING.

mark_track_dead(track_id)

Set track status to DEAD.

prune_old_detections(age_threshold)

Remove unassociated detections older than threshold.

Parameters:

age_threshold (float) – Maximum age in seconds relative to newest detection.

Returns:

Number of detections pruned.

Return type:

int

prune_dead_tracks(age_threshold)

Remove tracks that have been DEAD for longer than threshold.

Removes the track record, its state history, and associations.

Parameters:

age_threshold (float) – Maximum age in seconds relative to newest track update.

Returns:

Number of tracks pruned.

Return type:

int

merge_tracks(track_id_keep, track_id_merge)

Merge one track into another.

Combines state histories, re-associates detections from the merged track to the kept track, and marks the merged track DEAD.

Parameters:

• track_id_keep (str) – Track to keep.

• track_id_merge (str) – Track to merge in and mark dead.

Raises:

KeyError – If either track is absent from the database.

Notes

Track-level fields are combined as well as the history: birth_time becomes the earlier of the two, last_update_time the later, and the merged track’s metadata keys are folded in without overwriting keys the kept track already has.

None of that used to happen (gh-21). History, associations and detections were re-assigned and the hit counters summed, but the kept track’s last_update_time was left behind – so a merge that brought in newer states made the track look stale, and any staleness-based pruning would then delete the track that had just been reinforced.

track_to_pytcl(track_id)

Convert stored track to pytcl Track NamedTuple.

Parameters:

track_id (str) – Track identifier.

Returns:

pytcl Track NamedTuple.

Return type:

Track

tracks_to_tracklist(status=None)

Convert stored tracks to pytcl TrackList.

Parameters:

status (TrackDatabaseStatus, optional) – Filter by status.

Returns:

pytcl TrackList container.

Return type:

TrackList

store_from_track(track, timestamp=None)

Store a pytcl Track NamedTuple into the database.

Parameters:

• track (Track) – pytcl Track NamedTuple.

• timestamp (float, optional) – Override timestamp. Uses track.time if not provided.

store_from_tracklist(track_list)

Store all tracks from a TrackList.

Parameters:

track_list (TrackList) – pytcl TrackList container.

Measurement Readers

CSV and Parquet readers for tabular measurement data.

CSV and Parquet readers for tabular measurement data.

Both readers turn a flat table (one row per measurement report) into a MeasurementSet: measurements grouped into scans by the exact value of a timestamp column, ascending. This is the ingest side of the pipeline whose export side is pytcl.io.dataframes – a table written by metrics_to_polars (or any tool) with a time column and one column per measurement component reads back through these functions unchanged.

polars is the parsing engine for both readers (the dataframe extra): read_measurements_csv uses pl.read_csv, read_measurements_parquet uses pl.read_parquet, and both feed the same downstream grouping so a CSV and a Parquet file holding identical data produce identical MeasurementSet values – including dtype-inferred columns like a numeric id_column. polars is imported lazily by both, mirroring the guard in pytcl.io.dataframes (_import_polars/_dependency_error), so importing this module never requires polars to be installed; calling either reader without it raises DependencyError.

class pytcl.io.readers.MeasurementSet(times, scans, ids)

Bases: NamedTuple

Measurements grouped into scans by exact timestamp.

Variables:

• times (ndarray, shape (n_scans,)) – Unique scan timestamps, strictly ascending, float64.

• scans (list of ndarray) – scans[k] holds the (n_k, n_cols) float64 measurement rows recorded at times[k], columns in measurement_columns order.

• ids (list of ndarray, or None) – ids[k] holds one identifier per row of scans[k], when id_column was given to the reader; None otherwise.

times: ndarray[tuple[Any, ...], dtype[float64]]

Alias for field number 0

scans: list[ndarray[tuple[Any, ...], dtype[float64]]]

Alias for field number 1

ids: list[ndarray[tuple[Any, ...], dtype[Any]]] | None

Alias for field number 2

pytcl.io.readers.read_measurements_csv(path, *, time_column, measurement_columns, id_column=None)

Read measurement reports from a CSV file into a MeasurementSet.

Parsed with polars (pl.read_csv); column dtypes are polars’ inferred schema, so e.g. an all-integer id_column reads back as a numpy integer array rather than strings – the same as read_measurements_parquet.

Parameters:

• path (str or path-like) – CSV file to read; must have a header row.

• time_column (str) – Name of the column holding each row’s scan timestamp. Rows with the exact same value (compared as parsed float) are grouped into the same scan.

• measurement_columns (sequence of str) – Names of the columns to stack into each scan’s measurement matrix, in order.

• id_column (str, optional) – Name of a column holding a per-row identifier. When given, the returned MeasurementSet.ids carries one array per scan; when omitted, MeasurementSet.ids is None.

Returns:

Scans ordered by ascending unique timestamp.

Return type:

MeasurementSet

Raises:

• ValueError – If time_column, any of measurement_columns, or id_column is not a column in the file; the message lists the available columns.

• DependencyError – If polars is not installed.

Examples

>>> import tempfile, os
>>> from pytcl.io.readers import read_measurements_csv
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "meas.csv")
>>> _ = open(path, "w").write(
...     "t,x,y\n0.0,1.0,2.0\n0.0,3.0,4.0\n1.0,5.0,6.0\n"
... )
>>> ms = read_measurements_csv(path, time_column="t",
...                             measurement_columns=["x", "y"])
>>> ms.times.tolist()
[0.0, 1.0]
>>> ms.scans[0].tolist()
[[1.0, 2.0], [3.0, 4.0]]
>>> ms.ids is None
True
>>> tmpdir.cleanup()

pytcl.io.readers.read_measurements_parquet(path, *, time_column, measurement_columns, id_column=None)

Read measurement reports from a Parquet file into a MeasurementSet.

Same grouping and column-mapping contract as read_measurements_csv; see that function’s docstring.

Parameters:

• path (str or path-like) – Parquet file to read.

• time_column (str) – Name of the column holding each row’s scan timestamp. Rows with the exact same value are grouped into the same scan.

• measurement_columns (sequence of str) – Names of the columns to stack into each scan’s measurement matrix, in order.

• id_column (str, optional) – Name of a column holding a per-row identifier.

Returns:

Scans ordered by ascending unique timestamp.

Return type:

MeasurementSet

Raises:

• ValueError – If time_column, any of measurement_columns, or id_column is not a column in the file; the message lists the available columns.

• DependencyError – If polars is not installed.

Examples

>>> import tempfile, os
>>> import polars as pl
>>> from pytcl.io.readers import read_measurements_parquet
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "meas.parquet")
>>> pl.DataFrame(
...     {"t": [0.0, 0.0, 1.0], "x": [1.0, 3.0, 5.0], "y": [2.0, 4.0, 6.0]}
... ).write_parquet(path)
>>> ms = read_measurements_parquet(path, time_column="t",
...                                 measurement_columns=["x", "y"])
>>> ms.times.tolist()
[0.0, 1.0]
>>> ms.scans[0].tolist()
[[1.0, 2.0], [3.0, 4.0]]
>>> tmpdir.cleanup()

DataFrame Accessors

polars accessors for track histories and scalar metrics (dataframe extra).

polars DataFrame accessors for track histories and scalar metrics.

polars is an optional dependency (the dataframe extra). It is imported lazily inside _import_polars, so importing this module never requires polars to be installed; calling any of the public functions without it raises DependencyError.

None of the public signatures below mention a polars type by name (the boundary rule for optional heavy dependencies) — return values are annotated Any and documented as polars.DataFrame in each docstring.

pytcl.io.dataframes.tracks_to_polars(history, times)

Flatten a per-scan track history into a long-format polars DataFrame.

Parameters:

• history (sequence of sequence of Track-like) – Per-scan lists of objects exposing id, state, covariance, and status (a TrackStatus enum or plain str) — the same shape consumed by pytcl.io.serialize.encode_tracks().

• times (sequence of float) – Timestamp for each scan in history; len(times) == len(history).

Returns:

One row per (scan, track) pair, columns track_id (Int64), t (Float64), status (String), state (List[Float64]), covariance (List[Float64], row-major flattened, length len(state) ** 2).

Return type:

polars.DataFrame

Raises:

DependencyError – If polars is not installed.

Examples

>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> df = tracks_to_polars([[track]], [0.0])
>>> df.columns
['track_id', 't', 'status', 'state', 'covariance']
>>> df.height
1

pytcl.io.dataframes.explode_state_columns(df, layout)

Add one Float64 column per state-vector component.

Parameters:

• df (polars.DataFrame) – A DataFrame with a state column of type List[Float64] (as produced by tracks_to_polars).

• layout (sequence of str) – Column name for each state-vector component, in order; its length must equal the dimension of the vectors in state.

Returns:

df with one extra Float64 column per layout entry, appended via df.with_columns.

Return type:

polars.DataFrame

Raises:

• ValueError – If layout’s length does not match the state dimension. Not raised for a zero-row df: the dimension can’t be read from empty data, so the check is skipped and layout-named columns are added empty instead.

• DependencyError – If polars is not installed.

Examples

>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> df = tracks_to_polars([[track]], [0.0])
>>> wide = explode_state_columns(df, ["x", "vx"])
>>> wide.row(0, named=True)["x"]
1.0

pytcl.io.dataframes.metrics_to_polars(times, **series)

Assemble scalar-per-scan metric series into a flat polars DataFrame.

Parameters:

• times (sequence of float) – Timestamp for each scan; becomes the t column.

• **series (array_like) – Named 1-D metric series (e.g. ospa=ospa_values), each of length len(times); each becomes a Float64 column of the same name.

Returns:

Columns ["t", *series] in the order times then series were given, all Float64.

Return type:

polars.DataFrame

Raises:

• ValueError – If a series is not 1-D or its length does not match times.

• DependencyError – If polars is not installed.

Examples

>>> import numpy as np
>>> df = metrics_to_polars(np.arange(3.0), ospa=np.array([1.0, 0.5, 0.1]))
>>> df.columns
['t', 'ospa']
>>> df["ospa"].to_list()
[1.0, 0.5, 0.1]

Serialization

msgspec JSON and MessagePack serialization for filter states and tracks.

msgspec-based serialization for filter states and track histories.

Two wire formats are supported via the fmt argument on every function:

• "msgpack" (default): compact binary, round-trips float64 bit patterns exactly, including NaN/inf.

• "json": human-readable text. JSON has no representation for NaN/inf, so encoding a state or covariance containing non-finite values raises ValueError rather than silently producing invalid JSON.

Decoding is strict: msgspec validates the incoming bytes against the target msgspec.Struct and raises on missing fields, wrong types, or malformed data rather than returning partial/garbage results.

class pytcl.io.serialize.TrackRecord(track_id, t, status, state, covariance)

Bases: Struct

One track’s state at one scan, ready for msgspec encoding.

Variables:

• track_id (int) – Track identifier.

• t (float) – Timestamp of the scan this record belongs to.

• status (str) – Track status (TrackStatus.value, e.g. "confirmed").

• state (list of float) – State estimate vector.

• covariance (list of float) – Row-major flattened state covariance; len == len(state) ** 2.

track_id: int

t: float

status: str

state: list[float]

covariance: list[float]

class pytcl.io.serialize.TrackSet(times, scans)

Bases: Struct

A full track history: scan timestamps plus per-scan track records.

Variables:

• times (list of float) – Timestamp for each scan.

• scans (list of list of TrackRecord) – Per-scan lists of track records, aligned with times.

times: list[float]

scans: list[list[TrackRecord]]

class pytcl.io.serialize.StateRecord(x, p_flat)

Bases: Struct

A single filter state estimate and its flattened covariance.

Variables:

• x (list of float) – State estimate vector.

• p_flat (list of float) – Row-major flattened covariance; len == len(x) ** 2.

x: list[float]

p_flat: list[float]

class pytcl.io.serialize.SimpleTrack(id, state, covariance, status)

Bases: NamedTuple

A decoded track: plain data, no tracker-class dependency.

Variables:

• id (int) – Track identifier.

• state (ndarray) – State estimate vector.

• covariance (ndarray, shape (n, n)) – State covariance matrix.

• status (str) – Track status value.

id: int

Alias for field number 0

state: ndarray[tuple[Any, ...], dtype[float64]]

Alias for field number 1

covariance: ndarray[tuple[Any, ...], dtype[float64]]

Alias for field number 2

status: str

Alias for field number 3

pytcl.io.serialize.encode_tracks(history, times, fmt='msgpack')

Serialize a per-scan track history to bytes.

Parameters:

• history (sequence of sequence of Track-like) – Per-scan lists of objects exposing id, state, covariance, and status (a TrackStatus enum or plain str).

• times (sequence of float) – Timestamp for each scan in history; len(times) == len(history).

• fmt ({"msgpack", "json"}, optional) – Wire format. With "json", any non-finite state or covariance value raises ValueError before encoding.

Returns:

Encoded track history, decodable with decode_tracks.

Return type:

bytes

Examples

>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> blob = encode_tracks([[track]], [0.0], fmt="json")
>>> times, history = decode_tracks(blob, fmt="json")
>>> times
[0.0]
>>> t2 = history[0][0]
>>> (t2.id, t2.status)
(1, 'confirmed')
>>> [round(v, 3) for v in t2.state.tolist()]
[1.0, 2.0]

pytcl.io.serialize.decode_tracks(data, fmt='msgpack')

Deserialize a track history produced by encode_tracks.

Parameters:

• data (bytes) – Encoded track history.

• fmt ({"msgpack", "json"}, optional) – Wire format data was encoded with.

Returns:

• times (list of float) – Timestamp for each scan.

• history (list of list of SimpleTrack) – Per-scan lists of decoded tracks, aligned with times.

Raises:

msgspec.ValidationError or msgspec.DecodeError – If data does not match the expected structure.

Return type:

tuple[list[float], list[list[SimpleTrack]]]

Examples

>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> blob = encode_tracks([[track]], [0.0], fmt="msgpack")
>>> times, history = decode_tracks(blob, fmt="msgpack")
>>> times
[0.0]
>>> history[0][0].covariance.tolist()
[[1.0, 0.0], [0.0, 1.0]]

pytcl.io.serialize.encode_states(x, P, fmt='msgpack')

Serialize a single filter state estimate and covariance to bytes.

Parameters:

• x (array_like) – State estimate vector, shape (n,).

• P (array_like) – State covariance matrix, shape (n, n).

• fmt ({"msgpack", "json"}, optional) – Wire format. With "json", non-finite values in x or P raise ValueError before encoding.

Returns:

Encoded state, decodable with decode_states.

Return type:

bytes

Examples

>>> import numpy as np
>>> x = np.array([1.0, 2.0])
>>> P = np.eye(2)
>>> blob = encode_states(x, P, fmt="json")
>>> x2, P2 = decode_states(blob, fmt="json")
>>> [round(v, 3) for v in x2.tolist()]
[1.0, 2.0]
>>> P2.tolist()
[[1.0, 0.0], [0.0, 1.0]]

pytcl.io.serialize.decode_states(data, fmt='msgpack')

Deserialize a state estimate and covariance produced by encode_states.

Parameters:

• data (bytes) – Encoded state.

• fmt ({"msgpack", "json"}, optional) – Wire format data was encoded with.

Returns:

• x (ndarray, shape (n,)) – State estimate vector.

• P (ndarray, shape (n, n)) – State covariance matrix.

Raises:

msgspec.ValidationError or msgspec.DecodeError – If data does not match the expected structure.

Return type:

tuple[ndarray[tuple[Any, …], dtype[float64]], ndarray[tuple[Any, …], dtype[float64]]]

Examples

>>> import numpy as np
>>> x = np.array([1.0, 2.0, 3.0])
>>> P = np.eye(3)
>>> x2, P2 = decode_states(encode_states(x, P, fmt="msgpack"), fmt="msgpack")
>>> x2.tolist()
[1.0, 2.0, 3.0]
>>> P2.shape
(3, 3)

ASDF Export

ASDF archival export/import (asdf extra).

ASDF export/import for pytcl track histories and single filter states.

ASDF is an optional dependency (the asdf extra). It is imported lazily inside _import_asdf, so importing this module never requires asdf to be installed; calling any of the public functions without it raises DependencyError.

Track histories are flattened into parallel arrays under a pytcl/tracks tree, one row per (scan, track) record – the same flattening pytcl.io.dataframes.tracks_to_polars uses, but written as an ASDF/ndarray tree rather than a DataFrame. States must be the same dimension across every record in a single history (save_tracks_asdf raises ValueError otherwise); save_states_asdf stores a single state/covariance pair the same way, without the track bookkeeping.

None of the public signatures below mention an asdf type by name (the boundary rule for optional heavy dependencies established in pytcl.io.dataframes) – path is a plain path, and results are plain tuples of Python/numpy objects.

pytcl.io.asdf_io.save_tracks_asdf(path, history, times)

Write a per-scan track history to an ASDF file.

Parameters:

• path (str or path-like) – Destination ASDF file; overwritten if it exists.

• history (sequence of sequence of Track-like) – Per-scan lists of objects exposing id, state, covariance, and status (a TrackStatus enum or plain str) – the same shape consumed by pytcl.io.serialize.encode_tracks().

• times (sequence of float) – Timestamp for each scan in history; len(times) == len(history).

Raises:

• ValueError – If the state vectors in history are not all the same dimension.

• DependencyError – If asdf is not installed.

Examples

>>> import tempfile, os
>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "tracks.asdf")
>>> save_tracks_asdf(path, [[track]], [0.0])
>>> times, history = load_tracks_asdf(path)
>>> times
[0.0]
>>> tmpdir.cleanup()

pytcl.io.asdf_io.load_tracks_asdf(path)

Read a track history written by save_tracks_asdf.

Parameters:

path (str or path-like) – ASDF file to read.

Returns:

• times (list of float) – Timestamp for each scan.

• history (list of list of SimpleTrack) – Per-scan lists of decoded tracks, aligned with times.

Raises:

DependencyError – If asdf is not installed.

Return type:

tuple[list[float], list[list[SimpleTrack]]]

Examples

>>> import tempfile, os
>>> import numpy as np
>>> from pytcl.trackers import Track, TrackStatus
>>> track = Track(id=1, state=np.array([1.0, 2.0]),
...                covariance=np.eye(2), status=TrackStatus.CONFIRMED,
...                hits=1, misses=0, time=0.0)
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "tracks.asdf")
>>> save_tracks_asdf(path, [[track]], [0.0])
>>> times, history = load_tracks_asdf(path)
>>> history[0][0].id
1
>>> tmpdir.cleanup()

pytcl.io.asdf_io.save_states_asdf(path, x, P)

Write a single filter state estimate and covariance to an ASDF file.

Parameters:

• path (str or path-like) – Destination ASDF file; overwritten if it exists.

• x (array_like) – State estimate vector, shape (n,).

• P (array_like) – State covariance matrix, shape (n, n).

Raises:

DependencyError – If asdf is not installed.

Examples

>>> import tempfile, os
>>> import numpy as np
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "state.asdf")
>>> save_states_asdf(path, np.array([1.0, 2.0]), np.eye(2))
>>> x2, P2 = load_states_asdf(path)
>>> x2.tolist()
[1.0, 2.0]
>>> tmpdir.cleanup()

pytcl.io.asdf_io.load_states_asdf(path)

Read a state estimate and covariance written by save_states_asdf.

Parameters:

path (str or path-like) – ASDF file to read.

Returns:

• x (ndarray, shape (n,)) – State estimate vector.

• P (ndarray, shape (n, n)) – State covariance matrix.

Raises:

DependencyError – If asdf is not installed.

Return type:

tuple[ndarray[tuple[Any, …], dtype[float64]], ndarray[tuple[Any, …], dtype[float64]]]

Examples

>>> import tempfile, os
>>> import numpy as np
>>> tmpdir = tempfile.TemporaryDirectory()
>>> path = os.path.join(tmpdir.name, "state.asdf")
>>> save_states_asdf(path, np.array([1.0, 2.0, 3.0]), np.eye(3))
>>> x2, P2 = load_states_asdf(path)
>>> P2.shape
(3, 3)
>>> tmpdir.cleanup()

Sessions

Full tracker/filter state snapshot and resume.

Session save/restore: full tracker/filter state snapshot and resume.

A “session” is a self-describing snapshot of a stateful tracker/filter object – config, current estimate, and anything else needed to resume predict/update calls exactly where they left off. Two wire formats are supported via the fmt argument, matching pytcl.io.serialize:

• "msgpack" (default): round-trips float64 bit patterns exactly, including NaN/inf.

• "json": human-readable text. JSON has no representation for NaN/inf, so saving a session containing non-finite values raises ValueError rather than silently producing invalid JSON.

Decoding is strict: malformed or truncated bytes, or bytes from a newer schema version, raise FormatError.

Some snapshotted objects (trackers built with callable dynamics rather than fixed matrices) cannot be fully rebuilt from the snapshot alone – loading such a session without the matching keyword argument raises ConfigurationError.

pytcl.io.session.save_session(obj, *, fmt='msgpack')

Serialize a tracker/filter’s full state to bytes.

Parameters:

• obj (SingleTargetTracker, MultiTargetTracker, MHTTracker, IMMEstimator, GaussianSumFilter, or RBPFFilter) – The object to snapshot.

• fmt ({"msgpack", "json"}, optional) – Wire format. With "json", non-finite values anywhere in the snapshot raise ValueError before encoding.

Returns:

Encoded session, decodable with load_session.

Return type:

bytes

Raises:

• ConfigurationError – If obj’s type has no registered snapshotter.

• ValueError – If fmt is "json" and obj contains non-finite values.

pytcl.io.session.save_session_file(obj, path, *, fmt='msgpack')

Serialize a tracker/filter’s full state to a file.

Parameters:

• obj (SingleTargetTracker, MultiTargetTracker, MHTTracker, IMMEstimator, GaussianSumFilter, or RBPFFilter) – The object to snapshot.

• path (str or Path) – Destination file path.

• fmt ({"msgpack", "json"}, optional) – Wire format; see save_session.

pytcl.io.session.load_session(data, *, fmt='msgpack', **models)

Deserialize a tracker/filter’s full state from bytes.

Parameters:

• data (bytes) – Encoded session produced by save_session.

• fmt ({"msgpack", "json"}, optional) – Wire format data was encoded with.

• **models (Any) – Rehydration arguments for snapshots that could not capture callable dynamics (F=/Q= for a SingleTargetTracker, MultiTargetTracker, or MHTTracker built with callable F/Q). Consumed only where the snapshot actually needs them, one matrix at a time: if the snapshot’s config already has a matrix for F (or Q), passing that kwarg raises rather than silently overriding the saved dynamics; if the config lacks it, omitting the kwarg raises rather than restoring a tracker that cannot predict. Snapshot types with no callable-dynamics escape hatch at all (IMMEstimator, GaussianSumFilter, RBPFFilter – these take models per predict/update call, not at construction) are fully self-contained and reject every keyword argument.

Returns:

The restored tracker/filter, resumable via its normal predict/update API.

Return type:

object

Raises:

• FormatError – If data is malformed, or was produced by a newer schema version than this pytcl supports.

• ConfigurationError – If the snapshot needs a rehydration keyword argument that was not supplied, or was given one it does not need (either because the snapshot’s config already carries that matrix, or because the snapshot type is fully self-contained and takes none at all).

pytcl.io.session.load_session_file(path, *, fmt='msgpack', **models)

Deserialize a tracker/filter’s full state from a file.

Parameters:

• path (str or Path) – Source file path, as written by save_session_file.

• fmt ({"msgpack", "json"}, optional) – Wire format; see load_session.

• **models (Any) – Rehydration arguments; see load_session.

Returns:

The restored tracker/filter.

Return type:

object

Migration Tools

Utilities for moving v1.x tracking pipelines and stored data to v2.x.

Migration tools for transitioning v1.x tracking pipelines to v2.0.0.

Provides utilities for analyzing legacy code, converting data formats, and generating v2.0.0 template code using TrackDatabaseManager and TrackHDF5Storage.

Examples

Analyze a legacy tracking script:

>>> from pytcl.io.migration import MigrationHelper
>>> helper = MigrationHelper()
>>> analysis = helper.analyze_v1_code("legacy_tracker.py")
>>> print(analysis.recommendations)

Convert legacy pickle tracks to SQL:

>>> helper.convert_legacy_tracks_to_sql(
...     legacy_data={"trk_0": {"states": states, "times": times}},
...     output_db="new_tracks.db",
... )

Generate a v2.0.0 template:

>>> template = helper.generate_v2_template(backend="sql")
>>> print(template)

class pytcl.io.migration.AnalysisResult

Bases: object

Result of analyzing a v1.x tracking pipeline.

Variables:

• filter_types (list of str) – Detected filter types (e.g., [‘kf’, ‘ekf’, ‘ukf’]).

• storage_patterns (list of str) – Detected storage patterns (e.g., [‘pickle’, ‘numpy’, ‘csv’]).

• recommendations (list of str) – Suggested migration steps.

• estimated_complexity (str) – ‘low’, ‘medium’, or ‘high’.

• detected_imports (list of str) – pytcl imports found in the source.

__init__()

summary()

Return a human-readable summary.

class pytcl.io.migration.MigrationHelper

Bases: object

Utilities for migrating v1.x tracking pipelines to v2.0.0.

Provides code analysis, data conversion, and template generation.

Examples

>>> helper = MigrationHelper()
>>> result = helper.analyze_v1_code("my_tracker.py")
>>> print(result.summary())

analyze_v1_code(source)

Analyze a v1.x tracking pipeline for migration.

Parameters:

source (str) – Either a file path or a string of Python source code.

Returns:

Analysis with detected patterns and recommendations.

Return type:

AnalysisResult

convert_legacy_tracks_to_sql(legacy_data, output_db)

Convert legacy track data to a SQL database.

Parameters:

• legacy_data (dict) – Dictionary of track_id -> {“states”: ndarray (N, state_dim), “covariances”: ndarray (N, state_dim, state_dim) or None, “timestamps”: ndarray (N,) or None}. Covariances default to identity if not provided. Timestamps default to 0..N-1 if not provided.

• output_db (str) – Path to the output SQLite database.

Returns:

Number of tracks converted.

Return type:

int

convert_legacy_tracks_to_hdf5(legacy_data, output_h5, scenario_id='migrated')

Convert legacy track data to an HDF5 archive.

Parameters:

• legacy_data (dict) – Same format as convert_legacy_tracks_to_sql.

• output_h5 (str) – Path to the output HDF5 file.

• scenario_id (str) – Scenario identifier. Default is ‘migrated’.

Returns:

Number of tracks converted.

Return type:

int

generate_v2_template(backend='sql', filter_type='kf', n_targets=3)

Generate v2.0.0 template code for a tracking pipeline.

Parameters:

• backend (str) – Target backend: ‘sql’, ‘hdf5’, or ‘both’. Default is ‘sql’.

• filter_type (str) – Filter type. Default is ‘kf’. Only three distinct templates exist – (‘sql’, ‘kf’), (‘sql’, ‘ekf’) which ‘ukf’ also uses, and one apiece for the ‘hdf5’ and ‘both’ backends. Anything else, including ‘imm’ and ‘particle’, falls back to the nearest template for that backend; the returned source names which one it is in its own header.

• n_targets (int) – Number of targets in the template. Default is 3.

Returns:

Python source code for the tracking pipeline template.

Return type:

str

Notes

The parameter list previously read “‘kf’, ‘ekf’, ‘ukf’, ‘imm’, ‘particle’” as though each had its own template. Four templates exist. 'imm' and 'particle' return Kalman scaffolding, and the ‘hdf5’ and ‘both’ backends ignore filter_type entirely – deliberate, but not what the list implied.

static generate_migration_checklist()

Generate a migration validation checklist.

Returns:

Markdown-formatted checklist.

Return type:

str

v1.x Compatibility Adapters

Adapters connecting v1.x filter outputs to the v2.x storage layer.

Backward compatibility adapters for v1.x filter integration.

Provides adapter classes that connect pytcl’s existing filter outputs (KF, EKF, UKF, IMM, particle filters, MHT) to SQL persistence via TrackDatabaseManager. Every adapter in this module persists exclusively through self._db (SQL); none of them import or call TrackHDF5Storage. HDF5 archival is a separate, explicit step performed after the fact with TrackHDF5Storage.import_from_sql() – it is not something these adapters do automatically.

JPDA is not adapted here: pytcl has no stateful JPDA tracker class (only assignment functions in pytcl.assignment_algorithms.jpda), and MultiTargetTracker/TrackerDatabaseAdapter do not reference JPDA at all, so there is no path – direct or indirect – from JPDA to these adapters.

Examples

Wrap a Kalman filter loop with SQL persistence:

>>> from pytcl.io.compat import KalmanTrackAdapter
>>> adapter = KalmanTrackAdapter(db, "trk_001", F, H, Q, R)
>>> adapter.initialize(x0, P0, timestamp=0.0)
>>> for k, z in enumerate(measurements):
...     adapter.predict_update(z, timestamp=float(k + 1))

Use with MultiTargetTracker:

>>> from pytcl.io.compat import TrackerDatabaseAdapter
>>> adapter = TrackerDatabaseAdapter(db, tracker)
>>> for k, measurements in enumerate(scans):
...     adapter.process_scan(measurements, dt=1.0, timestamp=float(k))

class pytcl.io.compat.KalmanTrackAdapter(db, track_id, F, H, Q, R)

Bases: object

Adapter connecting Kalman filter predict/update to SQL storage.

Wraps a single track’s filter state and persists every predict/update into a TrackDatabaseManager.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Unique track identifier.

• F (NDArray) – State transition matrix.

• H (NDArray) – Measurement matrix.

• Q (NDArray) – Process noise covariance.

• R (NDArray) – Measurement noise covariance.

Examples

>>> adapter = KalmanTrackAdapter(db, "trk_001", F, H, Q, R)
>>> adapter.initialize(x0, P0, timestamp=0.0)
>>> adapter.predict(dt=1.0, timestamp=1.0)
>>> adapter.update(z, timestamp=1.0)

__init__(db, track_id, F, H, Q, R)

property state: ndarray[tuple[Any, ...], dtype[float64]] | None

Current state estimate.

property covariance: ndarray[tuple[Any, ...], dtype[float64]] | None

Current covariance estimate.

property track_id: str

Track identifier.

initialize(x0, P0, timestamp=0.0, metadata=None)

Initialize the track with an initial state estimate.

Parameters:

• x0 (ArrayLike) – Initial state vector.

• P0 (ArrayLike) – Initial covariance matrix.

• timestamp (float) – Birth time.

• metadata (dict, optional) – Additional metadata.

predict(timestamp)

Run Kalman prediction and store result.

Parameters:

timestamp (float) – Time of prediction.

update(measurement, timestamp, detection_id=None, sensor_id='default')

Run Kalman update and store result.

Parameters:

• measurement (ArrayLike) – Measurement vector.

• timestamp (float) – Time of measurement.

• detection_id (str, optional) – Detection identifier. Auto-generated if not provided.

• sensor_id (str) – Sensor identifier.

predict_update(measurement, timestamp, detection_id=None, sensor_id='default')

Combined predict + update step.

Parameters:

• measurement (ArrayLike) – Measurement vector.

• timestamp (float) – Time of measurement.

• detection_id (str, optional) – Detection identifier.

• sensor_id (str) – Sensor identifier.

class pytcl.io.compat.EKFTrackAdapter(db, track_id, f, F_func, h, H_func, Q, R)

Bases: object

Adapter connecting Extended Kalman filter to SQL storage.

Similar to KalmanTrackAdapter but uses nonlinear dynamics/measurement functions with Jacobians.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Unique track identifier.

• f (callable) – State transition function f(x) -> x_pred.

• F_func (callable) – Jacobian of f, F_func(x) -> F matrix.

• h (callable) – Measurement function h(x) -> z_pred.

• H_func (callable) – Jacobian of h, H_func(x) -> H matrix.

• Q (NDArray) – Process noise covariance.

• R (NDArray) – Measurement noise covariance.

__init__(db, track_id, f, F_func, h, H_func, Q, R)

property state: ndarray[tuple[Any, ...], dtype[float64]] | None

Current state estimate.

property covariance: ndarray[tuple[Any, ...], dtype[float64]] | None

Current covariance estimate.

initialize(x0, P0, timestamp=0.0)

Initialize the track.

predict(timestamp)

EKF prediction step.

update(measurement, timestamp)

EKF update step.

class pytcl.io.compat.UKFTrackAdapter(db, track_id, f, h, Q, R, alpha=0.001, beta=2.0, kappa=0.0)

Bases: object

Adapter connecting Unscented Kalman filter to SQL storage.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Unique track identifier.

• f (callable) – State transition function.

• h (callable) – Measurement function.

• Q (NDArray) – Process noise covariance.

• R (NDArray) – Measurement noise covariance.

• alpha (float) – UKF spread parameter.

• beta (float) – UKF distribution parameter.

• kappa (float) – UKF scaling parameter.

__init__(db, track_id, f, h, Q, R, alpha=0.001, beta=2.0, kappa=0.0)

property state: ndarray[tuple[Any, ...], dtype[float64]] | None

Current state estimate.

property covariance: ndarray[tuple[Any, ...], dtype[float64]] | None

Current covariance estimate.

initialize(x0, P0, timestamp=0.0)

Initialize the track.

predict(timestamp)

UKF prediction step.

update(measurement, timestamp)

UKF update step.

class pytcl.io.compat.TrackerDatabaseAdapter(db, tracker, confirm_hits=3, max_misses=5)

Bases: object

Adapter connecting MultiTargetTracker or MHTTracker to SQL storage.

Automatically persists track state after each scan, handles initiation and deletion, and keeps the database in sync with the tracker.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• tracker (object) – A pytcl tracker with a process(measurements, dt) method that returns a list of Track-like objects. (process_scan was previously documented as an alternative; the adapter only ever calls process, and a tracker exposing just process_scan raises AttributeError.)

• confirm_hits (int) – Number of hits to confirm a track. Default is 3.

• max_misses (int) – Consecutive misses before marking dead. Default is 5.

Examples

>>> adapter = TrackerDatabaseAdapter(db, tracker)
>>> for k, meas in enumerate(scans):
...     tracks = adapter.process_scan(meas, dt=1.0, timestamp=float(k))

__init__(db, tracker, confirm_hits=3, max_misses=5)

process_scan(measurements, dt, timestamp, sensor_id='default')

Process a scan of measurements and persist results.

Parameters:

• measurements (sequence of ArrayLike) – Measurement vectors for this scan.

• dt (float) – Time step since last scan.

• timestamp (float) – Current time.

• sensor_id (str) – Sensor identifier for stored detections.

Returns:

Track objects from the underlying tracker.

Return type:

list

get_track_list(status=None)

Get current tracks as a pytcl TrackList.

Parameters:

status (TrackDatabaseStatus, optional) – Filter by status.

Returns:

Container of pytcl Track objects.

Return type:

TrackList

class pytcl.io.compat.IMMTrackAdapter(db, track_id, imm)

Bases: object

Adapter connecting IMM estimator to SQL storage.

Persists the combined IMM state after each predict/update. Mode probabilities are NOT stored – they are exposed read-only via the mode_probs property while the filter object lives, but nothing writes them to track metadata, so they do not survive a round trip through storage. (This previously claimed they were stored.)

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Unique track identifier.

• imm (object) – An IMMEstimator instance.

__init__(db, track_id, imm)

initialize(x0, P0, timestamp=0.0, mode_probs=None)

Initialize the IMM and store initial state.

predict(timestamp)

IMM prediction step.

update(measurement, timestamp)

IMM update step.

property mode_probs: ndarray[tuple[Any, ...], dtype[float64]]

Current mode probabilities.

class pytcl.io.compat.ParticleFilterTrackAdapter(db, track_id, f, h, Q, R, n_particles=200)

Bases: object

Adapter connecting particle filter to SQL storage.

Like the other adapters in this module, persistence is SQL-only via self._db; it does not write to TrackHDF5Storage. HDF5 archival of the resulting SQL records is a separate step: run TrackHDF5Storage.import_from_sql() after the fact.

Stores the weighted mean and covariance from the particle cloud as the track state after each step.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Unique track identifier.

• f (callable) – State transition function.

• h (callable) – Measurement function.

• Q (NDArray) – Process noise covariance.

• R (NDArray) – Measurement noise covariance.

• n_particles (int) – Number of particles.

__init__(db, track_id, f, h, Q, R, n_particles=200)

initialize(x0, P0, timestamp=0.0)

Initialize particles around an initial state.

predict_update(measurement, timestamp)

Particle filter step: predict, update, store.

property particles: ndarray[tuple[Any, ...], dtype[float64]] | None

Current particle cloud.

property weights: ndarray[tuple[Any, ...], dtype[float64]] | None

Current particle weights.

pytcl.io.compat.store_filter_result(db, track_id, result, timestamp, update_type='update')

Store any filter result (KalmanUpdate, IMMUpdate, etc.) in the database.

Extracts x and P attributes from the result and stores them. Works with KalmanPrediction, KalmanUpdate, IMMUpdate, SRKalmanUpdate, etc.

Parameters:

• db (TrackDatabaseManager) – Open database connection.

• track_id (str) – Track identifier.

• result (object) – Filter result with .x and .P attributes. For square-root results with .S, computes P = S @ S.T.

• timestamp (float) – Time of the result.

• update_type (str) – ‘prediction’, ‘update’, or ‘smoothed’.