use crate::control::dense_ops::{dense_mul, dense_mul_adjoint_lhs};
use crate::decomp::{DecompError, DenseDecompParams, dense_self_adjoint_eigen, dense_svd};
use crate::sparse::compensated::CompensatedField;
use alloc::vec::Vec;
use core::fmt;
use faer::{Col, Mat, MatRef};
use faer_traits::ext::ComplexFieldExt;
use num_traits::{Float, One, Zero};
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum HsvdInternalsLevel {
#[default]
Summary,
Factors,
Full,
}
impl HsvdInternalsLevel {
#[inline]
fn keep_factors(self) -> bool {
!matches!(self, Self::Summary)
}
#[inline]
fn keep_full(self) -> bool {
matches!(self, Self::Full)
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct HsvdParams<R> {
pub order: Option<usize>,
pub sigma_tol: Option<R>,
pub internals: HsvdInternalsLevel,
}
impl<R> Default for HsvdParams<R> {
fn default() -> Self {
Self {
order: None,
sigma_tol: None,
internals: HsvdInternalsLevel::Summary,
}
}
}
impl<R> HsvdParams<R> {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn with_order(mut self, order: usize) -> Self {
self.order = Some(order);
self
}
#[must_use]
pub fn with_sigma_tol(mut self, sigma_tol: R) -> Self {
self.sigma_tol = Some(sigma_tol);
self
}
#[must_use]
pub fn with_internals(mut self, internals: HsvdInternalsLevel) -> Self {
self.internals = internals;
self
}
}
#[derive(Clone, Debug)]
pub struct HsvdInternals<T: CompensatedField>
where
T::Real: Float,
{
pub controllability_factor: Mat<T>,
pub observability_factor: Mat<T>,
pub core: Mat<T>,
pub core_u: Mat<T>,
pub core_singular_values: Col<T::Real>,
pub core_vh: Mat<T>,
pub dense_controllability_gramian: Option<Mat<T>>,
pub dense_observability_gramian: Option<Mat<T>>,
pub dense_controllability_sqrt: Option<Mat<T>>,
pub dense_observability_sqrt: Option<Mat<T>>,
}
#[derive(Clone, Debug)]
pub struct HsvdResult<T: CompensatedField>
where
T::Real: Float,
{
pub hankel_singular_values: Col<T::Real>,
pub reduced_order: usize,
pub error_bound: Option<T::Real>,
pub left_projection: Mat<T>,
pub right_projection: Mat<T>,
pub internals: Option<HsvdInternals<T>>,
}
#[derive(Debug)]
pub enum HsvdError<R> {
Decomposition(DecompError),
DimensionMismatch {
controllability_nrows: usize,
observability_nrows: usize,
},
InvalidOrder {
requested: usize,
available: usize,
},
EmptyRetainedSpectrum,
IndefiniteGramian {
which: &'static str,
eigenvalue: R,
},
NonFiniteResult {
which: &'static str,
},
}
impl<R: fmt::Debug> fmt::Display for HsvdError<R> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt::Debug::fmt(self, f)
}
}
impl<R: fmt::Debug> core::error::Error for HsvdError<R> {}
impl<R> From<DecompError> for HsvdError<R> {
fn from(value: DecompError) -> Self {
Self::Decomposition(value)
}
}
#[derive(Clone, Debug)]
struct DenseFactorData<T: CompensatedField>
where
T::Real: Float,
{
factor: Mat<T>,
dense_gramian: Mat<T>,
dense_sqrt: Mat<T>,
}
#[derive(Clone, Debug)]
struct BalanceCore<T: CompensatedField>
where
T::Real: Float,
{
hankel_singular_values: Col<T::Real>,
reduced_order: usize,
error_bound: Option<T::Real>,
left_projection: Mat<T>,
right_projection: Mat<T>,
core: Mat<T>,
core_u: Mat<T>,
core_singular_values: Col<T::Real>,
core_vh: Mat<T>,
}
pub fn hsvd_from_dense_gramians<T>(
controllability_gramian: MatRef<'_, T>,
observability_gramian: MatRef<'_, T>,
params: &HsvdParams<T::Real>,
) -> Result<HsvdResult<T>, HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
validate_dense_gramians(controllability_gramian, observability_gramian)?;
let wc_factor = dense_psd_factor("controllability", controllability_gramian)?;
let wo_factor = dense_psd_factor("observability", observability_gramian)?;
let core = build_balance_core(wc_factor.factor.as_ref(), wo_factor.factor.as_ref(), params)?;
let internals = dense_internals(params.internals, wc_factor, wo_factor, &core);
Ok(HsvdResult {
hankel_singular_values: core.hankel_singular_values.clone(),
reduced_order: core.reduced_order,
error_bound: core.error_bound,
left_projection: core.left_projection.clone(),
right_projection: core.right_projection.clone(),
internals,
})
}
pub fn hsvd_from_factors<T>(
controllability_factor: MatRef<'_, T>,
observability_factor: MatRef<'_, T>,
params: &HsvdParams<T::Real>,
) -> Result<HsvdResult<T>, HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
validate_factor_rows(controllability_factor, observability_factor)?;
let core = build_balance_core(controllability_factor, observability_factor, params)?;
let internals = factor_internals(
params.internals,
controllability_factor.to_owned(),
observability_factor.to_owned(),
&core,
);
Ok(HsvdResult {
hankel_singular_values: core.hankel_singular_values.clone(),
reduced_order: core.reduced_order,
error_bound: core.error_bound,
left_projection: core.left_projection.clone(),
right_projection: core.right_projection.clone(),
internals,
})
}
fn validate_dense_gramians<T>(
controllability_gramian: MatRef<'_, T>,
observability_gramian: MatRef<'_, T>,
) -> Result<(), HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
let wc_ok = controllability_gramian.nrows() == controllability_gramian.ncols();
let wo_ok = observability_gramian.nrows() == observability_gramian.ncols();
let same = controllability_gramian.nrows() == observability_gramian.nrows();
if wc_ok && wo_ok && same {
Ok(())
} else {
Err(HsvdError::DimensionMismatch {
controllability_nrows: controllability_gramian.nrows(),
observability_nrows: observability_gramian.nrows(),
})
}
}
fn validate_factor_rows<T>(
controllability_factor: MatRef<'_, T>,
observability_factor: MatRef<'_, T>,
) -> Result<(), HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
if controllability_factor.nrows() == observability_factor.nrows() {
Ok(())
} else {
Err(HsvdError::DimensionMismatch {
controllability_nrows: controllability_factor.nrows(),
observability_nrows: observability_factor.nrows(),
})
}
}
fn dense_internals<T>(
level: HsvdInternalsLevel,
wc: DenseFactorData<T>,
wo: DenseFactorData<T>,
core: &BalanceCore<T>,
) -> Option<HsvdInternals<T>>
where
T: CompensatedField,
T::Real: Float,
{
if !level.keep_factors() {
return None;
}
Some(HsvdInternals {
controllability_factor: wc.factor,
observability_factor: wo.factor,
core: core.core.clone(),
core_u: core.core_u.clone(),
core_singular_values: core.core_singular_values.clone(),
core_vh: core.core_vh.clone(),
dense_controllability_gramian: level.keep_full().then_some(wc.dense_gramian),
dense_observability_gramian: level.keep_full().then_some(wo.dense_gramian),
dense_controllability_sqrt: level.keep_full().then_some(wc.dense_sqrt),
dense_observability_sqrt: level.keep_full().then_some(wo.dense_sqrt),
})
}
fn factor_internals<T>(
level: HsvdInternalsLevel,
controllability_factor: Mat<T>,
observability_factor: Mat<T>,
core: &BalanceCore<T>,
) -> Option<HsvdInternals<T>>
where
T: CompensatedField,
T::Real: Float,
{
if !level.keep_factors() {
return None;
}
Some(HsvdInternals {
controllability_factor,
observability_factor,
core: core.core.clone(),
core_u: core.core_u.clone(),
core_singular_values: core.core_singular_values.clone(),
core_vh: core.core_vh.clone(),
dense_controllability_gramian: None,
dense_observability_gramian: None,
dense_controllability_sqrt: None,
dense_observability_sqrt: None,
})
}
fn dense_psd_factor<T>(
which: &'static str,
gramian: MatRef<'_, T>,
) -> Result<DenseFactorData<T>, HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
let eig = dense_self_adjoint_eigen(gramian, &DenseDecompParams::<T>::new())?;
let mut max_abs = <T::Real as Zero>::zero();
for i in 0..eig.values.nrows() {
let abs = eig.values[i].abs();
if abs > max_abs {
max_abs = abs;
}
}
let eig_tol = <T::Real as Float>::epsilon().sqrt() * max_abs;
let mut kept = Vec::new();
for i in 0..eig.values.nrows() {
let lambda = eig.values[i].real();
if lambda < -eig_tol {
return Err(HsvdError::IndefiniteGramian {
which,
eigenvalue: lambda,
});
}
if lambda > eig_tol {
kept.push((i, lambda.sqrt()));
}
}
let factor = Mat::from_fn(gramian.nrows(), kept.len(), |row, col| {
eig.vectors[(row, kept[col].0)].mul_real(kept[col].1)
});
if !factor.as_ref().is_all_finite() {
return Err(HsvdError::NonFiniteResult { which });
}
Ok(DenseFactorData {
factor: factor.clone(),
dense_gramian: gramian.to_owned(),
dense_sqrt: factor,
})
}
fn build_balance_core<T>(
controllability_factor: MatRef<'_, T>,
observability_factor: MatRef<'_, T>,
params: &HsvdParams<T::Real>,
) -> Result<BalanceCore<T>, HsvdError<T::Real>>
where
T: CompensatedField,
T::Real: Float,
{
let core = dense_mul_adjoint_lhs(observability_factor, controllability_factor);
let svd = dense_svd(core.as_ref(), &DenseDecompParams::<T>::new())?;
let hankel_singular_values = Col::from_fn(svd.s.nrows(), |i| svd.s[i].abs());
let mut max_sigma = <T::Real as Zero>::zero();
for i in 0..hankel_singular_values.nrows() {
let sigma = hankel_singular_values[i];
if sigma > max_sigma {
max_sigma = sigma;
}
}
let sigma_tol = params
.sigma_tol
.unwrap_or_else(|| <T::Real as Float>::epsilon().sqrt() * max_sigma);
let available = (0..hankel_singular_values.nrows())
.take_while(|&i| hankel_singular_values[i] > sigma_tol)
.count();
let reduced_order = match params.order {
Some(order) if order > hankel_singular_values.nrows() => {
return Err(HsvdError::InvalidOrder {
requested: order,
available: hankel_singular_values.nrows(),
});
}
Some(order) if params.sigma_tol.is_some() => order.min(available),
Some(order) if order > available => {
return Err(HsvdError::InvalidOrder {
requested: order,
available,
});
}
Some(order) => order,
None => available,
};
if reduced_order == 0 && params.order != Some(0) {
return Err(HsvdError::EmptyRetainedSpectrum);
}
let error_bound = Some(
(<T::Real as One>::one() + <T::Real as One>::one())
* tail_sum(hankel_singular_values.as_ref(), reduced_order),
);
let right_projection = build_projection(
controllability_factor,
svd.v.as_ref(),
hankel_singular_values.as_ref(),
reduced_order,
);
let left_projection = build_projection(
observability_factor,
svd.u.as_ref(),
hankel_singular_values.as_ref(),
reduced_order,
);
if !right_projection.as_ref().is_all_finite() {
return Err(HsvdError::NonFiniteResult {
which: "right_projection",
});
}
if !left_projection.as_ref().is_all_finite() {
return Err(HsvdError::NonFiniteResult {
which: "left_projection",
});
}
Ok(BalanceCore {
hankel_singular_values: hankel_singular_values.clone(),
reduced_order,
error_bound,
left_projection,
right_projection,
core,
core_u: svd.u,
core_singular_values: hankel_singular_values,
core_vh: svd.v.adjoint().to_owned(),
})
}
fn build_projection<T>(
factor: MatRef<'_, T>,
basis: MatRef<'_, T>,
hankel_singular_values: faer::ColRef<'_, T::Real>,
order: usize,
) -> Mat<T>
where
T: CompensatedField,
T::Real: Float,
{
let scaled_basis = Mat::from_fn(basis.nrows(), order, |row, col| {
let sigma_inv_sqrt = hankel_singular_values[col].sqrt().recip();
basis[(row, col)].mul_real(sigma_inv_sqrt)
});
dense_mul(factor, scaled_basis.as_ref())
}
fn tail_sum<R>(values: faer::ColRef<'_, R>, from: usize) -> R
where
R: Float,
{
let mut sum = <R as Zero>::zero();
for i in from..values.nrows() {
sum = sum + values[i];
}
sum
}
#[cfg(test)]
mod tests {
use super::{
HsvdError, HsvdInternalsLevel, HsvdParams, hsvd_from_dense_gramians, hsvd_from_factors,
};
use faer::Mat;
fn assert_close(lhs: &Mat<f64>, rhs: &Mat<f64>, tol: f64) {
assert_eq!(lhs.nrows(), rhs.nrows());
assert_eq!(lhs.ncols(), rhs.ncols());
for col in 0..lhs.ncols() {
for row in 0..lhs.nrows() {
let err = (lhs[(row, col)] - rhs[(row, col)]).abs();
assert!(err <= tol, "entry ({row}, {col}) mismatch: {err} > {tol}");
}
}
}
#[test]
fn dense_hsvd_matches_diagonal_gramian_case() {
let wc = Mat::from_fn(2, 2, |row, col| match (row, col) {
(0, 0) => 4.0,
(1, 1) => 1.0,
_ => 0.0,
});
let wo = Mat::from_fn(2, 2, |row, col| match (row, col) {
(0, 0) => 9.0,
(1, 1) => 16.0,
_ => 0.0,
});
let hsvd = hsvd_from_dense_gramians(
wc.as_ref(),
wo.as_ref(),
&HsvdParams::new().with_internals(HsvdInternalsLevel::Full),
)
.unwrap();
assert_eq!(hsvd.reduced_order, 2);
let sigma0: f64 = hsvd.hankel_singular_values[0];
let sigma1: f64 = hsvd.hankel_singular_values[1];
assert!((sigma0 - 6.0_f64).abs() <= 1.0e-12_f64);
assert!((sigma1 - 4.0_f64).abs() <= 1.0e-12_f64);
let internals = hsvd.internals.unwrap();
assert!(internals.dense_controllability_gramian.is_some());
assert!(internals.dense_observability_gramian.is_some());
}
#[test]
fn factor_hsvd_matches_dense_hsvd_for_same_factors() {
let rc = Mat::from_fn(3, 2, |row, col| match (row, col) {
(0, 0) => 2.0,
(1, 1) => 1.0,
(2, 0) => 0.5,
_ => 0.0,
});
let ro = Mat::from_fn(3, 2, |row, col| match (row, col) {
(0, 0) => 1.5,
(1, 1) => 3.0,
(2, 1) => -0.25,
_ => 0.0,
});
let wc = super::dense_mul(rc.as_ref(), rc.adjoint().to_owned().as_ref());
let wo = super::dense_mul(ro.as_ref(), ro.adjoint().to_owned().as_ref());
let dense = hsvd_from_dense_gramians(wc.as_ref(), wo.as_ref(), &HsvdParams::new()).unwrap();
let factor = hsvd_from_factors(rc.as_ref(), ro.as_ref(), &HsvdParams::new()).unwrap();
assert_eq!(dense.reduced_order, factor.reduced_order);
for i in 0..dense.hankel_singular_values.nrows() {
let dense_sigma: f64 = dense.hankel_singular_values[i];
let factor_sigma: f64 = factor.hankel_singular_values[i];
assert!((dense_sigma - factor_sigma).abs() <= 1.0e-10);
}
let dense_identity = super::dense_mul_adjoint_lhs(
dense.left_projection.as_ref(),
dense.right_projection.as_ref(),
);
let factor_identity = super::dense_mul_adjoint_lhs(
factor.left_projection.as_ref(),
factor.right_projection.as_ref(),
);
let expected_identity = Mat::<f64>::identity(dense.reduced_order, dense.reduced_order);
assert_close(
&dense_identity.as_ref().to_owned(),
&expected_identity,
1.0e-10,
);
assert_close(
&factor_identity.as_ref().to_owned(),
&expected_identity,
1.0e-10,
);
}
#[test]
fn rejects_invalid_order() {
let rc = Mat::from_fn(2, 1, |row, _| if row == 0 { 1.0 } else { 0.5 });
let ro = Mat::from_fn(2, 1, |row, _| if row == 0 { 1.0 } else { 0.25 });
let err = hsvd_from_factors(rc.as_ref(), ro.as_ref(), &HsvdParams::new().with_order(2))
.unwrap_err();
assert!(matches!(
err,
HsvdError::InvalidOrder {
requested: 2,
available: 1
}
));
}
}