Track Management & Data Persistence

This notebook provides a hands-on tutorial for managing track lifecycles and persisting tracking data using PyTCL’s SQL and HDF5 storage backends.

What You’ll Learn

1. Track Lifecycle Theory - Detection, initiation, confirmation, coasting, and deletion

2. SQL Backend (TrackDatabaseManager) - Real-time track storage, queries, and lifecycle operations

3. HDF5 Backend (TrackHDF5Storage) - Scenario archival, time-series retrieval, and compression

4. Workflow Integration - Real-time SQL pipeline feeding into archival HDF5 storage

5. Performance Analysis - Query latency and storage efficiency across backends

Key Concepts

• TrackDatabaseManager: SQLite-backed real-time track lifecycle management

• TrackHDF5Storage: HDF5-backed archival storage for large-scale tracking datasets

• Lifecycle states: TENTATIVE → CONFIRMED → COASTING → DEAD

• Detection association: Linking raw sensor measurements to maintained tracks

• Scenario archival: Exporting completed missions from SQL to compressed HDF5

Prerequisites

pip install nrl-tracker matplotlib numpy h5py

Estimated Time: 30-40 minutes

import os
import tempfile
import time

import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots

from pytcl.dynamic_estimation.kalman import kf_predict, kf_update
from pytcl.io import (
    TrackDatabaseManager,
    TrackDatabaseStatus,
    TrackHDF5Storage,
)

np.random.seed(42)

# Plotly dark theme template
dark_template = go.layout.Template()
dark_template.layout = go.Layout(
    paper_bgcolor='#0d1117',
    plot_bgcolor='#0d1117',
    font=dict(color='#e6edf3'),
    xaxis=dict(gridcolor='#30363d', zerolinecolor='#30363d'),
    yaxis=dict(gridcolor='#30363d', zerolinecolor='#30363d'),
)

# Temporary directory for database files
TMPDIR = tempfile.mkdtemp(prefix="pytcl_nb_")

print("PyTCL track management modules imported successfully")
print(f"Working directory: {TMPDIR}")

1. Track Lifecycle Theory

The Detection-to-Track Pipeline

In a tracking system, raw sensor measurements (detections) must be processed through a pipeline that manages the full lifecycle of each track:

Sensor Measurements
        │
        ▼
  ┌─────────────┐
  │  Detection   │  Store raw measurements with sensor ID, timestamp
  │  Storage     │
  └──────┬──────┘
         │
         ▼
  ┌─────────────┐
  │    Data      │  Associate detections to existing tracks
  │ Association  │  (GNN, JPDA, MHT)
  └──────┬──────┘
         │
    ┌────┴────┐
    │         │
    ▼         ▼
Matched    Unmatched
    │         │
    ▼         ▼
  ┌──────┐  ┌──────────┐
  │Update│  │ Initiate │  Create new tentative track
  │Track │  │ New Track│
  └──────┘  └──────────┘

Track Lifecycle States

Each track transitions through a state machine:

State	Meaning	Transition Rule
TENTATIVE	Newly created, awaiting confirmation	Created from unassociated detections
CONFIRMED	Reliably tracked target	After N consecutive hits (e.g., 3)
COASTING	Temporarily lost, predicting forward	After M consecutive misses without deletion
DEAD	Terminated, awaiting cleanup	After K total misses or manual deletion

Track Quality Metrics

• Hits: Number of successful measurement-to-track associations

• Misses: Consecutive scans without association (resets on hit)

• Total misses: Cumulative missed associations over track lifetime

• Confidence score: Derived metric combining hits, misses, and association quality

Database Design

PyTCL provides two complementary storage backends:

Feature	SQL (TrackDatabaseManager)	HDF5 (TrackHDF5Storage)
Use case	Real-time operations	Post-mission archival
Query speed	Fast random access by ID/time	Fast sequential/bulk reads
Schema	Relational (detections, tracks, states)	Hierarchical (groups, datasets)
Compression	None (small records)	gzip chunked (large arrays)
Concurrency	SQLite WAL mode	Single-writer
Best for	Live tracking loop	Analysis & replay

2. SQL Tutorial: TrackDatabaseManager

The TrackDatabaseManager provides real-time track lifecycle management backed by SQLite. It stores:

• Detections — raw sensor measurements with metadata

• Tracks — track state, status, and lifecycle counters

• Track states — full state/covariance history over time

• Associations — detection-to-track linkages with confidence

2.1 Creating the Database and Storing Detections

# Create a new tracking database
db_path = os.path.join(TMPDIR, "tracking_tutorial.db")
db = TrackDatabaseManager(db_path)
db.open(mode="w")

# Simulate a simple radar scenario: 3 targets, 30 time steps
dt = 1.0
n_steps = 30
n_targets = 3
R = np.eye(2) * 4.0  # Measurement noise covariance (2m std)

# True target trajectories: [x, vx, y, vy]
targets = [
    {"id": "tgt_A", "x0": np.array([0.0, 2.0, 0.0, 1.5])},
    {"id": "tgt_B", "x0": np.array([80.0, -1.0, 10.0, 2.0])},
    {"id": "tgt_C", "x0": np.array([40.0, 0.5, 60.0, -1.0])},
]

# Constant velocity transition matrix
F = np.array([
    [1, dt, 0, 0],
    [0, 1, 0, 0],
    [0, 0, 1, dt],
    [0, 0, 0, 1],
])

H = np.array([[1, 0, 0, 0], [0, 0, 1, 0]])  # Observe position only

# Generate and store detections
rng = np.random.default_rng(42)
true_states = {t["id"]: [] for t in targets}
detection_count = 0

for k in range(n_steps):
    timestamp = k * dt
    for tgt in targets:
        # Propagate true state
        state = tgt["x0"].copy()
        state[0] += state[1] * timestamp
        state[2] += state[3] * timestamp
        true_states[tgt["id"]].append(state.copy())

        # Generate detection (90% probability of detection)
        if rng.random() < 0.9:
            pos = H @ state
            meas = pos + rng.multivariate_normal([0, 0], R)

            det_id = f"det_{k:03d}_{tgt['id']}"
            db.store_detection(
                detection_id=det_id,
                measurement=meas,
                sensor_id="radar_01",
                timestamp=timestamp,
                covariance=R,
                metadata={"snr": float(rng.uniform(10, 30))},
            )
            detection_count += 1

# Convert to arrays for later use
for tid in true_states:
    true_states[tid] = np.array(true_states[tid])

print(f"Database created: {db_path}")
print(f"Stored {detection_count} detections over {n_steps} time steps")

# Query detections by time range and sensor
early_dets = db.retrieve_detections(start_time=0.0, end_time=5.0, sensor_id="radar_01")
print(f"Detections in [0, 5] seconds: {len(early_dets)}")

# Inspect a single detection
det = db.retrieve_detection("det_000_tgt_A")
print(f"\nDetection 'det_000_tgt_A':")
print(f"  Measurement: [{det['measurement'][0]:.2f}, {det['measurement'][1]:.2f}]")
print(f"  Timestamp:   {det['timestamp']}")
print(f"  Sensor:      {det['sensor_id']}")
print(f"  Status:      {det['association_status']}")
print(f"  Metadata:    SNR = {det['metadata'].get('snr', 'N/A'):.1f} dB")

2.2 Track Initiation and State Updates

Now we’ll run a Kalman filter over the detections and store each track’s state history in the database. This demonstrates the core real-time workflow:

1. Initiate tracks from early detections

2. Predict → Update with the Kalman filter

3. Store each state update in the database

4. Associate detections to tracks

# Process noise covariance
q = 0.1
Q = q * np.array([
    [dt**3/3, dt**2/2, 0, 0],
    [dt**2/2, dt, 0, 0],
    [0, 0, dt**3/3, dt**2/2],
    [0, 0, dt**2/2, dt],
])

# Initial covariance (large uncertainty)
P0 = np.diag([R[0, 0], 1.0, R[1, 1], 1.0])

# Track state: {track_id: (x, P)}
track_filters = {}

for k in range(n_steps):
    timestamp = k * dt

    # Get detections at this time step
    dets = db.retrieve_detections(start_time=timestamp, end_time=timestamp)

    # Predict existing tracks
    for tid, (x, P) in track_filters.items():
        pred = kf_predict(x, P, F, Q)
        track_filters[tid] = (pred.x, pred.P)

    # Simple nearest-neighbor association for each target
    for tgt in targets:
        tid = f"trk_{tgt['id']}"
        det_id = f"det_{k:03d}_{tgt['id']}"

        # Find the matching detection (if it exists)
        matching = [d for d in dets if d["detection_id"] == det_id]
        if not matching:
            # Missed detection: store prediction as coasting update
            if tid in track_filters:
                x, P = track_filters[tid]
                db.update_track_state(tid, x, P, timestamp, update_type="prediction")
            continue

        z = matching[0]["measurement"]

        if tid not in track_filters:
            # Initiate new track
            x0 = np.array([z[0], 0.0, z[1], 0.0])
            db.initiate_track(tid, x0, P0, timestamp)
            track_filters[tid] = (x0, P0.copy())

        # Kalman update
        x, P = track_filters[tid]
        upd = kf_update(x, P, z, H, R)
        track_filters[tid] = (upd.x, upd.P)

        # Store updated state and associate detection
        db.update_track_state(tid, upd.x, upd.P, timestamp, update_type="update")
        db.associate_detection(det_id, tid, confidence=0.95)

# Confirm all tracks (they've been running for 30 steps)
for tgt in targets:
    tid = f"trk_{tgt['id']}"
    db.confirm_track(tid)

print(f"Processed {n_steps} time steps")
print(f"Active tracks: {len(track_filters)}")
for tid in sorted(track_filters):
    info = db.get_track(tid)
    print(f"  {tid}: status={info['status']}, hits={info['hits']}, misses={info['misses']}")

2.3 Querying Track History and Visualization

The database stores the full state/covariance timeline for each track. Let’s retrieve and visualize it.

# Retrieve and plot track histories vs ground truth
colors = ['#00d4ff', '#ff4757', '#00ff88']
fig = make_subplots(
    rows=1, cols=2,
    subplot_titles=('2D Track Trajectories', 'Position Error Over Time'),
    horizontal_spacing=0.12,
)

for i, tgt in enumerate(targets):
    tid = f"trk_{tgt['id']}"
    history = db.get_track_history(tid)
    est_states = history["states"]
    est_times = history["timestamps"]
    truth = true_states[tgt["id"]]

    # Left: 2D trajectories
    fig.add_trace(
        go.Scatter(x=truth[:, 0], y=truth[:, 2], mode='lines',
                   name=f'True {tgt["id"]}', line=dict(color=colors[i], width=2),
                   opacity=0.4, legendgroup=tgt["id"]),
        row=1, col=1,
    )
    fig.add_trace(
        go.Scatter(x=est_states[:, 0], y=est_states[:, 2], mode='lines',
                   name=f'Est {tgt["id"]}',
                   line=dict(color=colors[i], width=2, dash='dash'),
                   legendgroup=tgt["id"]),
        row=1, col=1,
    )

    # Right: Position error
    min_len = min(len(est_states), len(truth))
    pos_err = np.sqrt(
        (est_states[:min_len, 0] - truth[:min_len, 0]) ** 2
        + (est_states[:min_len, 2] - truth[:min_len, 2]) ** 2
    )
    fig.add_trace(
        go.Scatter(x=est_times[:min_len], y=pos_err, mode='lines',
                   name=f'Error {tgt["id"]}', line=dict(color=colors[i], width=2),
                   legendgroup=tgt["id"], showlegend=False),
        row=1, col=2,
    )

fig.update_layout(
    template=dark_template, height=450,
    legend=dict(x=0.5, y=-0.15, xanchor='center', orientation='h'),
)
fig.update_xaxes(title_text='X (m)', row=1, col=1)
fig.update_yaxes(title_text='Y (m)', row=1, col=1)
fig.update_xaxes(title_text='Time (s)', row=1, col=2)
fig.update_yaxes(title_text='Position Error (m)', row=1, col=2)
fig.show()

2.4 Track Lifecycle Management

The database supports full lifecycle operations: marking tracks as coasting/dead, pruning old data, and merging duplicate tracks.

# Demonstrate lifecycle transitions
print("=== Track Lifecycle Demo ===\n")

# 1. Check current status
all_tracks = db.retrieve_all_tracks()
print("Current track statuses:")
for t in all_tracks:
    print(f"  {t['track_id']}: {t['status']} (hits={t['hits']}, misses={t['misses']})")

# 2. Mark one track as coasting (simulating lost contact)
db.mark_track_coasting("trk_tgt_C")
info = db.get_track("trk_tgt_C")
print(f"\nAfter marking trk_tgt_C as coasting: status={info['status']}")

# 3. Mark it dead (target left surveillance region)
db.mark_track_dead("trk_tgt_C")
info = db.get_track("trk_tgt_C")
print(f"After marking trk_tgt_C as dead: status={info['status']}")

# 4. Query by status
confirmed = db.retrieve_all_tracks(status=TrackDatabaseStatus.CONFIRMED)
dead = db.retrieve_all_tracks(status=TrackDatabaseStatus.DEAD)
print(f"\nConfirmed tracks: {len(confirmed)}")
print(f"Dead tracks: {len(dead)}")

# 5. Prune old unassociated detections (older than 10s from newest)
pruned = db.prune_old_detections(age_threshold=10.0)
print(f"\nPruned {pruned} old unassociated detections")

# 6. Restore trk_tgt_C for subsequent sections
db.mark_track_confirmed("trk_tgt_C")
print("\nRestored trk_tgt_C to confirmed for next sections")

2.5 Bulk Operations and pytcl Integration

The TrackDatabaseManager can convert between database records and pytcl’s native Track / TrackList containers, enabling seamless integration with existing filter pipelines.

# Convert database tracks to pytcl Track objects
track_obj = db.track_to_pytcl("trk_tgt_A")
print(f"pytcl Track object for trk_tgt_A:")
print(f"  id:     {track_obj.id}")
print(f"  state:  [{', '.join(f'{v:.2f}' for v in track_obj.state)}]")
print(f"  status: {track_obj.status}")
print(f"  hits:   {track_obj.hits}")
print(f"  time:   {track_obj.time}")

# Convert all confirmed tracks to a TrackList
track_list = db.tracks_to_tracklist(status=TrackDatabaseStatus.CONFIRMED)
print(f"\nTrackList with {len(track_list)} confirmed tracks")

# Store a batch of state history at once
batch_states = np.random.randn(5, 4)  # 5 timesteps, 4-dim state
batch_covs = np.array([np.eye(4) for _ in range(5)])
batch_times = np.arange(100.0, 105.0)

db.initiate_track("trk_batch_demo", batch_states[0], batch_covs[0], 100.0)
db.store_track_history("trk_batch_demo", batch_states, batch_covs, batch_times)
history = db.get_track_history("trk_batch_demo")
print(f"\nBatch-stored track 'trk_batch_demo': {len(history['timestamps'])} state entries")

3. HDF5 Tutorial: TrackHDF5Storage

The TrackHDF5Storage backend is optimized for archiving completed tracking scenarios. It uses:

• Chunked datasets for efficient sequential I/O

• gzip compression for reduced file size (typically 5-10x)

• Hierarchical groups for organizing tracks and scenarios

• Resizable datasets for appending new states incrementally

3.1 Storing Tracks in HDF5

# Create HDF5 storage and archive track data from the SQL database
h5_path = os.path.join(TMPDIR, "scenario_archive.h5")
store = TrackHDF5Storage(h5_path, compression="gzip", compression_level=4)
store.open(mode="w")

# Store each track's full trajectory from the SQL database
for tgt in targets:
    tid = f"trk_{tgt['id']}"
    history = db.get_track_history(tid)
    track_info = db.get_track(tid)

    store.store_track(
        track_id=tid,
        states=history["states"],
        covariances=history["covariances"],
        timestamps=history["timestamps"],
        metadata={
            "status": track_info["status"],
            "hits": track_info["hits"],
            "birth_time": track_info["birth_time"],
        },
    )

# List stored tracks
stored_tracks = store.list_tracks()
print(f"HDF5 file: {h5_path}")
print(f"Stored tracks: {stored_tracks}")

# Retrieve and verify a track
retrieved = store.retrieve_track("trk_tgt_A")
print(f"\nRetrieved trk_tgt_A:")
print(f"  States shape:      {retrieved['states'].shape}")
print(f"  Covariances shape: {retrieved['covariances'].shape}")
print(f"  Time range:        [{retrieved['timestamps'][0]:.1f}, {retrieved['timestamps'][-1]:.1f}] s")
print(f"  Metadata:          {retrieved['metadata']}")

3.2 Time-Series Queries

HDF5 supports efficient time-sliced retrieval and interpolation — useful for post-mission analysis.

# Time-sliced trajectory extraction
traj = store.get_track_trajectory("trk_tgt_B", start_time=5.0, end_time=20.0)
print(f"Trajectory slice [5s, 20s]:")
print(f"  States shape: {traj['states'].shape}")
print(f"  Time range:   [{traj['timestamps'][0]:.1f}, {traj['timestamps'][-1]:.1f}] s")

# State interpolation at arbitrary time
state_nearest = store.get_state_at_time("trk_tgt_B", time=7.5, interpolate=False)
state_interp = store.get_state_at_time("trk_tgt_B", time=7.5, interpolate=True)

print(f"\nState at t=7.5s (nearest):      [{', '.join(f'{v:.2f}' for v in state_nearest['state'])}]")
print(f"State at t=7.5s (interpolated): [{', '.join(f'{v:.2f}' for v in state_interp['state'])}]")
print(f"Nearest timestamp: {state_nearest['timestamp']:.1f}s")
print(f"Interpolated timestamp: {state_interp['timestamp']:.1f}s")

# Spatial query: find tracks within a bounding box
tracks_in_region = store.get_tracks_in_region(
    bbox=[20, 20, 80, 80],  # [x_min, y_min, x_max, y_max]
    time_range=[10.0, 25.0],
    state_indices=(0, 2),  # x is index 0, y is index 2
)
print(f"\nTracks passing through [20,20]-[80,80] between t=10-25s: {tracks_in_region}")

3.3 Scenario Archival and Comparison

HDF5 supports storing complete tracking scenarios (tracks + detections) under named groups, enabling side-by-side comparison of different algorithm configurations.

# Store two scenarios with different noise levels
for scenario_name, noise_mult in [("low_noise", 0.5), ("high_noise", 2.0)]:
    tracks_data = {}
    rng_sc = np.random.default_rng(123)

    for tgt in targets:
        tid = tgt["id"]
        n = n_steps
        states_sc = np.zeros((n, 4))
        covs_sc = np.array([np.eye(4) * noise_mult for _ in range(n)])
        times_sc = np.arange(n, dtype=np.float64)

        for k in range(n):
            state = tgt["x0"].copy()
            state[0] += state[1] * k * dt
            state[2] += state[3] * k * dt
            # Add noise proportional to scenario
            state[:2] += rng_sc.normal(0, noise_mult, 2)
            state[2:] += rng_sc.normal(0, noise_mult, 2)
            states_sc[k] = state

        tracks_data[tid] = {
            "states": states_sc,
            "covariances": covs_sc,
            "timestamps": times_sc,
        }

    store.store_tracking_scenario(
        scenario_id=scenario_name,
        tracks=tracks_data,
        metadata={"noise_multiplier": noise_mult, "n_targets": n_targets},
    )

# List scenarios and compare
scenarios = store.list_scenarios()
print(f"Stored scenarios: {scenarios}")

comparison = store.compare_scenarios("low_noise", "high_noise")
print(f"\nScenario comparison (low_noise vs high_noise):")
print(f"  Common tracks:  {comparison['common_tracks']}")
print(f"  State RMSE per track:")
for tid, rmse in comparison["state_differences"].items():
    print(f"    {tid}: {rmse:.3f}")

# Visualize scenario comparison
fig = make_subplots(
    rows=1, cols=2,
    subplot_titles=('Low Noise Scenario', 'High Noise Scenario'),
    horizontal_spacing=0.1,
)

for col, scenario_id in enumerate(["low_noise", "high_noise"], 1):
    scenario = store.retrieve_tracking_scenario(scenario_id)

    for i, (tid, tdata) in enumerate(scenario["tracks"].items()):
        fig.add_trace(
            go.Scatter(
                x=tdata["states"][:, 0], y=tdata["states"][:, 2],
                mode='lines', name=tid if col == 1 else None,
                line=dict(color=colors[i], width=2),
                legendgroup=tid, showlegend=(col == 1),
            ),
            row=1, col=col,
        )

fig.update_layout(
    template=dark_template, height=400,
    legend=dict(x=0.5, y=-0.15, xanchor='center', orientation='h'),
)
fig.update_xaxes(title_text='X (m)', row=1, col=1)
fig.update_xaxes(title_text='X (m)', row=1, col=2)
fig.update_yaxes(title_text='Y (m)', row=1, col=1)
fig.update_yaxes(title_text='Y (m)', row=1, col=2)
fig.show()

3.4 Compression Analysis

HDF5 compression reduces storage requirements significantly for tracking data, which contains highly correlated floating-point arrays.

# Compare file sizes across compression levels
compression_results = []
n_tracks_large = 20
n_steps_large = 100
state_dim = 6  # [x, vx, y, vy, z, vz]

# Generate a larger dataset
rng_comp = np.random.default_rng(99)
large_tracks = {}
for i in range(n_tracks_large):
    states_l = np.cumsum(rng_comp.normal(0, 1, (n_steps_large, state_dim)), axis=0)
    covs_l = np.array([np.eye(state_dim) * (1 + 0.01 * k) for k in range(n_steps_large)])
    times_l = np.arange(n_steps_large, dtype=np.float64)
    large_tracks[f"trk_{i:03d}"] = {
        "states": states_l,
        "covariances": covs_l,
        "timestamps": times_l,
    }

# Raw data size
raw_bytes = n_tracks_large * n_steps_large * (
    state_dim * 8 + state_dim * state_dim * 8 + 8  # states + covs + timestamps
)

for level in [0, 1, 4, 9]:
    path = os.path.join(TMPDIR, f"compress_test_{level}.h5")
    s = TrackHDF5Storage(path, compression="gzip", compression_level=max(level, 1))
    s.open(mode="w")
    s.store_tracking_scenario("test", large_tracks)
    s.close()
    file_size = os.path.getsize(path)
    ratio = raw_bytes / file_size if file_size > 0 else 0
    compression_results.append({
        "level": level,
        "file_kb": file_size / 1024,
        "ratio": ratio,
    })

# Display results
print(f"Raw data size: {raw_bytes / 1024:.1f} KB ({n_tracks_large} tracks x {n_steps_large} steps x {state_dim}D state)")
print(f"\n{'Level':>6} {'File Size':>12} {'Compression Ratio':>18}")
print("-" * 40)
for r in compression_results:
    print(f"{r['level']:>6} {r['file_kb']:>10.1f} KB {r['ratio']:>15.1f}x")

# Visualize
fig = go.Figure()
fig.add_trace(go.Bar(
    x=[str(r["level"]) for r in compression_results],
    y=[r["file_kb"] for r in compression_results],
    text=[f"{r['ratio']:.1f}x" for r in compression_results],
    textposition='outside',
    marker_color=['#ff4757', '#ffb800', '#00d4ff', '#00ff88'],
))
fig.add_hline(y=raw_bytes / 1024, line_dash="dash", line_color="#e6edf3",
              annotation_text=f"Raw: {raw_bytes/1024:.0f} KB", annotation_position="top left")
fig.update_layout(
    template=dark_template, height=400,
    title='HDF5 Compression: File Size by gzip Level',
    xaxis_title='gzip Compression Level',
    yaxis_title='File Size (KB)',
)
fig.show()

4. Workflow Demo: Real-Time SQL to Archival HDF5

A common operational pattern is:

1. Real-time phase: Use SQL for fast detection storage, track updates, and lifecycle management during a live mission

2. Archive phase: Export completed mission data from SQL to compressed HDF5 for long-term storage and offline analysis

This demonstrates the import_from_sql / export_to_sql interoperability.

# === PHASE 1: Real-time tracking with SQL ===
rt_db_path = os.path.join(TMPDIR, "realtime_mission.db")
rt_db = TrackDatabaseManager(rt_db_path)
rt_db.open(mode="w")

# Simulate a 50-step mission with 5 targets
rng_rt = np.random.default_rng(77)
n_rt_targets = 5
n_rt_steps = 50

# Generate diverse target initial conditions
rt_targets = []
for i in range(n_rt_targets):
    angle = 2 * np.pi * i / n_rt_targets
    x0 = np.array([
        50 * np.cos(angle), 1.5 * np.cos(angle + 0.3),
        50 * np.sin(angle), 1.5 * np.sin(angle + 0.3),
    ])
    rt_targets.append(x0)

# Run real-time tracking loop
rt_filters = {}
for k in range(n_rt_steps):
    t = k * dt
    for i, x0 in enumerate(rt_targets):
        tid = f"rt_trk_{i:02d}"
        # True position (constant velocity)
        true_pos = np.array([x0[0] + x0[1] * t, x0[2] + x0[3] * t])

        # Detection (95% PD)
        if rng_rt.random() > 0.95:
            if tid in rt_filters:
                x, P = rt_filters[tid]
                pred = kf_predict(x, P, F, Q)
                rt_filters[tid] = (pred.x, pred.P)
                rt_db.update_track_state(tid, pred.x, pred.P, t, update_type="prediction")
            continue

        z = true_pos + rng_rt.multivariate_normal([0, 0], R)
        det_id = f"rt_det_{k:03d}_{i:02d}"
        rt_db.store_detection(det_id, z, f"sensor_{i % 2}", t)

        if tid not in rt_filters:
            x_init = np.array([z[0], 0.0, z[1], 0.0])
            rt_db.initiate_track(tid, x_init, P0, t)
            rt_filters[tid] = (x_init, P0.copy())
            if k > 3:
                rt_db.confirm_track(tid)

        x, P = rt_filters[tid]
        pred = kf_predict(x, P, F, Q)
        upd = kf_update(pred.x, pred.P, z, H, R)
        rt_filters[tid] = (upd.x, upd.P)
        rt_db.update_track_state(tid, upd.x, upd.P, t, update_type="update")
        rt_db.associate_detection(det_id, tid)

# Confirm all tracks
for i in range(n_rt_targets):
    tid = f"rt_trk_{i:02d}"
    rt_db.confirm_track(tid)

rt_tracks = rt_db.retrieve_all_tracks()
print(f"=== Real-Time Phase Complete ===")
print(f"SQL database: {rt_db_path}")
print(f"Tracks: {len(rt_tracks)}")
print(f"Database size: {os.path.getsize(rt_db_path) / 1024:.1f} KB")

# === PHASE 2: Archive to HDF5 ===
archive_path = os.path.join(TMPDIR, "mission_archive.h5")
archive = TrackHDF5Storage(archive_path, compression="gzip", compression_level=4)
archive.open(mode="w")

# Import from SQL into HDF5 as a named scenario
t_start = time.perf_counter()
archive.import_from_sql(rt_db, scenario_id="mission_001")
t_archive = time.perf_counter() - t_start

archive.close()
rt_db.close()

archive_size = os.path.getsize(archive_path)
sql_size = os.path.getsize(rt_db_path)

print(f"=== Archive Phase Complete ===")
print(f"HDF5 archive: {archive_path}")
print(f"Archive time: {t_archive * 1000:.1f} ms")
print(f"\nStorage comparison:")
print(f"  SQL database: {sql_size / 1024:.1f} KB")
print(f"  HDF5 archive: {archive_size / 1024:.1f} KB")
print(f"  Ratio: {sql_size / archive_size:.1f}x")

# Verify round-trip: read back from HDF5
archive.open(mode="r")
scenario = archive.retrieve_tracking_scenario("mission_001")
print(f"\nRound-trip verification:")
print(f"  Tracks recovered: {len(scenario['tracks'])}")
print(f"  Detections recovered: {len(scenario['detections'])}")

# Spot-check a track
tid_check = "rt_trk_00"
if tid_check in scenario["tracks"]:
    t_data = scenario["tracks"][tid_check]
    print(f"  {tid_check} states shape: {t_data['states'].shape}")

archive.close()

5. Interactive Exploration: Query Performance

Understanding the performance characteristics of each backend helps you choose the right tool for your use case. Let’s measure query latency across different operations.

# Benchmark SQL operations
perf_db_path = os.path.join(TMPDIR, "perf_bench.db")
perf_db = TrackDatabaseManager(perf_db_path)
perf_db.open(mode="w")

n_bench_tracks = 50
n_bench_steps = 100
rng_bench = np.random.default_rng(55)

# Populate benchmark database
for i in range(n_bench_tracks):
    tid = f"bench_trk_{i:03d}"
    x0 = rng_bench.normal(0, 10, 4)
    perf_db.initiate_track(tid, x0, np.eye(4), 0.0)
    for k in range(1, n_bench_steps):
        x_k = x0 + rng_bench.normal(0, 0.1, 4) * k
        perf_db.update_track_state(tid, x_k, np.eye(4), float(k))

    det_id = f"bench_det_{i:03d}"
    perf_db.store_detection(det_id, rng_bench.normal(0, 5, 2), "radar", float(i))

# Benchmark: single track state retrieval
n_trials = 50
sql_timings = {}

t0 = time.perf_counter()
for _ in range(n_trials):
    perf_db.get_track_state(f"bench_trk_{rng_bench.integers(0, n_bench_tracks):03d}")
sql_timings["get_track_state"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: track history retrieval
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_db.get_track_history(f"bench_trk_{rng_bench.integers(0, n_bench_tracks):03d}")
sql_timings["get_track_history"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: detection query by time range
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_db.retrieve_detections(start_time=0.0, end_time=25.0)
sql_timings["query_detections"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: list all tracks
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_db.retrieve_all_tracks()
sql_timings["list_all_tracks"] = (time.perf_counter() - t0) / n_trials * 1000

perf_db.close()

# Benchmark HDF5 operations
perf_h5_path = os.path.join(TMPDIR, "perf_bench.h5")
perf_store = TrackHDF5Storage(perf_h5_path)
perf_store.open(mode="w")

# Populate HDF5
h5_tracks = {}
for i in range(n_bench_tracks):
    tid = f"bench_trk_{i:03d}"
    states_b = np.cumsum(rng_bench.normal(0, 1, (n_bench_steps, 4)), axis=0)
    covs_b = np.array([np.eye(4) for _ in range(n_bench_steps)])
    times_b = np.arange(n_bench_steps, dtype=np.float64)
    perf_store.store_track(tid, states_b, covs_b, times_b)

h5_timings = {}

# Benchmark: full track retrieval
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_store.retrieve_track(f"bench_trk_{rng_bench.integers(0, n_bench_tracks):03d}")
h5_timings["retrieve_track"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: trajectory slice
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_store.get_track_trajectory(
        f"bench_trk_{rng_bench.integers(0, n_bench_tracks):03d}",
        start_time=20.0, end_time=60.0,
    )
h5_timings["trajectory_slice"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: state interpolation
t0 = time.perf_counter()
for _ in range(n_trials):
    perf_store.get_state_at_time(
        f"bench_trk_{rng_bench.integers(0, n_bench_tracks):03d}",
        time=45.5, interpolate=True,
    )
h5_timings["interpolate_state"] = (time.perf_counter() - t0) / n_trials * 1000

# Benchmark: spatial query
t0 = time.perf_counter()
for _ in range(min(n_trials, 10)):
    perf_store.get_tracks_in_region([-10, -10, 10, 10])
h5_timings["spatial_query"] = (time.perf_counter() - t0) / min(n_trials, 10) * 1000

perf_store.close()

# Display results
print(f"=== Query Performance ({n_bench_tracks} tracks, {n_bench_steps} steps each) ===\n")
print(f"{'SQL Operation':<25} {'Latency':>10}")
print("-" * 37)
for op, ms in sql_timings.items():
    print(f"  {op:<23} {ms:>8.2f} ms")

print(f"\n{'HDF5 Operation':<25} {'Latency':>10}")
print("-" * 37)
for op, ms in h5_timings.items():
    print(f"  {op:<23} {ms:>8.2f} ms")

# Visualize performance comparison
all_ops = list(sql_timings.keys()) + list(h5_timings.keys())
all_vals = list(sql_timings.values()) + list(h5_timings.values())
all_backends = ["SQL"] * len(sql_timings) + ["HDF5"] * len(h5_timings)
bar_colors = ["#00d4ff"] * len(sql_timings) + ["#00ff88"] * len(h5_timings)

fig = go.Figure()
fig.add_trace(go.Bar(
    x=list(sql_timings.keys()),
    y=list(sql_timings.values()),
    name="SQL",
    marker_color="#00d4ff",
    text=[f"{v:.2f}" for v in sql_timings.values()],
    textposition="outside",
))
fig.add_trace(go.Bar(
    x=list(h5_timings.keys()),
    y=list(h5_timings.values()),
    name="HDF5",
    marker_color="#00ff88",
    text=[f"{v:.2f}" for v in h5_timings.values()],
    textposition="outside",
))
fig.update_layout(
    template=dark_template, height=450,
    title="Query Latency by Backend and Operation",
    xaxis_title="Operation",
    yaxis_title="Latency (ms)",
    barmode="group",
)
fig.show()

Exercises

Exercise 1: Track Confirmation Logic

Objective: Implement an M-of-N confirmation rule

Task:

1. Create a new TrackDatabaseManager database

2. Initiate 5 tentative tracks

3. Simulate detections where each track receives hits with 80% probability per scan

4. Implement a 3-of-5 confirmation rule: confirm a track if it gets at least 3 hits in the first 5 scans

5. Print which tracks were confirmed and which were deleted

Hint: Use db.get_track(tid) to check hits and misses counts, then call db.confirm_track() or db.mark_track_dead().

---

Exercise 2: HDF5 Scenario Replay

Objective: Load an archived scenario and replay it through a different filter

Task:

1. Open the scenario_archive.h5 file created earlier (read-only)

2. Retrieve the low_noise scenario

3. For each track, treat the stored states as “measurements” and run a Kalman smoother (or simply a KF with different Q)

4. Store the re-filtered results as a new scenario called refiltered

5. Use compare_scenarios to quantify the difference between original and refiltered tracks

Hint: Use store.retrieve_tracking_scenario("low_noise") then store.store_tracking_scenario("refiltered", ...).

---

Exercise 3: Multi-Sensor Detection Fusion

Objective: Manage detections from multiple sensors with different noise characteristics

Task:

1. Create a database and store detections from two sensors:

   • radar_01: 5m noise std, 1Hz update rate, 85% PD

   • lidar_01: 1m noise std, 10Hz update rate, 95% PD

2. Initiate tracks and run a Kalman filter that uses detections from both sensors

3. Store separate covariance matrices for each sensor’s detections

4. Query detections by sensor_id to analyze each sensor’s contribution

5. Visualize the track uncertainty over time, highlighting which sensor provided each update

Hint: Use db.store_detection(..., sensor_id="radar_01") and query with db.retrieve_detections(sensor_id="lidar_01").

---

Exercise 4: Track Merge Detection

Objective: Detect and merge duplicate tracks from the same target

Task:

1. Create a scenario where two tracks are initiated on the same target (simulating a brief detection gap)

2. Store both tracks’ state histories in the database

3. Implement a merge criterion: if two tracks’ latest states are within 5m Mahalanobis distance, merge them

4. Use db.merge_tracks(keep_id, merge_id) to combine the histories

5. Verify that the merged track has the combined hit count and the merged track is marked DEAD

Hint: Compare db.get_track_state(tid1) and db.get_track_state(tid2) states using Mahalanobis distance.

Key Takeaways

• TrackDatabaseManager (SQL) is ideal for real-time tracking loops: fast random access, lifecycle management, detection-track association

• TrackHDF5Storage (HDF5) excels at post-mission archival: compressed storage, time-series queries, scenario comparison

• Lifecycle management (TENTATIVE → CONFIRMED → COASTING → DEAD) provides structured track quality control

• SQL → HDF5 pipeline enables a natural workflow: live tracking followed by compressed archival

• Both backends interoperate via import_from_sql / export_to_sql for seamless data exchange

Next Steps

• Explore Notebook 03 (Multi-Target Tracking) for data association algorithms (GNN, JPDA)

• See examples/track_management_workflows.py for complete end-to-end pipeline examples

• Review pytcl.io API documentation for the full method reference

References

Core Textbooks

1. Bar-Shalom, Y., Li, X. R., & Kirubarajan, T. (2001). Estimation with Applications to Tracking and Navigation. Wiley. — Chapters 6-8 on track management and data association

2. Blackman, S. S., & Popoli, R. (1999). Design and Analysis of Modern Tracking Systems. Artech House. — Comprehensive track lifecycle management

Data Storage

3. The HDF Group. HDF5 User’s Guide. — Chunking, compression, and hierarchical data organization

4. SQLite Documentation. Write-Ahead Logging. — WAL mode for concurrent read/write access

PyTCL API

• pytcl.io.TrackDatabaseManager — SQL track lifecycle management

• pytcl.io.TrackHDF5Storage — HDF5 archival storage

• pytcl.io.TrackDatabaseStatus — Lifecycle state enumeration

• examples/track_management_workflows.py — End-to-end workflow examples

# Cleanup: close open connections and remove temp files
db.close()
store.close()

import shutil
shutil.rmtree(TMPDIR, ignore_errors=True)
print("Temporary files cleaned up.")