use log::info;
use nalgebra::DMatrix;
use rand::rngs::SmallRng;
use rand::seq::SliceRandom;
use rand::{RngExt, SeedableRng};
use rayon::prelude::*;
use rustc_hash::FxHashMap as HashMap;
use std::collections::VecDeque;
pub const LOG_EPS: f64 = 1e-9;
#[inline]
fn fast_ln(x: f64) -> f64 {
f64::from((x as f32).ln())
}
#[derive(Clone, Debug)]
pub enum ProfileSource {
Raw,
Projected { basis: DMatrix<f32> },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FeatureWeighting {
None,
FisherInfoNb,
}
#[derive(Clone, Debug)]
pub struct RefineParams {
pub num_gibbs: usize,
pub num_greedy: usize,
pub feature_weighting: FeatureWeighting,
pub seed: u64,
pub gibbs_stagnation: f64,
pub profile_source: ProfileSource,
pub parallel: bool,
pub prior_count: f32,
pub min_gain_z: f32,
pub size_floor: f32,
pub size_cap: f32,
}
impl Default for RefineParams {
fn default() -> Self {
Self {
num_gibbs: 20,
num_greedy: 10,
feature_weighting: FeatureWeighting::FisherInfoNb,
seed: 42,
gibbs_stagnation: 0.005,
profile_source: ProfileSource::Raw,
parallel: true,
prior_count: 0.5,
min_gain_z: 1.5,
size_floor: 0.9,
size_cap: 1.25,
}
}
}
pub const BATCH_FACTOR_PSEUDO_COUNT: f64 = 10.0;
pub const BATCH_FACTOR_BOUND: f32 = 8.0;
const BATCH_FACTOR_ROUNDS: usize = 4;
pub struct Profiles {
pub rows: Vec<Vec<(u32, f32)>>,
pub size_factor: Vec<f32>,
pub num_entities: usize,
pub num_features: usize,
}
impl Profiles {
pub fn from_gene_sums(gene_sums: &[Vec<(usize, f32)>], num_features: usize) -> Self {
let num_entities = gene_sums.len();
let (rows, size_factor): (Vec<Vec<(u32, f32)>>, Vec<f32>) = gene_sums
.par_iter()
.map(|row| {
let mut out: Vec<(u32, f32)> = row
.iter()
.filter(|(_, v)| *v > 0.0)
.map(|(g, v)| (*g as u32, *v))
.collect();
out.sort_unstable_by_key(|&(g, _)| g);
let sf: f32 = out.iter().map(|(_, v)| *v).sum();
(out, sf)
})
.unzip();
Self {
rows,
size_factor,
num_entities,
num_features,
}
}
pub fn from_projection(basis: &DMatrix<f32>, entity_to_cells: &[Vec<usize>]) -> Self {
let num_features = basis.nrows();
let num_entities = entity_to_cells.len();
let (rows, size_factor): (Vec<Vec<(u32, f32)>>, Vec<f32>) = entity_to_cells
.par_iter()
.map(|cells| {
let mut acc = vec![0f32; num_features];
for &c in cells {
for d in 0..num_features {
acc[d] += basis[(d, c)];
}
}
let mut out = Vec::with_capacity(num_features);
let mut sf = 0f32;
for (d, &v) in acc.iter().enumerate() {
assert!(
v >= 0.0,
"projection profile is negative at dim {d} ({v}); \
DC-Poisson requires a nonnegative basis"
);
if v > 0.0 {
out.push((d as u32, v));
sf += v;
}
}
(out, sf)
})
.unzip();
Self {
rows,
size_factor,
num_entities,
num_features,
}
}
pub fn weight_by_vec(&mut self, w: &[f32]) {
assert_eq!(
w.len(),
self.num_features,
"weight vector length must match num_features"
);
self.rows
.par_iter_mut()
.zip(self.size_factor.par_iter_mut())
.for_each(|(row, sf)| {
let mut new_sf = 0f32;
for (g, v) in row.iter_mut() {
*v *= w[*g as usize];
new_sf += *v;
}
*sf = new_sf;
});
}
pub fn rescale_batches(&mut self, batch: &[usize], strata: Option<&[usize]>) {
let n = self.rows.len();
assert_eq!(batch.len(), n, "one batch per entity");
let nb = batch.iter().copied().max().map_or(0, |b| b + 1);
if nb < 2 {
return;
}
let single = vec![0usize; if strata.is_none() { n } else { 0 }];
let strata = strata.unwrap_or(&single);
assert_eq!(strata.len(), n, "one stratum per entity");
let ns = strata.iter().copied().max().map_or(0, |c| c + 1);
let m = self.num_features;
let mut observed = vec![0f64; nb * m];
let mut mass = vec![0f64; ns * nb];
let mut batch_members = vec![0usize; nb];
let mut of_stratum: Vec<Vec<u32>> = vec![Vec::new(); ns];
for (e, row) in self.rows.iter().enumerate() {
let (b, c) = (batch[e], strata[e]);
for &(g, v) in row {
observed[b * m + g as usize] += v as f64;
}
mass[c * nb + b] += self.size_factor[e] as f64;
batch_members[b] += 1;
of_stratum[c].push(e as u32);
}
let mut dense = vec![0f64; m];
let mut touched: Vec<u32> = Vec::new();
let mut stratum_sums: Vec<Vec<(u32, f64)>> = Vec::with_capacity(ns);
let mut stratum_batches: Vec<Vec<(usize, f64)>> = Vec::with_capacity(ns);
for (c, members) in of_stratum.iter().enumerate() {
touched.clear();
for &e in members {
for &(g, v) in &self.rows[e as usize] {
if dense[g as usize] == 0.0 {
touched.push(g);
}
dense[g as usize] += v as f64;
}
}
touched.sort_unstable();
touched.dedup();
stratum_sums.push(touched.iter().map(|&g| (g, dense[g as usize])).collect());
touched.iter().for_each(|&g| dense[g as usize] = 0.0);
stratum_batches.push(
(0..nb)
.map(|b| (b, mass[c * nb + b]))
.filter(|&(_, t)| t > 0.0)
.collect(),
);
}
let mut factor = vec![1f32; nb * m];
let mut expected = vec![0f64; nb * m];
for _ in 0..BATCH_FACTOR_ROUNDS {
expected.iter_mut().for_each(|x| *x = 0.0);
for (sums, present) in stratum_sums.iter().zip(&stratum_batches) {
for &(g, sum) in sums {
let g = g as usize;
let exposure: f64 = present
.iter()
.map(|&(b, t)| t * factor[b * m + g] as f64)
.sum();
if exposure > 0.0 {
let rate = sum / exposure;
for &(b, t) in present {
expected[b * m + g] += t * rate;
}
}
}
}
for ((f, &o), &x) in factor.iter_mut().zip(&observed).zip(&expected) {
*f = ((o + BATCH_FACTOR_PSEUDO_COUNT) / (x + BATCH_FACTOR_PSEUDO_COUNT)) as f32;
}
}
for (b, per_feature) in factor.chunks_mut(m).enumerate() {
let members = batch_members[b].max(1) as f32;
let trust = (members - 1.0) / members;
for f in per_feature {
*f = f
.powf(trust)
.clamp(1.0 / BATCH_FACTOR_BOUND, BATCH_FACTOR_BOUND);
}
}
self.rows
.par_iter_mut()
.zip(self.size_factor.par_iter_mut())
.zip(batch.par_iter())
.for_each(|((row, sf), &b)| {
let mut new_sf = 0f32;
for (g, v) in row.iter_mut() {
*v /= factor[b * m + *g as usize];
new_sf += *v;
}
*sf = new_sf;
});
}
pub fn apply_feature_weighting(&mut self, method: FeatureWeighting) {
match method {
FeatureWeighting::None => {}
FeatureWeighting::FisherInfoNb => {
let w = self.nb_fisher_weights();
self.weight_by_vec(&w);
}
}
}
pub fn nb_fisher_weights(&self) -> Vec<f32> {
use crate::alg::nb_dispersion::DispersionTrend;
use legume_numeric::matrix::sparse_stat::SparseRunningStatistics;
use legume_numeric::matrix::traits::RunningStatOps;
const ROWS_PER_LEAF: usize = 256;
let num_features = self.num_features;
let (stats, ..) = crate::qc::fold_in_fixed_order(
&self.rows,
ROWS_PER_LEAF,
&|| {
(
SparseRunningStatistics::<f32>::new(num_features),
Vec::<usize>::new(),
Vec::<f32>::new(),
)
},
&|(acc, col_rows, col_vals), row| {
col_rows.clear();
col_vals.clear();
for &(g, v) in row {
col_rows.push(g as usize);
col_vals.push(v);
}
acc.add_sparse_column(col_rows, col_vals);
Ok(())
},
&|(a, ..), (b, ..)| a.merge(&b),
)
.expect("the row fold cannot fail");
let trend = DispersionTrend::from_sparse_stats(&stats);
let means = stats.mean();
let sums = stats.sum();
let total_mass: f64 = sums.iter().map(|&s| s as f64).sum();
let avg_s = if self.num_entities > 0 {
(total_mass / self.num_entities as f64) as f32
} else {
1.0
};
let inv_total = if total_mass > 0.0 {
1.0 / total_mass as f32
} else {
0.0
};
(0..self.num_features)
.map(|g| trend.fisher_weight(sums[g] * inv_total, avg_s, means[g]))
.collect()
}
}
const NULL_STEP: f64 = 1.0 / 48.0;
const NULL_LN_MIN: f64 = -20.0;
const NULL_MU_MAX: f64 = 64.0;
#[derive(Clone)]
struct NullLog {
c: f64,
ln_c: f64,
mean: Vec<f64>,
var: Vec<f64>,
}
impl NullLog {
fn new(c: f64) -> Self {
let points = ((NULL_MU_MAX.ln() - NULL_LN_MIN) / NULL_STEP).ceil() as usize + 2;
let (mean, var) = (0..points)
.map(|i| Self::poisson_moments((NULL_LN_MIN + i as f64 * NULL_STEP).exp(), c))
.unzip();
Self {
c,
ln_c: fast_ln(c),
mean,
var,
}
}
fn poisson_moments(mu: f64, c: f64) -> (f64, f64) {
let last = (mu + 12.0 * mu.sqrt() + 25.0) as usize;
let mut p = (-mu).exp();
let (mut m1, mut m2) = (0f64, 0f64);
for n in 0..=last {
let l = (n as f64 / c).ln_1p();
m1 += p * l;
m2 += p * l * l;
p *= mu / (n as f64 + 1.0);
}
(m1, (m2 - m1 * m1).max(0.0))
}
#[inline]
fn at(&self, ln_mu: f64, mu: f64) -> (f64, f64) {
if mu >= NULL_MU_MAX {
let m = mu + self.c;
let r = 1.0 / m;
let v = mu * r * r;
let mean =
m.ln() - 0.5 * v + v * r / 3.0 - (3.0 * mu * mu + mu) * (r * r) * (r * r) / 4.0;
(mean, v)
} else if ln_mu > NULL_LN_MIN {
let pos = (ln_mu - NULL_LN_MIN) / NULL_STEP;
let i = pos as usize;
let f = pos - i as f64;
(
self.ln_c + self.mean[i] + f * (self.mean[i + 1] - self.mean[i]),
self.var[i] + f * (self.var[i + 1] - self.var[i]),
)
} else {
(self.ln_c, 0.0)
}
}
fn row(&self, size: f64, classes: &RateClasses, mean: &mut [f64], var: &mut [f64]) {
let ln_size = size.ln();
for (((m, v), &rate), &ln_rate) in mean
.iter_mut()
.zip(var.iter_mut())
.zip(&classes.rate)
.zip(&classes.ln_rate)
{
(*m, *v) = self.at(ln_size + ln_rate, size * rate);
}
}
}
const CLASS_STEP: f64 = std::f64::consts::LN_2 / 4.0;
const MAX_CLASSES: usize = 256;
#[derive(Clone)]
struct RateClasses {
top: f64,
rate: Vec<f64>,
ln_rate: Vec<f64>,
}
impl RateClasses {
fn new(max_pooled: f64, min_entry: f64, total: f64, rest_min: f64) -> Self {
let ln2 = std::f64::consts::LN_2;
let (top, len) = if total > 0.0 && max_pooled > 0.0 && min_entry.is_finite() {
let rest_min = if rest_min > 0.0 { rest_min } else { total };
let top = (max_pooled / rest_min.min(total)).ln() + ln2;
let bottom = (min_entry / total).ln() - ln2;
let span = ((top - bottom) / CLASS_STEP).ceil() as usize;
(top, (span + 2).clamp(3, MAX_CLASSES))
} else {
(0.0, 2)
};
let mut ln_rate: Vec<f64> = (0..len - 1).map(|j| top - j as f64 * CLASS_STEP).collect();
let mut rate: Vec<f64> = ln_rate.iter().map(|x| x.exp()).collect();
ln_rate.push(f64::NEG_INFINITY);
rate.push(0.0);
Self { top, rate, ln_rate }
}
fn len(&self) -> usize {
self.rate.len()
}
#[inline]
fn add(&self, pooled: f64, total: f64, weight: f64, mass: &mut [f64], sq: &mut [f64]) {
let zero = self.len() - 1;
if !(pooled > 0.0 && total > 0.0) {
mass[zero] += weight;
sq[zero] += weight * weight;
return;
}
let pos = ((self.top - fast_ln(pooled / total)) / CLASS_STEP).clamp(0.0, (zero - 1) as f64);
let lo = pos as usize;
let f = pos - lo as f64;
mass[lo] += weight * (1.0 - f);
sq[lo] += weight * weight * (1.0 - f);
if f > 0.0 {
mass[lo + 1] += weight * f;
sq[lo + 1] += weight * weight * f;
}
}
}
#[inline(always)]
fn class_dot(w: &[f32], t: &[f64]) -> f64 {
w.iter().zip(t).map(|(&w, &t)| w as f64 * t).sum()
}
#[derive(Clone)]
pub struct DcPoissonStats {
pub k: usize,
pub num_features: usize,
pub membership: Vec<usize>,
pub gene_sum: Vec<f64>,
pub size_sum: Vec<f64>,
pub log_gene: Vec<f32>,
pub log_size_offset: Vec<f32>,
pub count: Vec<usize>,
pub min_gain_z: f64,
prior: f64,
null: NullLog,
classes: RateClasses,
class_mass: Vec<f32>,
class_sq: Vec<f32>,
class_null: Vec<f64>,
class_var: Vec<f64>,
}
impl DcPoissonStats {
#[inline(always)]
pub fn idx(&self, k: usize, g: usize) -> usize {
g * self.k + k
}
pub fn from_profiles(profiles: &Profiles, k: usize, membership: &[usize]) -> Self {
Self::from_profiles_with(profiles, k, membership, &RefineParams::default())
}
pub fn from_profiles_with(
profiles: &Profiles,
k: usize,
membership: &[usize],
params: &RefineParams,
) -> Self {
let m = profiles.num_features;
let mut gene_sum = vec![0f64; k * m];
let mut size_sum = vec![0f64; k];
let mut count = vec![0usize; k];
let mut pooled = vec![0f64; m];
let mut min_entry = f64::INFINITY;
for (e, row) in profiles.rows.iter().enumerate() {
let z = membership[e];
assert!(z < k, "membership[{}]={} out of range 0..{}", e, z, k);
for &(g, v) in row {
gene_sum[g as usize * k + z] += v as f64;
pooled[g as usize] += v as f64;
if v > 0.0 {
min_entry = min_entry.min(v as f64);
}
}
size_sum[z] += profiles.size_factor[e] as f64;
count[z] += 1;
}
let total: f64 = pooled.iter().sum();
let max_pooled = pooled.iter().copied().fold(0f64, f64::max);
let max_sf = profiles
.size_factor
.iter()
.fold(0f64, |a, &s| a.max(s as f64));
let classes = RateClasses::new(max_pooled, min_entry, total, total - max_sf);
let jj = classes.len();
let n = profiles.rows.len();
let mut class_mass = vec![0f32; n * jj];
let mut class_sq = vec![0f32; n * jj];
class_mass
.par_chunks_mut(jj)
.zip(class_sq.par_chunks_mut(jj))
.zip(profiles.rows.par_iter())
.zip(profiles.size_factor.par_iter())
.for_each_init(
|| (vec![0f64; jj], vec![0f64; jj]),
|(mass, sq), (((mass_out, sq_out), row), &sf)| {
mass.iter_mut().for_each(|x| *x = 0.0);
sq.iter_mut().for_each(|x| *x = 0.0);
let rest = total - sf as f64;
for &(g, v) in row {
if v > 0.0 {
let p = pooled[g as usize];
let others = p - v as f64;
let others = if others > 1e-9 * p { others } else { 0.0 };
classes.add(others, rest, v as f64, mass, sq);
}
}
for (o, &x) in mass_out.iter_mut().zip(mass.iter()) {
*o = x as f32;
}
for (o, &x) in sq_out.iter_mut().zip(sq.iter()) {
*o = x as f32;
}
},
);
let prior = (params.prior_count as f64).max(LOG_EPS);
let mut stats = Self {
k,
num_features: m,
membership: membership.to_vec(),
gene_sum,
size_sum,
log_gene: vec![0f32; k * m],
log_size_offset: vec![0f32; k],
count,
min_gain_z: params.min_gain_z.max(0.0) as f64,
prior,
null: NullLog::new(prior),
classes,
class_mass,
class_sq,
class_null: vec![0f64; k * jj],
class_var: vec![0f64; k * jj],
};
stats.refresh_caches();
stats
}
#[inline]
fn refresh_block(&mut self, k: usize) {
let jj = self.classes.len();
self.log_size_offset[k] =
-fast_ln(self.size_sum[k] + self.num_features as f64 * LOG_EPS) as f32;
self.null.row(
self.size_sum[k],
&self.classes,
&mut self.class_null[k * jj..(k + 1) * jj],
&mut self.class_var[k * jj..(k + 1) * jj],
);
}
fn refresh_caches(&mut self) {
let c = self.prior;
self.log_gene
.par_iter_mut()
.zip(self.gene_sum.par_iter())
.with_min_len(1 << 14)
.for_each(|(l, &s)| *l = fast_ln(s + c) as f32);
for b in 0..self.k {
self.refresh_block(b);
}
}
pub fn delta_move(&mut self, e: usize, k_from: usize, k_to: usize, profiles: &Profiles) {
if k_from == k_to {
return;
}
let kk = self.k;
let c = self.prior;
for &(g, v) in &profiles.rows[e] {
let base = g as usize * kk;
let (i_from, i_to) = (base + k_from, base + k_to);
self.gene_sum[i_from] = (self.gene_sum[i_from] - v as f64).max(0.0);
self.gene_sum[i_to] += v as f64;
self.log_gene[i_from] = fast_ln(self.gene_sum[i_from] + c) as f32;
self.log_gene[i_to] = fast_ln(self.gene_sum[i_to] + c) as f32;
}
let sf = profiles.size_factor[e] as f64;
self.size_sum[k_from] = (self.size_sum[k_from] - sf).max(0.0);
self.size_sum[k_to] += sf;
self.count[k_from] = self.count[k_from].saturating_sub(1);
self.count[k_to] += 1;
self.refresh_block(k_from);
self.refresh_block(k_to);
self.membership[e] = k_to;
}
#[cfg(test)]
fn recompute(&mut self, profiles: &Profiles) {
let k = self.k;
self.gene_sum.iter_mut().for_each(|x| *x = 0.0);
self.size_sum.iter_mut().for_each(|x| *x = 0.0);
self.count.iter_mut().for_each(|x| *x = 0);
for (e, row) in profiles.rows.iter().enumerate() {
let z = self.membership[e];
for &(g, v) in row {
self.gene_sum[g as usize * k + z] += v as f64;
}
self.size_sum[z] += profiles.size_factor[e] as f64;
self.count[z] += 1;
}
self.refresh_caches();
}
#[inline(always)]
fn entity_classes(&self, e: usize) -> (&[f32], &[f32]) {
let jj = self.classes.len();
(
&self.class_mass[e * jj..(e + 1) * jj],
&self.class_sq[e * jj..(e + 1) * jj],
)
}
#[inline(always)]
fn block_classes(&self, k: usize) -> (&[f64], &[f64]) {
let jj = self.classes.len();
(
&self.class_null[k * jj..(k + 1) * jj],
&self.class_var[k * jj..(k + 1) * jj],
)
}
}
#[cfg(test)]
fn score_move(e: usize, k: usize, stats: &DcPoissonStats, profiles: &Profiles) -> f64 {
let jj = stats.classes.len();
let sf = profiles.size_factor[e] as f64;
let row = &profiles.rows[e];
let cur = stats.membership[e];
let (mass, sq) = stats.entity_classes(e);
let (mut loo_mean, mut loo_var) = ([0f64; MAX_CLASSES], [0f64; MAX_CLASSES]);
let loo_size = (stats.size_sum[cur] - sf).max(0.0);
stats.null.row(
loo_size,
&stats.classes,
&mut loo_mean[..jj],
&mut loo_var[..jj],
);
if cur == k {
let mut acc = -class_dot(mass, &loo_mean[..jj]);
for &(g, v) in row {
let vg = v as f64;
let loo = (stats.gene_sum[stats.idx(k, g as usize)] - vg).max(0.0);
acc += vg * fast_ln(loo + stats.prior);
}
acc
} else {
let (null, var) = stats.block_classes(k);
let mut acc = -class_dot(mass, null);
for &(g, v) in row {
acc += v as f64 * stats.log_gene[stats.idx(k, g as usize)] as f64;
}
if stats.min_gain_z > 0.0 {
let noise = class_dot(sq, &loo_var[..jj]) + class_dot(sq, var);
acc -= stats.min_gain_z * noise.sqrt();
}
acc
}
}
pub fn compute_log_probs_restricted(
e: usize,
stats: &DcPoissonStats,
profiles: &Profiles,
allowed: &[usize],
log_probs: &mut [f64],
) {
let kk = stats.k;
let jj = stats.classes.len();
let sf = profiles.size_factor[e] as f64;
let row = &profiles.rows[e];
let cur = stats.membership[e];
let c = stats.prior;
let (mass, sq) = stats.entity_classes(e);
let dense = allowed.len() == kk;
let score_cur = dense || allowed.contains(&cur);
let z = stats.min_gain_z;
let (mut loo_mean, mut loo_var) = ([0f64; MAX_CLASSES], [0f64; MAX_CLASSES]);
if score_cur || z > 0.0 {
let loo_size = (stats.size_sum[cur] - sf).max(0.0);
stats.null.row(
loo_size,
&stats.classes,
&mut loo_mean[..jj],
&mut loo_var[..jj],
);
}
let mut cur_acc = if score_cur {
-class_dot(mass, &loo_mean[..jj])
} else {
0.0
};
let mut add_cur = |base: usize, vg: f64| {
if score_cur {
let loo = (stats.gene_sum[base + cur] - vg).max(0.0);
cur_acc += vg * fast_ln(loo + c);
}
};
let start = |k: usize| -class_dot(mass, stats.block_classes(k).0);
let noise_cur = if z > 0.0 {
class_dot(sq, &loo_var[..jj])
} else {
0.0
};
let charge = |k: usize| z * (noise_cur + class_dot(sq, stats.block_classes(k).1)).sqrt();
if dense {
for (k, slot) in log_probs[..kk].iter_mut().enumerate() {
*slot = start(k);
}
for &(g, v) in row {
let vg = v as f64;
let base = g as usize * kk;
let logs = &stats.log_gene[base..base + kk];
for (slot, &l) in log_probs[..kk].iter_mut().zip(logs) {
*slot += vg * l as f64;
}
add_cur(base, vg);
}
if z > 0.0 {
for (k, slot) in log_probs[..kk].iter_mut().enumerate() {
*slot -= charge(k);
}
}
} else {
thread_local! {
static SCRATCH: std::cell::RefCell<Vec<f64>> = const { std::cell::RefCell::new(Vec::new()) };
}
SCRATCH.with(|scratch| {
let mut scratch = scratch.borrow_mut();
scratch.clear();
scratch.extend(allowed.iter().map(|&k| start(k)));
for &(g, v) in row {
let vg = v as f64;
let base = g as usize * kk;
let logs = &stats.log_gene[base..base + kk];
for (acc, &k) in scratch.iter_mut().zip(allowed) {
*acc += vg * logs[k] as f64;
}
add_cur(base, vg);
}
for (&acc, &k) in scratch.iter().zip(allowed) {
log_probs[k] = if z > 0.0 { acc - charge(k) } else { acc };
}
});
}
if score_cur {
log_probs[cur] = cur_acc;
}
}
#[cfg(test)]
fn compute_log_probs(e: usize, stats: &DcPoissonStats, profiles: &Profiles, log_probs: &mut [f64]) {
for (k, slot) in log_probs.iter_mut().enumerate().take(stats.k) {
*slot = score_move(e, k, stats, profiles);
}
}
pub fn sample_categorical_log_restricted(
log_probs: &[f64],
allowed: &[usize],
current: usize,
rng: &mut SmallRng,
) -> usize {
let mut best_key = f64::NEG_INFINITY;
let mut best_idx = current;
for &k in allowed {
let lp = log_probs[k];
if lp.is_finite() {
let key = gumbel_key(lp, rng);
if key > best_key {
best_key = key;
best_idx = k;
}
}
}
best_idx
}
#[inline]
fn gumbel_key(lp: f64, rng: &mut SmallRng) -> f64 {
let u: f64 = rng.random_range(1e-12..1.0_f64);
lp - (-u.ln()).ln()
}
pub fn argmax_log_restricted(log_probs: &[f64], allowed: &[usize]) -> usize {
let mut best = allowed[0];
let mut best_val = log_probs[best];
for &k in &allowed[1..] {
if log_probs[k] > best_val {
best = k;
best_val = log_probs[k];
}
}
best
}
pub fn argmax_log_restricted_from(log_probs: &[f64], allowed: &[usize], current: usize) -> usize {
let mut best = if allowed.contains(¤t) {
current
} else {
allowed[0]
};
let mut best_val = log_probs[best];
for &k in allowed {
if log_probs[k] > best_val {
best = k;
best_val = log_probs[k];
}
}
best
}
#[inline]
fn staying_score(log_probs: &[f64], allowed: &[usize], current: usize) -> f64 {
if allowed.contains(¤t) {
log_probs[current]
} else {
f64::NEG_INFINITY
}
}
fn sample_improving_log(
log_probs: &[f64],
allowed: &[usize],
current: usize,
rng: &mut SmallRng,
) -> usize {
let cur_val = staying_score(log_probs, allowed, current);
let mut best_idx = current;
let mut best_key = if cur_val.is_finite() {
gumbel_key(cur_val, rng)
} else {
f64::NEG_INFINITY
};
for &k in allowed {
let lp = log_probs[k];
if k != current && lp.is_finite() && lp > cur_val {
let key = gumbel_key(lp, rng);
if key > best_key {
best_key = key;
best_idx = k;
}
}
}
best_idx
}
fn argmax_log_staying(log_probs: &[f64], allowed: &[usize], current: usize) -> usize {
let mut best = current;
let mut best_val = staying_score(log_probs, allowed, current);
for &k in allowed {
if k != current && log_probs[k] > best_val {
best = k;
best_val = log_probs[k];
}
}
best
}
pub fn compact_labels<K>(labels: &[K]) -> (Vec<usize>, usize)
where
K: Copy + Eq + std::hash::Hash,
{
let mut map: HashMap<K, usize> = HashMap::default();
let mut next = 0usize;
let mut out = Vec::with_capacity(labels.len());
for &g in labels {
let new = *map.entry(g).or_insert_with(|| {
let id = next;
next += 1;
id
});
out.push(new);
}
(out, next)
}
pub fn compute_sibling_sets(
refined: &[Vec<usize>],
level: usize,
num_groups_at_level: usize,
) -> Vec<Vec<usize>> {
let num_levels = refined.len();
let num_entities = refined[level].len();
if level + 1 >= num_levels {
let all_groups: Vec<usize> = (0..num_groups_at_level).collect();
return vec![all_groups; num_entities];
}
let mut parent_to_children: HashMap<usize, Vec<usize>> = HashMap::default();
for (child, parent) in refined[level].iter().zip(refined[level + 1].iter()) {
let entry = parent_to_children.entry(*parent).or_default();
if !entry.contains(child) {
entry.push(*child);
}
}
for v in parent_to_children.values_mut() {
v.sort_unstable();
}
(0..num_entities)
.map(|e| {
let parent = refined[level + 1][e];
parent_to_children.get(&parent).cloned().unwrap_or_default()
})
.collect()
}
pub trait CandidateProposer {
fn propose(&self, labels: &[usize]) -> Vec<Vec<usize>>;
}
pub trait MoveGuard {
fn accept_move(&self, entity: usize, from: usize, labels: &[usize]) -> bool;
}
pub struct NoGuard;
impl MoveGuard for NoGuard {
#[inline(always)]
fn accept_move(&self, _e: usize, _from: usize, _labels: &[usize]) -> bool {
true
}
}
pub fn intersect_with_siblings_fallback(
siblings: &[usize],
neighbor_groups: &[usize],
current: usize,
) -> Vec<usize> {
if siblings.is_empty() {
return Vec::new();
}
if siblings.len() == 1 {
return siblings.to_vec();
}
let intersect: Vec<usize> = siblings
.iter()
.copied()
.filter(|g| neighbor_groups.binary_search(g).is_ok())
.collect();
if intersect.is_empty() {
return siblings.to_vec();
}
if intersect.contains(¤t) {
intersect
} else {
let mut c = intersect;
c.push(current);
c.sort_unstable();
c
}
}
pub struct RefineContext<'a> {
pub profiles: &'a Profiles,
pub k: usize,
pub params: &'a RefineParams,
pub level_label: &'a str,
}
#[derive(Default, Clone, Copy)]
struct SweepCounts {
moves: usize,
vetoed: usize,
}
struct SizeLimits {
floor: Vec<usize>,
cap: Vec<usize>,
}
fn millionths(x: f32) -> u128 {
(x as f64 * 1e6).round().max(0.0) as u128
}
impl SizeLimits {
fn new(count: &[usize], params: &RefineParams) -> Self {
const UNIT: u128 = 1_000_000;
let occupied = count.iter().filter(|&&c| c > 0).count().max(1) as u128;
let total = count.iter().sum::<usize>() as u128;
let size_floor = millionths(params.size_floor);
let size_cap = millionths(params.size_cap);
let limit = |frac: u128, base: u128| (frac * base).div_ceil(UNIT * occupied) as usize;
let (floor, cap) = count
.iter()
.map(|&n0| {
let n0 = n0 as u128 * occupied;
let floor = if size_floor > 0 {
limit(size_floor, n0.min(total)).max(1)
} else {
0
};
let cap = if size_cap > 0 {
limit(size_cap, n0.max(total))
} else {
usize::MAX
};
(floor, cap)
})
.unzip();
Self { floor, cap }
}
#[inline]
fn can_leave(&self, count: &[usize], from: usize) -> bool {
count[from] > self.floor[from]
}
#[inline]
fn admits(&self, count: &[usize], from: usize, to: usize) -> bool {
self.can_leave(count, from) && count[to] < self.cap[to]
}
}
struct SweepEnv<'a, G> {
candidates: &'a [Vec<usize>],
profiles: &'a Profiles,
limits: &'a SizeLimits,
guard: &'a G,
}
struct SweepScratch {
proposals: Vec<usize>,
gains: Vec<f32>,
movers: Vec<u32>,
deferred: Vec<u32>,
changed: Vec<bool>,
log_probs: Vec<f64>,
waiting: HashMap<(usize, usize), VecDeque<u32>>,
}
impl SweepScratch {
fn new(num_entities: usize, k: usize, parallel: bool) -> Self {
let per_entity = if parallel { num_entities } else { 0 };
Self {
proposals: vec![0; per_entity],
gains: vec![0.0; per_entity],
movers: Vec::with_capacity(per_entity),
deferred: Vec::new(),
changed: vec![false; k],
log_probs: vec![f64::NEG_INFINITY; k],
waiting: HashMap::default(),
}
}
}
struct Mover<'a, G> {
env: &'a SweepEnv<'a, G>,
stats: &'a mut DcPoissonStats,
changed: &'a mut [bool],
log_probs: &'a mut [f64],
waiting: &'a mut HashMap<(usize, usize), VecDeque<u32>>,
counts: SweepCounts,
}
impl<'a, G: MoveGuard> Mover<'a, G> {
fn new(
env: &'a SweepEnv<'a, G>,
stats: &'a mut DcPoissonStats,
changed: &'a mut [bool],
log_probs: &'a mut [f64],
waiting: &'a mut HashMap<(usize, usize), VecDeque<u32>>,
) -> Self {
waiting.clear();
Self {
env,
stats,
changed,
log_probs,
waiting,
counts: SweepCounts::default(),
}
}
fn gains(&mut self, e: usize, from: usize, to: usize) -> bool {
compute_log_probs_restricted(
e,
self.stats,
self.env.profiles,
&[from, to],
self.log_probs,
);
self.log_probs[to] > self.log_probs[from]
}
fn accepted(&mut self, e: usize, from: usize) -> bool {
let yes = self.env.guard.accept_move(e, from, &self.stats.membership);
if !yes {
self.counts.vetoed += 1;
}
yes
}
fn apply(&mut self, e: usize, from: usize, to: usize) {
self.stats.delta_move(e, from, to, self.env.profiles);
self.changed[from] = true;
self.changed[to] = true;
}
fn settle(&mut self, e: usize, to: usize, recheck: bool) -> bool {
let from = self.stats.membership[e];
if from == to || (recheck && !self.gains(e, from, to)) {
return false;
}
if self.env.limits.admits(&self.stats.count, from, to) {
if self.accepted(e, from) {
self.apply(e, from, to);
self.counts.moves += 1;
}
return false;
}
while let Some(f) = self
.waiting
.get_mut(&(to, from))
.and_then(VecDeque::pop_front)
{
let f = f as usize;
if self.stats.membership[f] != to {
continue;
}
if !self.accepted(e, from) {
self.waiting
.entry((to, from))
.or_default()
.push_front(f as u32);
return false;
}
self.stats.delta_move(e, from, to, self.env.profiles);
if self.gains(f, to, from) && self.accepted(f, to) {
self.apply(f, to, from);
self.counts.moves += 2;
return false;
}
self.stats.delta_move(e, to, from, self.env.profiles);
}
self.waiting
.entry((from, to))
.or_default()
.push_back(e as u32);
true
}
}
fn apply_proposals<G: MoveGuard>(
env: &SweepEnv<G>,
scratch: &mut SweepScratch,
stats: &mut DcPoissonStats,
) -> SweepCounts {
let SweepScratch {
proposals,
gains,
movers,
deferred,
changed,
log_probs,
waiting,
} = scratch;
movers.clear();
movers.extend(
proposals
.iter()
.zip(&stats.membership)
.enumerate()
.filter(|(_, (new, old))| new != old)
.map(|(e, _)| e as u32),
);
movers.sort_unstable_by(|&a, &b| {
gains[b as usize]
.total_cmp(&gains[a as usize])
.then(a.cmp(&b))
});
changed.iter_mut().for_each(|c| *c = false);
deferred.clear();
let mut mover = Mover::new(env, stats, changed, log_probs, waiting);
for &e in movers.iter() {
let (e, to) = (e as usize, proposals[e as usize]);
let from = mover.stats.membership[e];
let recheck = mover.changed[from] || mover.changed[to];
if mover.settle(e, to, recheck) {
deferred.push(e as u32);
}
}
if mover.counts.moves > 0 {
mover.waiting.clear();
for &e in deferred.iter() {
mover.settle(e as usize, proposals[e as usize], true);
}
}
mover.counts
}
fn sweep_sequential<G, F, I>(
env: &SweepEnv<G>,
stats: &mut DcPoissonStats,
scratch: &mut SweepScratch,
order: I,
mut pick: F,
) -> SweepCounts
where
G: MoveGuard,
F: FnMut(&[f64], &[usize], usize) -> usize,
I: IntoIterator<Item = usize>,
{
let mut mover = Mover::new(
env,
stats,
&mut scratch.changed,
&mut scratch.log_probs,
&mut scratch.waiting,
);
for e in order {
let cand = &env.candidates[e];
if cand.len() < 2 {
continue;
}
let old = mover.stats.membership[e];
compute_log_probs_restricted(e, mover.stats, env.profiles, cand, mover.log_probs);
let new = pick(mover.log_probs, cand, old);
mover.settle(e, new, false);
}
mover.counts
}
fn sweep_jacobi<G: MoveGuard, F>(
env: &SweepEnv<G>,
stats: &mut DcPoissonStats,
scratch: &mut SweepScratch,
pick: F,
) -> SweepCounts
where
F: Fn(&[f64], &[usize], usize, usize) -> usize + Sync,
{
let SweepEnv {
candidates,
profiles,
..
} = *env;
debug_assert_eq!(scratch.proposals.len(), candidates.len());
{
let stats: &DcPoissonStats = stats;
let k = stats.k;
scratch
.proposals
.par_iter_mut()
.zip(scratch.gains.par_iter_mut())
.enumerate()
.with_min_len(256)
.for_each_init(
|| vec![f64::NEG_INFINITY; k],
|log_probs, (e, (prop, gain))| {
let cand = &candidates[e];
let current = stats.membership[e];
*prop = current;
if cand.len() < 2 {
return;
}
compute_log_probs_restricted(e, stats, profiles, cand, log_probs);
let new = pick(log_probs, cand, e, current);
if new != current {
*prop = new;
*gain = (log_probs[new] - staying_score(log_probs, cand, current)) as f32;
}
},
);
}
apply_proposals(env, scratch, stats)
}
pub fn refine_with_candidates_guarded<G: MoveGuard>(
labels: &mut [usize],
candidates: &[Vec<usize>],
guard: &G,
rng: &mut SmallRng,
ctx: &RefineContext,
) -> usize {
let RefineContext {
profiles,
k,
params,
level_label,
} = *ctx;
let num_entities = labels.len();
let mut stats = DcPoissonStats::from_profiles_with(profiles, k, labels, params);
let limits = SizeLimits::new(&stats.count, params);
let env = SweepEnv {
candidates,
profiles,
limits: &limits,
guard,
};
let mut total_moves = 0usize;
let mut total_vetoed = 0usize;
let mut scratch = SweepScratch::new(num_entities, k, params.parallel);
let mut order: Vec<usize> = if !params.parallel {
(0..num_entities).collect()
} else {
Vec::new()
};
let max_sweeps = (params.num_gibbs + params.num_greedy) as u64;
let prog_bar = legume_numeric::matrix::progress::new_progress_bar(max_sweeps)
.with_message(format!("{level_label} sweeps"));
prog_bar.enable_steady_tick(std::time::Duration::from_millis(100));
let jacobi_base_seed = rng.random::<u64>() | 1;
if params.num_gibbs > 0 {
let mut low_sweeps = 0usize;
for sweep in 0..params.num_gibbs {
let sc = if params.parallel {
let sweep_seed = jacobi_base_seed.wrapping_mul(sweep as u64 + 1);
sweep_jacobi(
&env,
&mut stats,
&mut scratch,
|log_probs, cand, e, current| {
let vertex_seed = sweep_seed ^ (e as u64).wrapping_mul(2654435761);
let mut rng = SmallRng::seed_from_u64(vertex_seed);
sample_improving_log(log_probs, cand, current, &mut rng)
},
)
} else {
order.shuffle(rng);
sweep_sequential(
&env,
&mut stats,
&mut scratch,
order.iter().copied(),
|log_probs, cand, current| sample_improving_log(log_probs, cand, current, rng),
)
};
total_moves += sc.moves;
total_vetoed += sc.vetoed;
prog_bar.inc(1);
if params.gibbs_stagnation > 0.0 {
if (sc.moves as f64) < params.gibbs_stagnation * (num_entities as f64) {
low_sweeps += 1;
if low_sweeps >= 3 {
break;
}
} else {
low_sweeps = 0;
}
}
}
}
for _sweep in 0..params.num_greedy {
let sc = if params.parallel {
sweep_jacobi(
&env,
&mut stats,
&mut scratch,
|log_probs, cand, _e, current| argmax_log_staying(log_probs, cand, current),
)
} else {
sweep_sequential(
&env,
&mut stats,
&mut scratch,
0..num_entities,
argmax_log_staying,
)
};
total_moves += sc.moves;
total_vetoed += sc.vetoed;
prog_bar.inc(1);
if sc.moves == 0 {
break;
}
}
prog_bar.finish_and_clear();
if total_vetoed > 0 {
log::debug!(
"{}: {} moves, {} vetoed by MoveGuard",
level_label,
total_moves,
total_vetoed
);
}
labels.copy_from_slice(&stats.membership);
total_moves
}
pub fn refine_with_candidates(
labels: &mut [usize],
candidates: &[Vec<usize>],
rng: &mut SmallRng,
ctx: &RefineContext,
) -> usize {
refine_with_candidates_guarded(labels, candidates, &NoGuard, rng, ctx)
}
pub fn refine_with_proposer_guarded<P: CandidateProposer, G: MoveGuard>(
labels: &mut [usize],
proposer: &P,
guard: &G,
rng: &mut SmallRng,
ctx: &RefineContext,
) -> usize {
let candidates = proposer.propose(labels);
let moves = refine_with_candidates_guarded(labels, &candidates, guard, rng, ctx);
info!(" {}: {} DC-Poisson moves", ctx.level_label, moves);
moves
}
pub fn refine_with_proposer<P: CandidateProposer>(
labels: &mut [usize],
proposer: &P,
rng: &mut SmallRng,
ctx: &RefineContext,
) -> usize {
let candidates = proposer.propose(labels);
let moves = refine_with_candidates(labels, &candidates, rng, ctx);
info!(" {}: {} DC-Poisson moves", ctx.level_label, moves);
moves
}
#[cfg(test)]
#[path = "dc_poisson_tests.rs"]
mod tests;