Skip to content
Closed
2 changes: 1 addition & 1 deletion book/RBPF.md
Original file line number Diff line number Diff line change
Expand Up @@ -460,7 +460,7 @@ fn test_rbpf_closed_loop_on_real_data() {
let stream = build_event_stream(&records, &cfg);

// Run closed-loop
let results = run_closed_loop(&mut rbpf, stream, None)
let results = run_closed_loop(&mut rbpf, stream, None, None)
.expect("RBPF closed-loop should complete");

// Compute error metrics
Expand Down
164 changes: 162 additions & 2 deletions core/src/sim.rs
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ use log::{debug, info, warn};
use std::fmt::{Debug, Display};
use std::io::{self, Read, Write};
use std::path::Path;
use std::time::{Duration as StdDuration, Instant};

use anyhow::{Result, bail};
use chrono::{DateTime, Duration, Utc};
Expand All @@ -60,9 +61,11 @@ use crate::messages::{Event, EventStream, GnssFaultModel, GnssScheduler};

use crate::{IMUData, StrapdownState, forward};
use health::HealthMonitor;
use execution::ExecutionMonitor;

// Re-export HealthLimits for easier access in tests and external users
pub use health::HealthLimits;
pub use execution::{ExecutionLimits, ExecutionMonitor};

pub const DEFAULT_PROCESS_NOISE: [f64; 15] = [
// Default process noise if not provided
Expand All @@ -83,6 +86,10 @@ pub const DEFAULT_PROCESS_NOISE: [f64; 15] = [
1e-8, // gyro bias z noise
];

pub const DEFAULT_MAX_WALL_CLOCK_RATIO: f64 = 0.25;
pub const DEFAULT_MAX_WALL_CLOCK_S: f64 = 1200.0;
pub const DEFAULT_MAX_NO_PROGRESS_S: f64 = 600.0;

fn de_f64_nan<'de, D>(deserializer: D) -> Result<f64, D::Error>
where
D: Deserializer<'de>,
Expand Down Expand Up @@ -1903,19 +1910,31 @@ pub fn dead_reckoning(records: &[TestDataRecord]) -> Vec<NavigationResult> {
/// * `filter` - Mutable reference to a type implementing NavigationFilter
/// * `stream` - Event stream containing IMU and measurement events
/// * `health_limits` - Optional health limits for monitoring
/// * `execution_limits` - Optional wall-clock and no-progress limits
///
/// # Returns
/// * `Vec<NavigationResult>` - A vector of navigation results
pub fn run_closed_loop<F: NavigationFilter>(
filter: &mut F,
stream: EventStream,
health_limits: Option<HealthLimits>,
execution_limits: Option<ExecutionLimits>,
) -> anyhow::Result<Vec<NavigationResult>> {
let start_time = stream.start_time;
let mut results: Vec<NavigationResult> = Vec::with_capacity(stream.events.len());
let total = stream.events.len();
let mut last_ts: Option<DateTime<Utc>> = None;
let mut monitor = HealthMonitor::new(health_limits.unwrap_or_default());
let sim_duration_s = stream
.events
.last()
.map(|event| match event {
Event::Imu { elapsed_s, .. } => *elapsed_s,
Event::Measurement { elapsed_s, .. } => *elapsed_s,
})
.unwrap_or(0.0);
let mut execution_monitor =
execution_limits.map(|limits| ExecutionMonitor::new(limits, sim_duration_s));

info!(
"Starting closed-loop navigation filter with {} events",
Expand Down Expand Up @@ -2005,6 +2024,10 @@ pub fn run_closed_loop<F: NavigationFilter>(
debug!("Filter state at {}: {:?}", ts, mean);
results.push(NavigationResult::from((&ts, &mean, &cov)));
}

if let Some(ref mut monitor) = execution_monitor {
monitor.check("closed-loop")?;
}
}
debug!("Closed-loop simulation complete");
Ok(results)
Expand Down Expand Up @@ -2525,6 +2548,125 @@ pub fn print_sim_status<F: NavigationFilter>(filter: &F) {
);
}

pub mod execution {
use super::*;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ExecutionLimits {
/// Max wall-clock time as a ratio of simulated duration (<= 0 disables).
#[serde(default = "default_max_wall_clock_ratio")]
pub max_wall_clock_ratio: f64,
/// Max wall-clock time per trajectory in seconds (<= 0 disables).
#[serde(default = "default_max_wall_clock_s")]
pub max_wall_clock_s: f64,
/// Max wall-clock time without progress in seconds (<= 0 disables).
#[serde(default = "default_max_no_progress_s")]
pub max_no_progress_s: f64,
}

fn default_max_wall_clock_ratio() -> f64 {
DEFAULT_MAX_WALL_CLOCK_RATIO
}

fn default_max_wall_clock_s() -> f64 {
DEFAULT_MAX_WALL_CLOCK_S
}

fn default_max_no_progress_s() -> f64 {
DEFAULT_MAX_NO_PROGRESS_S
}

impl Default for ExecutionLimits {
fn default() -> Self {
Self {
max_wall_clock_ratio: default_max_wall_clock_ratio(),
max_wall_clock_s: default_max_wall_clock_s(),
max_no_progress_s: default_max_no_progress_s(),
}
}
}

#[derive(Clone, Debug)]
pub struct ExecutionMonitor {
start_time: Instant,
last_progress: Instant,
max_wall_clock: Option<StdDuration>,
max_no_progress: Option<StdDuration>,
}

impl ExecutionMonitor {
pub fn new(limits: ExecutionLimits, sim_duration_s: f64) -> Self {
let max_wall_clock = compute_max_wall_clock(
sim_duration_s,
limits.max_wall_clock_ratio,
limits.max_wall_clock_s,
);
let max_no_progress = if limits.max_no_progress_s > 0.0 {
Some(StdDuration::from_secs_f64(limits.max_no_progress_s))
} else {
None
};
let now = Instant::now();

Self {
start_time: now,
last_progress: now,
max_wall_clock,
max_no_progress,
}
}

pub fn check(&mut self, context: &str) -> Result<()> {
let now = Instant::now();
if let Some(max_wall_clock) = self.max_wall_clock {
if now.duration_since(self.start_time) > max_wall_clock {
bail!(
"Execution timeout ({context}): exceeded wall-clock limit of {:.2} s",
max_wall_clock.as_secs_f64()
);
}
}
if let Some(max_no_progress) = self.max_no_progress {
let since_progress = now.duration_since(self.last_progress);
if since_progress > max_no_progress {
bail!(
"Execution timeout ({context}): no progress for {:.2} s (limit {:.2} s)",
since_progress.as_secs_f64(),
max_no_progress.as_secs_f64()
);
}
}
self.last_progress = now;

Copilot AI Jan 16, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The check method always updates last_progress on line 2639, which means the no-progress timeout can never trigger. The last_progress field should only be updated when actual progress is made (e.g., when an event is successfully processed), not on every check. Consider removing line 2639 and requiring callers to explicitly call a separate method like mark_progress() when progress occurs.

Copilot uses AI. Check for mistakes.
Ok(())
}
}

fn compute_max_wall_clock(
sim_duration_s: f64,
max_ratio: f64,
max_wall_clock_s: f64,
) -> Option<StdDuration> {
let mut max_s = None;
if sim_duration_s > 0.0 && max_ratio > 0.0 {
max_s = Some(sim_duration_s * max_ratio);
}
if max_wall_clock_s > 0.0 {
max_s = Some(match max_s {
Some(current) => current.min(max_wall_clock_s),
None => max_wall_clock_s,
});
}

max_s.and_then(|s| {
if s > 0.0 {
Some(StdDuration::from_secs_f64(s))
} else {
None
}
})
}
}

pub mod health {
use super::*;

Expand Down Expand Up @@ -3004,6 +3146,9 @@ pub struct SimulationConfig {
/// Generate performance plot comparing navigation output to GPS measurements
#[serde(default)]
pub generate_plot: bool,
/// Execution time limits (wall-clock and no-progress)
#[serde(default)]
pub execution_limits: ExecutionLimits,
/// Logging configuration
#[serde(default)]
pub logging: LoggingConfig,
Expand Down Expand Up @@ -3034,6 +3179,7 @@ impl Default for SimulationConfig {
seed: default_seed(),
parallel: false,
generate_plot: false,
execution_limits: ExecutionLimits::default(),
logging: LoggingConfig::default(),
closed_loop: Some(ClosedLoopConfig::default()),
particle_filter: None,
Expand Down Expand Up @@ -3704,7 +3850,7 @@ mod tests {
events: vec![event],
};

let res = run_closed_loop(&mut ukf, stream, None);
let res = run_closed_loop(&mut ukf, stream, None, None);
assert!(!res.unwrap().is_empty());
}
#[test]
Expand Down Expand Up @@ -4084,6 +4230,20 @@ mod tests {
assert!(limits.cov_diag_max > 0.0);
}

#[test]
fn test_execution_monitor_no_progress_timeout() {
let limits = ExecutionLimits {
max_wall_clock_ratio: 0.0,
max_wall_clock_s: 0.0,
max_no_progress_s: 0.01,
};
let mut monitor = ExecutionMonitor::new(limits, 1.0);

std::thread::sleep(std::time::Duration::from_millis(20));
let result = monitor.check("test");
assert!(result.is_err());
Comment on lines +4301 to +4303

Copilot AI Jan 16, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test will always pass regardless of the no-progress logic due to the bug in Comment 3. The test expects a timeout after 20ms of no progress, but since check() always updates last_progress, the timeout condition can never be satisfied. This test needs to be updated once the no-progress logic is fixed.

Suggested change
std::thread::sleep(std::time::Duration::from_millis(20));
let result = monitor.check("test");
assert!(result.is_err());
// NOTE: As of now, ExecutionMonitor::check() always updates `last_progress`,
// so the no-progress timeout condition cannot be triggered. This test currently
// verifies that no error is returned in that situation.
//
// TODO: Once the no-progress logic in ExecutionMonitor is fixed so that
// `last_progress` is only updated on real progress, update this test to
// expect a timeout error instead.
std::thread::sleep(std::time::Duration::from_millis(20));
let result = monitor.check("test");
assert!(result.is_ok());

Copilot uses AI. Check for mistakes.
}

#[test]
fn test_health_monitor_check_valid_state() {
let limits = HealthLimits::default();
Expand Down Expand Up @@ -4574,7 +4734,7 @@ mod tests {
};

let health_limits = HealthLimits::default();
let result = run_closed_loop(&mut ukf, stream, Some(health_limits));
let result = run_closed_loop(&mut ukf, stream, Some(health_limits), None);
assert!(result.is_ok());
}

Expand Down
26 changes: 13 additions & 13 deletions core/tests/integration_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -590,7 +590,7 @@ fn test_ukf_closed_loop_on_real_data() {

// Run closed-loop filter
let results =
run_closed_loop(&mut ukf, stream, None).expect("Closed-loop filter should complete");
run_closed_loop(&mut ukf, stream, None, None).expect("Closed-loop filter should complete");

// Verify results
assert!(
Expand Down Expand Up @@ -743,7 +743,7 @@ fn test_ukf_with_degraded_gnss() {
let stream = build_event_stream(&records, &cfg);

// Run closed-loop filter
let results = run_closed_loop(&mut ukf, stream, None)
let results = run_closed_loop(&mut ukf, stream, None, None)
.expect("Closed-loop filter with degraded GNSS should complete");

// Compute error metrics
Expand Down Expand Up @@ -838,7 +838,7 @@ fn test_ukf_outperforms_dead_reckoning() {
};
let stream = build_event_stream(&records, &cfg);

let ukf_results = run_closed_loop(&mut ukf, stream, None).expect("UKF should complete");
let ukf_results = run_closed_loop(&mut ukf, stream, None, None).expect("UKF should complete");
let ukf_stats = compute_error_metrics(&ukf_results, &records);

// Print comparison
Expand Down Expand Up @@ -918,7 +918,7 @@ fn test_ekf_closed_loop_on_real_data() {

// Run closed-loop filter
let results =
run_closed_loop(&mut ekf, stream, None).expect("Closed-loop EKF filter should complete");
run_closed_loop(&mut ekf, stream, None, None).expect("Closed-loop EKF filter should complete");

// Verify results
assert!(
Expand Down Expand Up @@ -1068,7 +1068,7 @@ fn test_ekf_with_degraded_gnss() {
let stream = build_event_stream(&records, &cfg);

// Run closed-loop filter
let results = run_closed_loop(&mut ekf, stream, None)
let results = run_closed_loop(&mut ekf, stream, None, None)
.expect("Closed-loop EKF filter with degraded GNSS should complete");

// Compute error metrics
Expand Down Expand Up @@ -1176,7 +1176,7 @@ fn test_ekf_outperforms_dead_reckoning() {
};
let stream = build_event_stream(&records, &cfg);

let ekf_results = run_closed_loop(&mut ekf, stream, None).expect("EKF should complete");
let ekf_results = run_closed_loop(&mut ekf, stream, None, None).expect("EKF should complete");
let ekf_stats = compute_error_metrics(&ekf_results, &records);

// Print comparison
Expand Down Expand Up @@ -1256,7 +1256,7 @@ fn test_eskf_closed_loop_on_real_data() {

// Run closed-loop filter
let results =
run_closed_loop(&mut eskf, stream, None).expect("Closed-loop ESKF filter should complete");
run_closed_loop(&mut eskf, stream, None, None).expect("Closed-loop ESKF filter should complete");

// Verify results
assert!(
Expand Down Expand Up @@ -1405,7 +1405,7 @@ fn test_eskf_with_degraded_gnss() {
let stream = build_event_stream(&records, &cfg);

// Run closed-loop filter
let results = run_closed_loop(&mut eskf, stream, None)
let results = run_closed_loop(&mut eskf, stream, None, None)
.expect("Closed-loop ESKF filter with degraded GNSS should complete");

// Compute error metrics
Expand Down Expand Up @@ -1505,7 +1505,7 @@ fn test_eskf_outperforms_dead_reckoning() {
};
let stream = build_event_stream(&records, &cfg);

let eskf_results = run_closed_loop(&mut eskf, stream, None).expect("ESKF should complete");
let eskf_results = run_closed_loop(&mut eskf, stream, None, None).expect("ESKF should complete");
let eskf_stats = compute_error_metrics(&eskf_results, &records);

// Print comparison
Expand Down Expand Up @@ -1596,7 +1596,7 @@ fn test_eskf_stability_high_dynamics() {

// Run closed-loop filter
let results =
run_closed_loop(&mut eskf, stream, None).expect("ESKF with high dynamics should complete");
run_closed_loop(&mut eskf, stream, None, None).expect("ESKF with high dynamics should complete");

// Verify all results are valid (no NaN or Inf)
for (i, result) in results.iter().enumerate() {
Expand Down Expand Up @@ -1709,7 +1709,7 @@ fn test_filter_comparison() {
0.0,
);
let stream_ukf = build_event_stream(&records, &cfg);
let ukf_results = run_closed_loop(&mut ukf, stream_ukf, None).expect("UKF should complete");
let ukf_results = run_closed_loop(&mut ukf, stream_ukf, None, None).expect("UKF should complete");
let ukf_stats = compute_error_metrics(&ukf_results, &records);

// Run EKF
Expand All @@ -1721,7 +1721,7 @@ fn test_filter_comparison() {
true,
);
let stream_ekf = build_event_stream(&records, &cfg);
let ekf_results = run_closed_loop(&mut ekf, stream_ekf, None).expect("EKF should complete");
let ekf_results = run_closed_loop(&mut ekf, stream_ekf, None, None).expect("EKF should complete");
let ekf_stats = compute_error_metrics(&ekf_results, &records);

// Run ESKF
Expand All @@ -1736,7 +1736,7 @@ fn test_filter_comparison() {
);

let stream_eskf = build_event_stream(&records, &cfg);
let eskf_results = run_closed_loop(&mut eskf, stream_eskf, None).expect("ESKF should complete");
let eskf_results = run_closed_loop(&mut eskf, stream_eskf, None, None).expect("ESKF should complete");
let eskf_stats = compute_error_metrics(&eskf_results, &records);

// Print comparison
Expand Down
Loading
Loading