face-core 0.1.0

//! §5.2 Quantiles clustering: equal-population buckets over the
//! axis field's numeric value.
//!
//! - Same record-extraction policy as Bands: missing-path drops
//!   silently, present-but-non-numeric drops with a [`SkipReport`].
//! - Sorts the resolved values ascending and partitions them into
//!   `count` near-equal-population buckets using the **nearest-rank**
//!   convention: bucket `i` (zero-indexed) covers sorted positions
//!   `[floor(i * N / count), floor((i+1) * N / count))`. This is the
//!   simplest deterministic split, and avoids interpolation between
//!   adjacent values when `N` is not divisible by `count`.
//! - Empty buckets are dropped (only possible when `count > N`).
//! - Sort: numeric descending by bucket lower bound (matches Bands —
//!   highest-scoring ranges appear first).
//! - Cluster `value` is `{"min": lo, "max": hi}` where `lo` and `hi`
//!   are the actual lowest and highest record values inside the
//!   bucket — *not* a synthetic equal-width bound, since quantile
//!   bucket edges are determined by the data.
//! - Label format: same en-dash convention as Bands.

use serde_json::json;

use crate::{
    Cluster, ClusterId, Diagnostics, FaceError, Record, Strategy,
    cluster_tree::{
        AxisPlan, build_tree_impl, extend_id,
        util::{resolve_numeric, score_range},
    },
};

/// Cluster `items` by equal-population quantile bucket on the resolved
/// numeric value at `plan.axis.field`, recursing per cluster into
/// `plan.within`.
pub(super) fn cluster<D: Diagnostics + ?Sized>(
    plan: &AxisPlan,
    items: Vec<Record>,
    parent: &ClusterId,
    diag: &mut D,
) -> Result<Vec<Cluster>, FaceError> {
    let Strategy::Quantiles { count } = &plan.axis.strategy else {
        unreachable!("quantiles clusterer is only called for Strategy::Quantiles")
    };
    let count = (*count).max(1) as usize;

    let mut resolved = resolve_numeric(items, &plan.axis.field, diag);
    if resolved.is_empty() {
        return Ok(Vec::new());
    }

    // Sort ascending by numeric value. NaN was filtered upstream by
    // `is_finite`, so `total_cmp` is safe and deterministic.
    resolved.sort_by(|a, b| a.0.total_cmp(&b.0));

    let n = resolved.len();
    if n == 1 || count == 1 {
        // Single bucket spanning everything.
        let lo = resolved[0].0;
        let hi = resolved[n - 1].0;
        let group: Vec<Record> = resolved.into_iter().map(|(_, r)| r).collect();
        return Ok(vec![build_cluster(
            plan,
            parent,
            format_label(lo, hi, &[lo, hi]),
            lo,
            hi,
            group,
            diag,
        )?]);
    }

    // Compute boundaries: bucket i covers sorted positions
    // [floor(i * N / count), floor((i+1) * N / count)). The final
    // boundary lands at exactly `n` so the last bucket consumes any
    // remainder records.
    let bucket_starts: Vec<usize> = (0..=count).map(|i| i * n / count).collect();

    // Distribute the sorted records into buckets in a single pass.
    // `bucket_idx` advances when the current sorted index crosses the
    // next boundary; the last bucket has no upper boundary to cross,
    // so we never advance past `count - 1`.
    let mut buckets: Vec<Vec<(f64, Record)>> = (0..count).map(|_| Vec::new()).collect();
    let mut bucket_idx = 0usize;
    for (i, item) in resolved.into_iter().enumerate() {
        while bucket_idx + 1 < count && i >= bucket_starts[bucket_idx + 1] {
            bucket_idx += 1;
        }
        buckets[bucket_idx].push(item);
    }

    // Drop empty buckets and merge consecutive non-empty buckets that
    // span the same `[lo, hi]` range. Heavy duplication (more buckets
    // than distinct values) would otherwise emit clusters with
    // colliding labels and ids; merging preserves the equal-population
    // *intent* while honoring the §5.2 "empty buckets dropped (rare
    // unless heavy duplication)" rule.
    let mut merged: Vec<Vec<(f64, Record)>> = Vec::with_capacity(count);
    for bucket in buckets {
        if bucket.is_empty() {
            continue;
        }
        let lo = bucket.first().expect("non-empty checked").0;
        let hi = bucket.last().expect("non-empty checked").0;
        if let Some(prev) = merged.last_mut() {
            let prev_lo = prev.first().expect("non-empty by construction").0;
            let prev_hi = prev.last().expect("non-empty by construction").0;
            // Merge when both buckets collapse to the same single value
            // (the only case where labels and ids would collide). When
            // spans differ even if endpoints touch, we keep them
            // separate — `lo` and `hi` together still distinguish them.
            if prev_lo == prev_hi && lo == hi && prev_lo == lo {
                prev.extend(bucket);
                continue;
            }
        }
        merged.push(bucket);
    }

    let mut out = Vec::with_capacity(merged.len());
    for bucket in merged {
        let lo = bucket.first().expect("non-empty by construction").0;
        let hi = bucket.last().expect("non-empty by construction").0;
        let label = format_label(lo, hi, &[lo, hi]);
        let group: Vec<Record> = bucket.into_iter().map(|(_, r)| r).collect();
        out.push(build_cluster(plan, parent, label, lo, hi, group, diag)?);
    }
    out.reverse();

    Ok(out)
}

/// Build one cluster from a (label, bounds, records) triple, recursing
/// into `plan.within` if any.
fn build_cluster<D: Diagnostics + ?Sized>(
    plan: &AxisPlan,
    parent: &ClusterId,
    label: String,
    lo: f64,
    hi: f64,
    group: Vec<Record>,
    diag: &mut D,
) -> Result<Cluster, FaceError> {
    let id = extend_id(parent, &plan.axis.field, &label);
    let total = group.len() as u64;
    let (score_min, score_max) = score_range(&group);

    let children = if plan.within.is_empty() {
        Vec::new()
    } else {
        let mut acc: Vec<Cluster> = Vec::new();
        for child_plan in &plan.within {
            let child_clusters = build_tree_impl(child_plan, group.clone(), &id, diag)?;
            acc.extend(child_clusters);
        }
        acc
    };

    Ok(Cluster {
        id,
        label,
        axis: plan.axis.field.clone(),
        value: json!({"min": lo, "max": hi}),
        total,
        score_min,
        score_max,
        clusters: children,
    })
}

/// Render `lo–hi` with the en-dash (U+2013). If every bound is
/// integer-valued in floating point, render without decimals;
/// otherwise use two-decimal precision. Mirrors `bands::format_label`.
fn format_label(lo: f64, hi: f64, bounds: &[f64]) -> String {
    if bounds.iter().all(|b| b.fract() == 0.0) {
        format!("{}\u{2013}{}", lo as i64, hi as i64)
    } else {
        format!("{lo:.2}\u{2013}{hi:.2}")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Axis, NullDiagnostics, Strategy};
    use serde_json::json;

    fn axis(field: &str, count: u8) -> Axis {
        Axis {
            field: field.into(),
            strategy: Strategy::Quantiles { count },
            auto: false,
        }
    }

    #[test]
    fn equal_population_partitioning() {
        // 8 records, count=4 → 2 records per bucket.
        let plan = AxisPlan::leaf(axis("v", 4));
        let items: Vec<serde_json::Value> = (0..8).map(|i| json!({"v": i as f64})).collect();
        let mut diag = NullDiagnostics;
        let clusters = super::cluster(
            &plan,
            Record::from_items(items, None),
            &ClusterId::default(),
            &mut diag,
        )
        .unwrap();
        assert_eq!(clusters.len(), 4);
        for c in &clusters {
            assert_eq!(c.total, 2, "expected equal-population buckets, got {c:?}");
        }
        // Bucket bounds reflect actual records inside each bucket,
        // ordered high-to-low.
        assert_eq!(clusters[0].label, "6\u{2013}7");
        assert_eq!(clusters[3].label, "0\u{2013}1");
    }

    #[test]
    fn handles_duplicates() {
        // Heavy duplicates: buckets that span the same single value
        // collapse so labels and cluster ids stay unique. Two distinct
        // values → at most two emitted clusters, regardless of `count`.
        let plan = AxisPlan::leaf(axis("v", 4));
        let items = vec![
            json!({"v": 1.0}),
            json!({"v": 1.0}),
            json!({"v": 1.0}),
            json!({"v": 1.0}),
            json!({"v": 5.0}),
            json!({"v": 5.0}),
            json!({"v": 5.0}),
            json!({"v": 5.0}),
        ];
        let mut diag = NullDiagnostics;
        let clusters = super::cluster(
            &plan,
            Record::from_items(items, None),
            &ClusterId::default(),
            &mut diag,
        )
        .unwrap();
        assert_eq!(clusters.len(), 2);
        // Total records preserved across the merge.
        let total: u64 = clusters.iter().map(|c| c.total).sum();
        assert_eq!(total, 8);
        assert_eq!(clusters[0].value, json!({"min": 5.0, "max": 5.0}));
        assert_eq!(clusters[1].value, json!({"min": 1.0, "max": 1.0}));
    }

    #[test]
    fn fewer_records_than_buckets_drops_empties() {
        // 3 records into 5 buckets — 2 buckets are empty and dropped.
        let plan = AxisPlan::leaf(axis("v", 5));
        let items = vec![json!({"v": 0.0}), json!({"v": 1.0}), json!({"v": 2.0})];
        let mut diag = NullDiagnostics;
        let clusters = super::cluster(
            &plan,
            Record::from_items(items, None),
            &ClusterId::default(),
            &mut diag,
        )
        .unwrap();
        // 3 non-empty buckets, one record each.
        let total: u64 = clusters.iter().map(|c| c.total).sum();
        assert_eq!(total, 3);
        assert!(
            clusters.iter().all(|c| c.total >= 1),
            "no empty buckets should be emitted"
        );
    }

    #[test]
    fn empty_records_returns_empty() {
        let plan = AxisPlan::leaf(axis("v", 4));
        let items: Vec<serde_json::Value> = vec![]; // truly empty input
        let mut diag = NullDiagnostics;
        let clusters = super::cluster(
            &plan,
            Record::from_items(items, None),
            &ClusterId::default(),
            &mut diag,
        )
        .unwrap();
        assert!(clusters.is_empty());
    }
}