use ndarray::ArrayView2;
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct RoutabilityFloor {
pub p: usize,
pub n_blocks: usize,
pub b_max: usize,
pub delta: f64,
pub floor: f64,
}
pub fn routability_floor(p: usize, n_blocks: usize, b_max: usize, delta: f64) -> RoutabilityFloor {
assert!(p > 0, "routability_floor requires p >= 1");
assert!(n_blocks > 0, "routability_floor requires n_blocks >= 1");
assert!(b_max > 0, "routability_floor requires b_max >= 1");
assert!(
b_max <= p,
"routability_floor requires b_max <= p (a b-frame must fit in R^p)"
);
assert!(
delta.is_finite() && delta > 0.0,
"routability_floor requires a finite delta > 0"
);
let pf = p as f64;
let subspace = (b_max as f64 / pf).sqrt();
let log_arg = (n_blocks as f64) / delta;
let union = (2.0 * log_arg.ln().max(0.0) / pf).sqrt();
let floor = subspace + union;
RoutabilityFloor {
p,
n_blocks,
b_max,
delta,
floor,
}
}
pub fn minimum_routable_energy(floor: &RoutabilityFloor) -> f64 {
let f2 = floor.floor * floor.floor;
f2 / (1.0 + f2)
}
#[derive(Clone, Debug)]
pub struct RoutabilityAudit {
pub n_rows: usize,
pub floor: RoutabilityFloor,
pub quantiles: Vec<(f64, f64)>,
pub empirical_mean: f64,
pub empirical_max: f64,
pub confidence_quantile: f64,
pub coherence_excess: f64,
pub fraction_below_floor: f64,
}
pub fn routability_audit(
decoder: ArrayView2<'_, f32>,
residuals: ArrayView2<'_, f32>,
block_size: usize,
delta: f64,
quantile_levels: &[f64],
) -> Result<RoutabilityAudit, String> {
let k_rows = decoder.nrows();
let p = decoder.ncols();
if k_rows == 0 || p == 0 {
return Err("routability_audit: decoder must be a non-empty K×P matrix".to_string());
}
if block_size == 0 {
return Err("routability_audit: block_size must be >= 1".to_string());
}
if k_rows % block_size != 0 {
return Err(format!(
"routability_audit: decoder has K={k_rows} rows, not a multiple of block_size {block_size}"
));
}
if residuals.ncols() != p {
return Err(format!(
"routability_audit: residuals have P={} columns but the decoder has P={p}",
residuals.ncols()
));
}
if !quantile_levels.iter().all(|&q| (0.0..=1.0).contains(&q)) {
return Err("routability_audit: quantile levels must lie in [0, 1]".to_string());
}
let n_blocks = k_rows / block_size;
let floor = routability_floor(p, n_blocks, block_size, delta);
let mut per_row: Vec<f64> = Vec::with_capacity(residuals.nrows());
for r in residuals.outer_iter() {
let mut norm2 = 0.0f64;
for &v in r.iter() {
norm2 += v as f64 * v as f64;
}
let norm = norm2.sqrt();
if norm <= 1.0e-12 {
continue; }
let mut best = 0.0f64;
for g in 0..n_blocks {
let mut energy = 0.0f64;
for row_off in 0..block_size {
let atom = decoder.row(g * block_size + row_off);
let mut dot = 0.0f64;
for (rv, av) in r.iter().zip(atom.iter()) {
dot += *rv as f64 * *av as f64;
}
energy += dot * dot;
}
let gate = energy.sqrt();
if gate > best {
best = gate;
}
}
per_row.push(best / norm);
}
let n_rows = per_row.len();
if n_rows == 0 {
return Err("routability_audit: no residual rows with positive norm".to_string());
}
let mut sorted = per_row.clone();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let quantile_at = |level: f64| -> f64 {
let idx = (level * (n_rows - 1) as f64).round() as usize;
sorted[idx.min(n_rows - 1)]
};
let quantiles: Vec<(f64, f64)> = quantile_levels
.iter()
.map(|&level| (level, quantile_at(level)))
.collect();
let empirical_mean = per_row.iter().sum::<f64>() / n_rows as f64;
let empirical_max = sorted[n_rows - 1];
let confidence_quantile = quantile_at((1.0 - delta).clamp(0.0, 1.0));
let coherence_excess = if floor.floor > 0.0 {
confidence_quantile / floor.floor
} else {
f64::INFINITY
};
let fraction_below_floor =
per_row.iter().filter(|&&v| v <= floor.floor).count() as f64 / n_rows as f64;
Ok(RoutabilityAudit {
n_rows,
floor,
quantiles,
empirical_mean,
empirical_max,
confidence_quantile,
coherence_excess,
fraction_below_floor,
})
}
#[cfg(test)]
#[path = "routability_tests.rs"]
mod routability_tests;