use nalgebra::{ComplexField, RealField, SVector, Vector3};
use crate::frames::Body;
use crate::init;
use crate::math::wrap_pi;
use crate::observation::heading::heading_jacobian;
use crate::state::{Covariance, ErrorState, State};
use crate::units::{HeadingNoise, MagField, Radians};
use crate::update::Observation;
pub(crate) fn heading_innovation(
state: &State,
field: MagField<Body>,
declination: Radians,
) -> f32 {
let m_n = state.attitude.quaternion().to_rotation_matrix() * field.vector();
-wrap_pi(RealField::atan2(m_n.y, m_n.x) - declination.as_radians())
}
pub(crate) fn heading_observation(
state: &State,
covariance: &Covariance,
field: MagField<Body>,
declination: Radians,
noise: HeadingNoise,
) -> Observation<1> {
let h = heading_jacobian(state);
let down = h
.fixed_view::<1, 3>(0, ErrorState::AttitudeX.index())
.transpose();
let r_m = SVector::<f32, 1>::new(noise.variance() + leveling_variance(covariance, field, down));
Observation {
y: SVector::<f32, 1>::new(heading_innovation(state, field, declination)),
h,
h_b: SVector::<f32, 1>::zeros(),
r_m,
r_gain: r_m,
}
}
fn leveling_variance(covariance: &Covariance, field: MagField<Body>, down: Vector3<f32>) -> f32 {
let Some(sensitivity) = init::heading_sensitivity(field, down) else {
return 0.0;
};
let f = sensitivity.horizontal;
let g = down.cross(&f);
let theta = ErrorState::AttitudeX.index();
let p_theta = covariance.as_matrix().fixed_view::<3, 3>(theta, theta);
let (pf, pg) = (p_theta * f, p_theta * g);
let (a, b, d) = (f.dot(&pf), f.dot(&pg), g.dot(&pg));
let half = 0.5 * (a - d);
let tilt_variance = 0.5 * (a + d) + ComplexField::sqrt(half * half + b * b);
sensitivity.tan_dip * sensitivity.tan_dip * tilt_variance
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use crate::init::heading_from_mag;
use crate::state::{AttitudeVariance, STATES};
use crate::units::Attitude;
use core::f32::consts::PI;
use nalgebra::{UnitQuaternion, Vector3};
const DIP: f32 = 1.107;
fn field_ned(declination: f32) -> Vector3<f32> {
Vector3::new(
DIP.cos() * declination.cos(),
DIP.cos() * declination.sin(),
DIP.sin(),
)
}
pub(crate) fn attitude_of(roll: f32, pitch: f32, yaw: f32) -> Attitude {
Attitude::from_quaternion(UnitQuaternion::from_euler_angles(roll, pitch, yaw))
}
pub(crate) fn measured(attitude: Attitude, declination: f32) -> MagField<Body> {
let body = attitude.quaternion().inverse() * field_ned(declination);
MagField::body(body.x, body.y, body.z)
}
fn state_at(attitude: Attitude) -> State {
State {
attitude,
..State::default()
}
}
#[test]
fn a_field_synthesised_for_the_estimated_attitude_is_no_news() {
let attitude = attitude_of(0.3, -0.2, 1.1);
let innovation = heading_innovation(
&state_at(attitude),
measured(attitude, -0.06),
Radians::from_radians(-0.06),
);
assert!(innovation.abs() < 1e-5, "innovation {innovation}");
}
#[test]
fn a_yaw_error_at_non_zero_tilt_is_recovered() {
let truth = attitude_of(0.3, -0.2, 1.1);
let estimate = attitude_of(0.3, -0.2, 1.1 + 0.15);
let innovation = heading_innovation(
&state_at(estimate),
measured(truth, -0.06),
Radians::from_radians(-0.06),
);
assert!(
(innovation + 0.15).abs() < 1e-4,
"a yaw estimate 0.15 rad high should innovate -0.15, got {innovation}"
);
}
#[test]
fn declination_shifts_the_heading_by_exactly_d_m() {
let attitude = attitude_of(0.3, -0.2, 1.1);
let field = measured(attitude, -0.06);
let state = state_at(attitude);
let at_charted = heading_innovation(&state, field, Radians::from_radians(-0.06));
let at_shifted = heading_innovation(&state, field, Radians::from_radians(0.14));
assert!(
(at_shifted - at_charted - 0.2).abs() < 1e-5,
"{at_charted} and {at_shifted} should differ by the 0.2 rad of declination"
);
}
#[test]
fn a_vehicle_heading_near_the_half_turn_innovates_the_short_way() {
let truth = attitude_of(0.0, 0.0, -179.0_f32.to_radians());
let estimate = attitude_of(0.0, 0.0, 179.0_f32.to_radians());
let innovation = heading_innovation(
&state_at(estimate),
measured(truth, 0.0),
Radians::from_radians(0.0),
);
assert!(
(innovation - 2.0_f32.to_radians()).abs() < 1e-4,
"expected 2 degrees the short way, got {} degrees",
innovation.to_degrees()
);
assert!(innovation.abs() <= PI);
}
#[test]
fn a_heading_error_past_the_half_turn_innovates_the_short_way_round() {
let declination = Radians::from_radians(0.35);
let truth = attitude_of(0.0, 0.0, 0.0);
let estimate = attitude_of(0.0, 0.0, 2.93);
let innovation =
heading_innovation(&state_at(estimate), measured(truth, 0.35), declination);
assert!(
(innovation + 2.93).abs() < 1e-4,
"expected -2.93, got {innovation}"
);
assert!(innovation.abs() <= PI, "{innovation} is outside (-π, π]");
}
#[test]
fn the_innovation_is_the_leveled_heading_less_the_estimated_yaw() {
let truth = attitude_of(0.25, 0.4, -2.9);
let estimate = attitude_of(0.25, 0.4, -2.9 + 0.3);
let field = measured(truth, -0.06);
let declination = Radians::from_radians(-0.06);
let (roll, pitch, yaw) = estimate.euler_angles();
let leveled = heading_from_mag(
field,
Radians::from_radians(roll),
Radians::from_radians(pitch),
declination,
);
let innovation = heading_innovation(&state_at(estimate), field, declination);
assert!(
(innovation - wrap_pi(leveled.as_radians() - yaw)).abs() < 1e-5,
"(35) gave {innovation}, (6) less the estimated yaw gave {}",
wrap_pi(leveled.as_radians() - yaw)
);
}
#[test]
fn a_dead_magnetometer_reading_zero_reports_a_heading_rather_than_a_nan() {
let innovation = heading_innovation(
&state_at(Attitude::level()),
MagField::body(0.0, 0.0, 0.0),
Radians::from_radians(-0.06),
);
assert_eq!(innovation, -0.06);
}
#[test]
fn the_jacobian_reads_yaw_out_of_the_attitude_block_and_nothing_else() {
let h = heading_jacobian(&state_at(Attitude::level()));
let dx = SVector::<f32, STATES>::from_fn(|i, _| i as f32 + 1.0);
assert_eq!((h * dx)[0], 9.0);
for i in 0..STATES {
if i != ErrorState::AttitudeZ.index() {
assert_eq!(h[(0, i)], 0.0, "state {i} is not the yaw axis");
}
}
}
#[test]
fn the_jacobian_leans_with_the_tilt_and_not_with_the_yaw() {
let yawed = heading_jacobian(&state_at(attitude_of(0.0, 0.0, 1.1)));
assert_eq!(yawed[(0, ErrorState::AttitudeZ.index())], 1.0);
let pitched = heading_jacobian(&state_at(attitude_of(0.0, 0.4, 1.1)));
assert!(
(pitched[(0, ErrorState::AttitudeZ.index())] - 0.4_f32.cos()).abs() < 1e-6,
"the yaw column should be cos(pitch)"
);
assert!(
(pitched[(0, ErrorState::AttitudeX.index())] + 0.4_f32.sin()).abs() < 1e-6,
"and the roll column -sin(pitch)"
);
}
fn tilt_covariance(sigma: f32) -> Covariance {
let mut sigmas = [0.0; STATES];
sigmas[ErrorState::AttitudeX.index()] = sigma;
sigmas[ErrorState::AttitudeY.index()] = sigma;
Covariance::from_sigmas(sigmas)
}
#[test]
fn a_filter_certain_of_its_tilt_adds_nothing_to_the_callers_variance() {
let attitude = attitude_of(0.1, 0.1, 0.5);
let observation = heading_observation(
&state_at(attitude),
&tilt_covariance(0.0),
measured(attitude, -0.06),
Radians::from_radians(-0.06),
HeadingNoise::from_sigma(0.25),
);
assert_eq!(observation.r_m[0], 0.0625);
assert!(observation.y[0].abs() < 1e-5);
}
#[test]
fn the_leveling_variance_is_the_dip_squared_times_the_tilt_variance() {
let attitude = attitude_of(0.0, 0.0, 0.0);
let observation = heading_observation(
&state_at(attitude),
&tilt_covariance(0.02),
measured(attitude, 0.0),
Radians::from_radians(0.0),
HeadingNoise::from_sigma(0.05),
);
let expected = 0.05 * 0.05 + DIP.tan() * DIP.tan() * 0.02 * 0.02;
assert!(
(observation.r_m[0] - expected).abs() < 1e-7,
"R = {} rather than {expected}",
observation.r_m[0]
);
}
#[test]
fn a_horizontal_field_has_no_leveling_error_to_price() {
let observation = heading_observation(
&state_at(Attitude::level()),
&tilt_covariance(0.3),
MagField::body(0.22, 0.0, 0.0),
Radians::from_radians(0.0),
HeadingNoise::from_sigma(0.05),
);
let bare = HeadingNoise::from_sigma(0.05).variance();
assert_eq!(observation.r_m[0], bare);
}
#[test]
fn a_field_with_no_horizontal_part_is_not_priced_infinitely() {
let observation = heading_observation(
&state_at(Attitude::level()),
&tilt_covariance(0.3),
MagField::body(0.0, 0.0, 0.49),
Radians::from_radians(0.0),
HeadingNoise::from_sigma(0.05),
);
assert!(observation.r_m[0].is_finite());
assert_eq!(
observation.r_m[0],
HeadingNoise::from_sigma(0.05).variance()
);
}
#[test]
fn heading_variance_is_the_heading_rows_h_p_ht() {
let state = state_at(attitude_of(0.4, 1.2, -2.0));
let mut p = crate::state::CovarianceMatrix::from_fn(|i, j| 0.01 * (1 + i.min(j)) as f32);
p.fill_diagonal(0.2);
let covariance = Covariance::from_matrix(p);
let h = heading_jacobian(&state);
let hpht = (h * covariance.as_matrix() * h.transpose())[0];
let heading = AttitudeVariance::of(&state.attitude, &covariance).heading;
assert!((hpht - heading).abs() < 1e-6, "{hpht} against {heading}");
}
fn down_of(attitude: Attitude) -> Vector3<f32> {
attitude.quaternion().inverse() * Vector3::z()
}
fn anisotropic(attitude: &Attitude) -> Covariance {
let r = attitude.quaternion().to_rotation_matrix().into_inner();
let azimuth = nalgebra::Rotation3::from_axis_angle(&Vector3::z_axis(), 0.6).into_inner();
let ned = azimuth
* nalgebra::Matrix3::from_diagonal(&Vector3::new(0.02, 0.002, 0.3))
* azimuth.transpose();
let mut p = crate::state::CovarianceMatrix::identity();
let theta = ErrorState::AttitudeX.index();
p.fixed_view_mut::<3, 3>(theta, theta)
.copy_from(&(r.transpose() * ned * r));
Covariance::from_matrix(p)
}
#[test]
fn the_leveling_variance_prices_the_worst_horizontal_axis() {
let attitude = attitude_of(0.3, -0.4, 0.9);
let state = state_at(attitude);
let covariance = anisotropic(&attitude);
let observation = heading_observation(
&state,
&covariance,
measured(attitude, 1.7),
Radians::from_radians(1.7),
HeadingNoise::from_sigma(0.05),
);
let r = attitude.quaternion().to_rotation_matrix().into_inner();
let theta = ErrorState::AttitudeX.index();
let ned = r * covariance.as_matrix().fixed_view::<3, 3>(theta, theta) * r.transpose();
let largest = ned
.fixed_view::<2, 2>(0, 0)
.into_owned()
.symmetric_eigenvalues()
.max();
let expected = HeadingNoise::from_sigma(0.05).variance() + DIP.tan() * DIP.tan() * largest;
assert!(
(observation.r_m[0] - expected).abs() < 1e-5 * expected,
"R = {} rather than {expected}",
observation.r_m[0]
);
let diagonal = DIP.tan() * DIP.tan() * ned[(0, 0)].max(ned[(1, 1)]);
assert!(observation.r_m[0] - HeadingNoise::from_sigma(0.05).variance() > 1.1 * diagonal);
}
#[test]
fn the_leveling_variance_does_not_depend_on_where_the_vehicle_points() {
let leveling = |yaw: f32| {
let attitude = attitude_of(0.3, -0.4, yaw);
let covariance = anisotropic(&attitude_of(0.3, -0.4, 0.0));
let field = measured(attitude, yaw + 1.7);
leveling_variance(&covariance, field, down_of(attitude))
};
let at_zero = leveling(0.0);
for yaw in [0.5, 1.3, 2.9, -2.2, -0.7] {
let turned = leveling(yaw);
assert!(
(turned - at_zero).abs() < 1e-5 * at_zero,
"at yaw {yaw} the price is {turned} against {at_zero}"
);
}
}
#[test]
fn on_its_tail_the_leveling_reads_tilt_and_not_heading() {
let attitude = attitude_of(0.0, core::f32::consts::FRAC_PI_2, 0.3);
let field = measured(attitude, 0.0);
let down = down_of(attitude);
let leveling = |heading: f32| {
let block = AttitudeVariance {
tilt_north: 1e-3,
tilt_east: 1e-3,
heading,
}
.in_body(&attitude);
let mut p = crate::state::CovarianceMatrix::identity();
let theta = ErrorState::AttitudeX.index();
p.fixed_view_mut::<3, 3>(theta, theta).copy_from(&block);
leveling_variance(&Covariance::from_matrix(p), field, down)
};
let (tight, loose) = (leveling(1e-4), leveling(0.5));
assert!(tight > 0.0, "a dipping field has a leveling error to price");
assert!(
(tight - loose).abs() < 1e-6 * tight.max(1e-9) + 1e-9,
"heading's variance leaked into the leveling: {tight} against {loose}"
);
}
}