Source code for pytcl.io.track_database

"""SQL-backed track lifecycle database manager.

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

import json
import sqlite3
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
from numpy.typing import ArrayLike, NDArray


[docs] class TrackDatabaseStatus(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"
[docs] class TrackDatabaseManager: """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() """
[docs] def __init__(self, path: str) -> None: self._path = path self._connection: Optional[sqlite3.Connection] = None self._cursor: Optional[sqlite3.Cursor] = None self._mode: Optional[str] = None
[docs] def open(self, mode: str = "a") -> None: """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 :class:`~pytcl.io.sql_storage.SQLStorage`, which was not applied here at the time. """ if mode not in ("r", "w", "a"): raise ValueError(f"mode must be 'r', 'w' or 'a', got {mode!r}") if mode == "r" and not Path(self._path).exists(): raise FileNotFoundError( f"No such database: {self._path}. Opening for reading does " f"not create one; use mode='w' or mode='a' to create it." ) self._mode = mode self._connection = sqlite3.connect(self._path) self._cursor = self._connection.cursor() if mode in ("w", "a"): self._initialize_tables()
[docs] def close(self) -> None: """Close database connection.""" if self._connection is not None: self._connection.commit() self._connection.close() self._connection = None self._cursor = None
def __enter__(self) -> "TrackDatabaseManager": return self def __exit__( self, exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Optional[Any], ) -> None: self.close() # ========================================================================= # Detection Management # =========================================================================
[docs] def store_detection( self, detection_id: str, measurement: ArrayLike, sensor_id: str, timestamp: float, covariance: Optional[ArrayLike] = None, metadata: Optional[Dict[str, Any]] = None, ) -> 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. """ self._check_open() meas = np.asarray(measurement, dtype=np.float64) meas_bytes, meas_dim, meas_dtype = self._serialize_array(meas) cov_bytes = None if covariance is not None: cov = np.asarray(covariance, dtype=np.float64) cov_bytes = cov.tobytes() meta_json = self._sanitize_metadata(metadata) self._cursor.execute( """INSERT INTO detections (detection_id, timestamp, sensor_id, measurement, measurement_dim, measurement_dtype, covariance, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( detection_id, timestamp, sensor_id, meas_bytes, meas_dim, meas_dtype, cov_bytes, meta_json, ), ) self._connection.commit()
[docs] def retrieve_detections( self, start_time: Optional[float] = None, end_time: Optional[float] = None, sensor_id: Optional[str] = None, association_status: Optional[str] = None, limit: Optional[int] = None, ) -> List[Dict[str, Any]]: """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 ------- list of dict 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. """ self._check_open() conditions: List[str] = [] params: List[Any] = [] if start_time is not None: conditions.append("timestamp >= ?") params.append(start_time) if end_time is not None: conditions.append("timestamp <= ?") params.append(end_time) if sensor_id is not None: conditions.append("sensor_id = ?") params.append(sensor_id) if association_status is not None: conditions.append("association_status = ?") params.append(association_status) query = "SELECT * FROM detections" if conditions: query += " WHERE " + " AND ".join(conditions) query += " ORDER BY timestamp" if limit is not None: query += f" LIMIT {limit}" self._cursor.execute(query, params) return [self._row_to_detection(row) for row in self._cursor.fetchall()]
[docs] def retrieve_detection(self, detection_id: str) -> Dict[str, Any]: """Retrieve a single detection by ID. Parameters ---------- detection_id : str Detection identifier. Returns ------- dict Detection record. Raises ------ KeyError If detection not found. """ self._check_open() self._cursor.execute( "SELECT * FROM detections WHERE detection_id = ?", (detection_id,) ) row = self._cursor.fetchone() if row is None: raise KeyError(f"Detection '{detection_id}' not found") return self._row_to_detection(row)
[docs] def retrieve_all_detections(self) -> List[Dict[str, Any]]: """Retrieve all detections. Returns ------- list of dict All detection records. """ return self.retrieve_detections()
[docs] def associate_detection( self, detection_id: str, track_id: str, confidence: float = 1.0, ) -> None: """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. """ self._check_open() # Get detection timestamp self._cursor.execute( "SELECT timestamp FROM detections WHERE detection_id = ?", (detection_id,), ) row = self._cursor.fetchone() if row is None: raise KeyError(f"Detection '{detection_id}' not found") timestamp = row[0] # Update detection status self._cursor.execute( """UPDATE detections SET association_status = 'associated', associated_track_id = ?, association_confidence = ? WHERE detection_id = ?""", (track_id, confidence, detection_id), ) # Record in associations table self._cursor.execute( """INSERT OR REPLACE INTO track_associations (track_id, detection_id, timestamp, confidence) VALUES (?, ?, ?, ?)""", (track_id, detection_id, timestamp, confidence), ) self._connection.commit()
# ========================================================================= # Track Initiation # =========================================================================
[docs] def initiate_track( self, track_id: str, initial_state: ArrayLike, initial_covariance: ArrayLike, timestamp: float, metadata: Optional[Dict[str, Any]] = None, ) -> 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. """ self._check_open() state = np.asarray(initial_state, dtype=np.float64) cov = np.asarray(initial_covariance, dtype=np.float64) meta_json = self._sanitize_metadata(metadata) self._cursor.execute( """INSERT INTO tracks (track_id, status, birth_time, last_update_time, state_dim, hits, misses, total_misses, confidence_score, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( track_id, TrackDatabaseStatus.TENTATIVE.value, timestamp, timestamp, len(state), 0, 0, 0, 0.0, meta_json, ), ) # Store initial state self._store_state_row(track_id, state, cov, timestamp, update_type="update") self._connection.commit()
[docs] def get_initiation_queue( self, max_age: Optional[float] = None, ) -> List[Dict[str, Any]]: """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 ------- list of dict Unassociated detection records. """ self._check_open() if max_age is not None: self._cursor.execute("SELECT MAX(timestamp) FROM detections") max_ts = self._cursor.fetchone()[0] if max_ts is None: return [] cutoff = max_ts - max_age return self.retrieve_detections( start_time=cutoff, association_status="unassociated" ) return self.retrieve_detections(association_status="unassociated")
[docs] def confirm_track(self, track_id: str) -> None: """Promote a tentative track to confirmed status. Parameters ---------- track_id : str Track identifier. """ self._set_track_status(track_id, TrackDatabaseStatus.CONFIRMED)
# ========================================================================= # Track State Maintenance # =========================================================================
[docs] def update_track_state( self, track_id: str, state: ArrayLike, covariance: ArrayLike, timestamp: float, residual: Optional[ArrayLike] = None, update_type: str = "update", ) -> None: """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. """ self._check_open() s = np.asarray(state, dtype=np.float64) c = np.asarray(covariance, dtype=np.float64) self._cursor.execute("SELECT 1 FROM tracks WHERE track_id = ?", (track_id,)) if self._cursor.fetchone() is None: raise KeyError( f"No track {track_id!r} in this database. Create it with " f"initiate_track() before recording state updates." ) self._store_state_row(track_id, s, c, timestamp, residual, update_type) # Update track metadata if update_type == "update": self._cursor.execute( """UPDATE tracks SET last_update_time = ?, hits = hits + 1, misses = 0 WHERE track_id = ?""", (timestamp, track_id), ) elif update_type == "prediction": self._cursor.execute( """UPDATE tracks SET last_update_time = ?, misses = misses + 1, total_misses = total_misses + 1 WHERE track_id = ?""", (timestamp, track_id), ) else: self._cursor.execute( """UPDATE tracks SET last_update_time = ? WHERE track_id = ?""", (timestamp, track_id), ) self._connection.commit()
[docs] def store_track_history( self, track_id: str, states: ArrayLike, covariances: ArrayLike, timestamps: ArrayLike, residuals: Optional[ArrayLike] = None, ) -> 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). """ self._check_open() s_arr = np.asarray(states, dtype=np.float64) c_arr = np.asarray(covariances, dtype=np.float64) t_arr = np.asarray(timestamps, dtype=np.float64) n = len(t_arr) for i in range(n): res = None if residuals is not None: res = np.asarray(residuals, dtype=np.float64)[i] self._store_state_row( track_id, s_arr[i], c_arr[i], float(t_arr[i]), residual=res, update_type="update", ) self._connection.commit()
[docs] def get_track_state(self, track_id: str) -> Dict[str, Any]: """Get the most recent state estimate for a track. Parameters ---------- track_id : str Track identifier. Returns ------- dict Keys: track_id, state, covariance, timestamp, status, hits, misses. """ self._check_open() # Get latest state self._cursor.execute( """SELECT state, covariance, state_dim, timestamp, residual FROM track_states WHERE track_id = ? ORDER BY timestamp DESC LIMIT 1""", (track_id,), ) state_row = self._cursor.fetchone() if state_row is None: raise KeyError(f"No state found for track '{track_id}'") # Get track metadata track_info = self.get_track(track_id) state = self._deserialize_vector(state_row[0], state_row[2]) cov = self._deserialize_matrix(state_row[1], state_row[2]) return { "track_id": track_id, "state": state, "covariance": cov, "timestamp": state_row[3], "status": TrackDatabaseStatus(track_info["status"]), "hits": track_info["hits"], "misses": track_info["misses"], }
[docs] def get_track_history( self, track_id: str, start_time: Optional[float] = None, end_time: Optional[float] = None, ) -> Dict[str, Any]: """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 ------- dict 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. 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. """ self._check_open() conditions = ["track_id = ?"] params: List[Any] = [track_id] if start_time is not None: conditions.append("timestamp >= ?") params.append(start_time) if end_time is not None: conditions.append("timestamp <= ?") params.append(end_time) query = ( "SELECT state, covariance, state_dim, timestamp, residual " "FROM track_states WHERE " + " AND ".join(conditions) + " ORDER BY timestamp" ) self._cursor.execute(query, params) rows = self._cursor.fetchall() if not rows: raise KeyError(f"No history found for track '{track_id}'") state_dim = rows[0][2] states = np.array([self._deserialize_vector(r[0], state_dim) for r in rows]) covariances = np.array( [self._deserialize_matrix(r[1], state_dim) for r in rows] ) timestamps = np.array([r[3] for r in rows]) # Scan every row, not just the first: a window can begin with a # prediction and still contain residuals further in. stored = [ np.frombuffer(r[4], dtype=np.float64) if r[4] is not None else None for r in rows ] present = [res for res in stored if res is not None] residuals = None if present: # One row per state, NaN where there is no residual, so the result # stays aligned with `timestamps` and keeps the documented shape. meas_dim = len(present[0]) residuals = np.full((len(rows), meas_dim), np.nan) for index, res in enumerate(stored): if res is not None: residuals[index] = res return { "states": states, "covariances": covariances, "timestamps": timestamps, "residuals": residuals, }
[docs] def get_track(self, track_id: str) -> Dict[str, Any]: """Get full track metadata. Parameters ---------- track_id : str Track identifier. Returns ------- dict Track metadata with keys: track_id, status, birth_time, last_update_time, state_dim, hits, misses, total_misses, confidence_score, metadata. """ self._check_open() self._cursor.execute("SELECT * FROM tracks WHERE track_id = ?", (track_id,)) row = self._cursor.fetchone() if row is None: raise KeyError(f"Track '{track_id}' not found") return self._row_to_track(row)
[docs] def retrieve_all_tracks( self, status: Optional[TrackDatabaseStatus] = None, ) -> List[Dict[str, Any]]: """Retrieve all tracks, optionally filtered by status. Parameters ---------- status : TrackDatabaseStatus, optional Filter by status. Returns ------- list of dict Track metadata records. """ self._check_open() if status is not None: self._cursor.execute( "SELECT * FROM tracks WHERE status = ?", (status.value,) ) else: self._cursor.execute("SELECT * FROM tracks") return [self._row_to_track(row) for row in self._cursor.fetchall()]
# ========================================================================= # Track Lifecycle Management # =========================================================================
[docs] def mark_track_tentative(self, track_id: str) -> None: """Set track status to TENTATIVE.""" self._set_track_status(track_id, TrackDatabaseStatus.TENTATIVE)
[docs] def mark_track_confirmed(self, track_id: str) -> None: """Set track status to CONFIRMED.""" self._set_track_status(track_id, TrackDatabaseStatus.CONFIRMED)
[docs] def mark_track_coasting(self, track_id: str) -> None: """Set track status to COASTING.""" self._set_track_status(track_id, TrackDatabaseStatus.COASTING)
[docs] def mark_track_dead(self, track_id: str) -> None: """Set track status to DEAD.""" self._set_track_status(track_id, TrackDatabaseStatus.DEAD)
[docs] def prune_old_detections(self, age_threshold: float) -> int: """Remove unassociated detections older than threshold. Parameters ---------- age_threshold : float Maximum age in seconds relative to newest detection. Returns ------- int Number of detections pruned. """ self._check_open() self._cursor.execute("SELECT MAX(timestamp) FROM detections") max_ts = self._cursor.fetchone()[0] if max_ts is None: return 0 cutoff = max_ts - age_threshold self._cursor.execute( """DELETE FROM detections WHERE association_status = 'unassociated' AND timestamp < ?""", (cutoff,), ) count = self._cursor.rowcount self._connection.commit() return count
[docs] def prune_dead_tracks(self, age_threshold: float) -> int: """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 ------- int Number of tracks pruned. """ self._check_open() self._cursor.execute("SELECT MAX(last_update_time) FROM tracks") max_ts = self._cursor.fetchone()[0] if max_ts is None: return 0 cutoff = max_ts - age_threshold # Find dead tracks to prune self._cursor.execute( """SELECT track_id FROM tracks WHERE status = ? AND last_update_time < ?""", (TrackDatabaseStatus.DEAD.value, cutoff), ) track_ids = [row[0] for row in self._cursor.fetchall()] if not track_ids: return 0 placeholders = ",".join("?" for _ in track_ids) # Delete state history self._cursor.execute( f"DELETE FROM track_states WHERE track_id IN ({placeholders})", track_ids, ) # Delete associations self._cursor.execute( f"DELETE FROM track_associations WHERE track_id IN ({placeholders})", track_ids, ) # Delete tracks self._cursor.execute( f"DELETE FROM tracks WHERE track_id IN ({placeholders})", track_ids, ) self._connection.commit() return len(track_ids)
[docs] def merge_tracks( self, track_id_keep: str, track_id_merge: str, ) -> None: """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. """ self._check_open() for track_id in (track_id_keep, track_id_merge): self._cursor.execute("SELECT 1 FROM tracks WHERE track_id = ?", (track_id,)) if self._cursor.fetchone() is None: raise KeyError(f"No track {track_id!r} in this database") # Re-assign state history self._cursor.execute( """UPDATE track_states SET track_id = ? WHERE track_id = ?""", (track_id_keep, track_id_merge), ) # Re-assign associations self._cursor.execute( """UPDATE track_associations SET track_id = ? WHERE track_id = ?""", (track_id_keep, track_id_merge), ) # Re-assign detections self._cursor.execute( """UPDATE detections SET associated_track_id = ? WHERE associated_track_id = ?""", (track_id_keep, track_id_merge), ) # Fold the merged track's counters, lifetime and metadata into the # kept one. The lifetime is the union of the two, so birth_time takes # the earlier and last_update_time the later. self._cursor.execute( "SELECT hits, total_misses, birth_time, last_update_time, metadata " "FROM tracks WHERE track_id = ?", (track_id_merge,), ) merged_hits, merged_misses, merged_birth, merged_last, merged_meta = ( self._cursor.fetchone() ) self._cursor.execute( "SELECT metadata FROM tracks WHERE track_id = ?", (track_id_keep,) ) kept_meta = self._cursor.fetchone()[0] # The kept track's own keys win: merging is additive, not a takeover. combined = {**json.loads(merged_meta or "{}"), **json.loads(kept_meta or "{}")} self._cursor.execute( """UPDATE tracks SET hits = hits + ?, total_misses = total_misses + ?, birth_time = MIN(birth_time, ?), last_update_time = MAX(last_update_time, ?), metadata = ? WHERE track_id = ?""", ( merged_hits, merged_misses, merged_birth, merged_last, json.dumps(combined), track_id_keep, ), ) # Mark merged track dead self._set_track_status(track_id_merge, TrackDatabaseStatus.DEAD) self._connection.commit()
# ========================================================================= # Conversion Helpers # =========================================================================
[docs] def track_to_pytcl(self, track_id: str) -> Any: """Convert stored track to pytcl Track NamedTuple. Parameters ---------- track_id : str Track identifier. Returns ------- Track pytcl Track NamedTuple. """ from pytcl.trackers.multi_target import Track, TrackStatus track_info = self.get_track(track_id) state_info = self.get_track_state(track_id) # Map database status to TrackStatus status_map = { "tentative": TrackStatus.TENTATIVE, "confirmed": TrackStatus.CONFIRMED, "coasting": TrackStatus.CONFIRMED, "dead": TrackStatus.DELETED, } status = status_map.get(track_info["status"], TrackStatus.DELETED) return Track( id=hash(track_id) % (2**31), state=state_info["state"], covariance=state_info["covariance"], status=status, hits=track_info["hits"], misses=track_info["misses"], time=state_info["timestamp"], )
[docs] def tracks_to_tracklist( self, status: Optional[TrackDatabaseStatus] = None, ) -> Any: """Convert stored tracks to pytcl TrackList. Parameters ---------- status : TrackDatabaseStatus, optional Filter by status. Returns ------- TrackList pytcl TrackList container. """ from pytcl.containers.track_list import TrackList tracks_info = self.retrieve_all_tracks(status) track_objects = [] for t in tracks_info: try: track_objects.append(self.track_to_pytcl(t["track_id"])) except KeyError: continue return TrackList(track_objects)
[docs] def store_from_track(self, track: Any, timestamp: Optional[float] = None) -> 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. """ self._check_open() t = timestamp if timestamp is not None else track.time track_id = f"trk_{track.id}" # Check if track exists self._cursor.execute( "SELECT track_id FROM tracks WHERE track_id = ?", (track_id,) ) if self._cursor.fetchone() is None: self.initiate_track(track_id, track.state, track.covariance, t) else: self.update_track_state(track_id, track.state, track.covariance, t)
[docs] def store_from_tracklist(self, track_list: Any) -> None: """Store all tracks from a TrackList. Parameters ---------- track_list : TrackList pytcl TrackList container. """ for track in track_list: self.store_from_track(track)
# ========================================================================= # Internal Helpers # ========================================================================= def _check_open(self) -> None: """Verify the database is open.""" if self._cursor is None: raise RuntimeError("Database not open. Call open() first.") @staticmethod def _sanitize_metadata(metadata: Optional[Dict[str, Any]]) -> str: """Serialize metadata to JSON, converting numpy types to native Python.""" meta = metadata or {} cleaned: Dict[str, Any] = {} for k, v in meta.items(): if isinstance(v, (np.integer,)): cleaned[k] = int(v) elif isinstance(v, (np.floating,)): cleaned[k] = float(v) elif isinstance(v, np.ndarray): cleaned[k] = v.tolist() else: cleaned[k] = v return json.dumps(cleaned) def _set_track_status(self, track_id: str, status: TrackDatabaseStatus) -> None: """Update a track's status.""" self._check_open() self._cursor.execute( "UPDATE tracks SET status = ? WHERE track_id = ?", (status.value, track_id), ) self._connection.commit() def _store_state_row( self, track_id: str, state: NDArray[np.float64], covariance: NDArray[np.float64], timestamp: float, residual: Optional[ArrayLike] = None, update_type: str = "update", ) -> None: """Insert a single state row into track_states.""" res_bytes = None if residual is not None: res_bytes = np.asarray(residual, dtype=np.float64).tobytes() self._cursor.execute( """INSERT INTO track_states (track_id, timestamp, state, covariance, state_dim, residual, update_type) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( track_id, timestamp, state.tobytes(), covariance.tobytes(), len(state), res_bytes, update_type, ), ) @staticmethod def _serialize_array( arr: NDArray[np.float64], ) -> Tuple[bytes, int, str]: """Serialize ndarray to (bytes, first_dim, dtype_str).""" return arr.tobytes(), arr.shape[0], str(arr.dtype) @staticmethod def _deserialize_vector(data: bytes, dim: int) -> NDArray[np.float64]: """Deserialize bytes to 1D ndarray.""" return np.frombuffer(data, dtype=np.float64)[:dim].copy() @staticmethod def _deserialize_matrix(data: bytes, dim: int) -> NDArray[np.float64]: """Deserialize bytes to 2D ndarray (dim x dim).""" return ( np.frombuffer(data, dtype=np.float64)[: dim * dim].reshape(dim, dim).copy() ) def _row_to_detection(self, row: tuple[Any, ...]) -> Dict[str, Any]: """Convert a detections table row to a dict.""" meas = np.frombuffer(row[3], dtype=np.dtype(row[5]))[: row[4]].copy() cov = None if row[6] is not None: cov_dim = row[4] cov = ( np.frombuffer(row[6], dtype=np.float64)[: cov_dim * cov_dim] .reshape(cov_dim, cov_dim) .copy() ) return { "detection_id": row[0], "timestamp": row[1], "sensor_id": row[2], "measurement": meas, "covariance": cov, "association_status": row[7], "associated_track_id": row[8], "association_confidence": row[9], "metadata": json.loads(row[10]) if row[10] else {}, } @staticmethod def _row_to_track(row: tuple[Any, ...]) -> Dict[str, Any]: """Convert a tracks table row to a dict.""" return { "track_id": row[0], "status": row[1], "birth_time": row[2], "last_update_time": row[3], "state_dim": row[4], "hits": row[5], "misses": row[6], "total_misses": row[7], "confidence_score": row[8], "metadata": json.loads(row[9]) if row[9] else {}, } def _initialize_tables(self) -> None: """Create all tables and indices if they do not exist.""" if self._cursor is None: return self._cursor.execute("""CREATE TABLE IF NOT EXISTS detections ( detection_id TEXT PRIMARY KEY, timestamp REAL NOT NULL, sensor_id TEXT NOT NULL, measurement BLOB NOT NULL, measurement_dim INTEGER NOT NULL, measurement_dtype TEXT NOT NULL, covariance BLOB, association_status TEXT DEFAULT 'unassociated', associated_track_id TEXT, association_confidence REAL, metadata TEXT DEFAULT '{}' )""") self._cursor.execute("""CREATE TABLE IF NOT EXISTS tracks ( track_id TEXT PRIMARY KEY, status TEXT NOT NULL, birth_time REAL NOT NULL, last_update_time REAL NOT NULL, state_dim INTEGER NOT NULL, hits INTEGER DEFAULT 0, misses INTEGER DEFAULT 0, total_misses INTEGER DEFAULT 0, confidence_score REAL DEFAULT 0.0, metadata TEXT DEFAULT '{}' )""") self._cursor.execute("""CREATE TABLE IF NOT EXISTS track_states ( id INTEGER PRIMARY KEY AUTOINCREMENT, track_id TEXT NOT NULL, timestamp REAL NOT NULL, state BLOB NOT NULL, covariance BLOB NOT NULL, state_dim INTEGER NOT NULL, residual BLOB, update_type TEXT DEFAULT 'update' )""") self._cursor.execute("""CREATE TABLE IF NOT EXISTS track_associations ( id INTEGER PRIMARY KEY AUTOINCREMENT, track_id TEXT NOT NULL, detection_id TEXT NOT NULL, timestamp REAL NOT NULL, confidence REAL DEFAULT 1.0, UNIQUE(track_id, detection_id) )""") self._cursor.execute("""CREATE TABLE IF NOT EXISTS _pytcl_track_db_metadata ( key TEXT PRIMARY KEY, value TEXT NOT NULL )""") # Create indices self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_det_time ON detections(timestamp)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_det_sensor ON detections(sensor_id)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_det_status " "ON detections(association_status)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_ts_track_time " "ON track_states(track_id, timestamp)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_trk_status ON tracks(status)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_trk_update ON tracks(last_update_time)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_assoc_track ON track_associations(track_id)" ) self._cursor.execute( "CREATE INDEX IF NOT EXISTS idx_assoc_det " "ON track_associations(detection_id)" ) # Store schema version self._cursor.execute("""INSERT OR REPLACE INTO _pytcl_track_db_metadata (key, value) VALUES ('schema_version', '1.0')""") self._connection.commit()