facett-flow/src/scene.rs
use serde::{Deserialize, Serialize};
use std::{
collections::{HashMap, HashSet},
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Node {
pub id: String,
pub label: String,
pub position: [f32; 2],
pub group: String,
pub kind: u32,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Edge {
pub from: String,
pub to: String,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Graph {
pub nodes: Vec<Node>,
pub edges: Vec<Edge>,
}
impl Graph {
pub fn validate(&self) -> Result<(), String> {
if self.nodes.len() > 1_000_000 || self.edges.len() > 4_000_000 {
return Err("Prototype budget exceeded".into());
}
let mut ids = HashSet::new();
for n in &self.nodes {
if n.id.trim().is_empty()
|| !ids.insert(n.id.as_str())
|| n.group.trim().is_empty()
|| n.kind > 3
|| n.position.iter().any(|v| !v.is_finite() || v.abs() > 1.0e7)
{
return Err(
"Nodes need unique IDs, a group, finite bounded positions and kind 0..3 (3 = hub)".into(),
);
}
}
if self
.edges
.iter()
.any(|e| !ids.contains(e.from.as_str()) || !ids.contains(e.to.as_str()))
{
return Err("Edge endpoint absent from graph".into());
}
Ok(())
}
/// Central hub with input clouds on the left/top and output clouds right/bottom.
/// Exactly `count` original nodes; group cards are an explicit overview of them.
pub fn hub_demo(count: usize) -> Self {
if count == 0 {
return Self {
nodes: Vec::new(),
edges: Vec::new(),
};
}
let mut nodes = Vec::with_capacity(count);
let mut edges = Vec::with_capacity(count.saturating_sub(1));
nodes.push(Node {
id: "flow-hub".into(),
label: "Hub".into(),
position: [0.0, 0.0],
group: "Central hub".into(),
kind: 3,
});
let names = [
["Files", "Tables", "Events", "APIs", "Code", "Documents"],
["Grid", "Graph", "Map", "Plot", "Docview", "Applications"],
[
"Signals",
"Changes",
"Streams",
"Records",
"Relations",
"Sources",
],
[
"Exports",
"Dashboards",
"Search",
"Models",
"Reports",
"Services",
],
];
let per_group = count.saturating_sub(1).div_ceil(24).max(1);
let columns = (per_group as f64 * 1.4).sqrt().ceil() as usize;
let rows = per_group.div_ceil(columns.max(1)).max(1);
for i in 0..count.saturating_sub(1) {
let side = i % 4;
let band = (i / 4) % 6;
let local = i / 24;
let along = -440.0 + band as f32 * 176.0;
let offset = [
(local % columns.max(1)) as f32 / columns.max(1) as f32 * 140.0 - 70.0,
(local / columns.max(1)) as f32 / rows as f32 * 110.0 - 55.0,
];
let base = match side {
0 => [-850.0, along],
1 => [850.0, along],
2 => [along, -850.0],
_ => [along, 850.0],
};
let incoming = side == 0 || side == 2;
let id = format!("node-{i:07}");
let label = names[side][band];
nodes.push(Node {
id: id.clone(),
label: format!("{label} · item {local:05}"),
position: [base[0] + offset[0], base[1] + offset[1]],
group: format!("{side}:{label}"),
kind: if incoming { 0 } else { 2 },
});
edges.push(if incoming {
Edge {
from: id,
to: "flow-hub".into(),
}
} else {
Edge {
from: "flow-hub".into(),
to: id,
}
});
}
Self { nodes, edges }
}
/// Deterministic five-stage synthetic DAG; exactly `count` nodes, two outgoing
/// connections per node except the last stage. No I/O or claimed real activity.
pub fn demo(count: usize) -> Self {
let mut nodes = Vec::with_capacity(count);
let per_stage = count.div_ceil(5).max(1);
// Keep stage grids proportional as the fixture grows. The 10k layout stays
// unchanged; 100k becomes wider as well as taller instead of a narrow strip.
let columns = ((per_stage as f64 / 5.0).sqrt().ceil() as usize).max(20);
let rows = per_stage.div_ceil(columns).max(1);
let stage_width = (columns as f32 * 12.0 + 322.0).max(550.0);
let names = ["Inputs", "Prepare", "Transform", "Model", "Outputs"];
for i in 0..count {
let stage = i / per_stage;
let slot = i % per_stage;
let col = slot / rows;
let row = slot % rows;
nodes.push(Node {
id: format!("node-{i:07}"),
label: format!("{} · item {slot:04}", names[stage.min(4)]),
position: [
stage as f32 * stage_width + col as f32 * 12.0,
row as f32 * 14.0,
],
group: format!("stage-{stage}-band-{}", row / 10),
kind: if stage == 0 {
0
} else if stage == 4 {
2
} else {
1
},
});
}
let mut edges = Vec::with_capacity(count * 2);
for i in 0..count {
if i + per_stage < count {
for j in [
i + per_stage,
((i % per_stage + 23) % per_stage) + (i / per_stage + 1) * per_stage,
] {
if j < count {
edges.push(Edge {
from: nodes[i].id.clone(),
to: nodes[j].id.clone(),
});
}
}
}
}
Self { nodes, edges }
}
}
#[repr(C)]
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "wgpu", derive(bytemuck::Pod, bytemuck::Zeroable))]
pub(crate) struct Point {
pub pos: [f32; 2],
pub radius: f32,
pub kind: f32,
}
#[repr(C)]
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "wgpu", derive(bytemuck::Pod, bytemuck::Zeroable))]
pub(crate) struct Link {
pub from: u32,
pub to: u32,
pub phase: f32,
pub weight: f32,
}
pub(crate) struct Geometry {
#[cfg_attr(not(feature = "wgpu"), allow(dead_code))]
pub generation: u64,
pub points: Vec<Point>,
pub links: Vec<Link>,
/// Upper bound on |B″(t)| for the cubic curves, in world units.
#[cfg(any(feature = "wgpu", test))]
curve_acceleration: f32,
}
impl Geometry {
fn new(points: Vec<Point>, links: Vec<Link>) -> Self {
#[cfg(any(feature = "wgpu", test))]
let curve_acceleration = links
.iter()
.map(|e| {
let a = egui::Pos2::from(points[e.from as usize].pos);
let b = egui::Pos2::from(points[e.to as usize].pos);
let delta = b - a;
let bend = if delta.y.abs() > delta.x.abs() {
egui::vec2(0.0, delta.y * 0.42)
} else {
egui::vec2(delta.x * 0.42, 0.0)
};
// B″ is linear between ±6(delta - 3*bend).
6.0 * (delta - 3.0 * bend).length()
})
.fold(0.0_f32, f32::max);
Self {
generation: fresh(),
points,
links,
#[cfg(any(feature = "wgpu", test))]
curve_acceleration,
}
}
/// Linear interpolation error ≤ max|B″| / (8*n²). Keep the old 16-segment
/// detail as the ceiling; reduce work only when the error is below half a pixel.
#[cfg(any(feature = "wgpu", test))]
pub(crate) fn curve_segments(&self, pixels_per_world: f32) -> u32 {
(self.curve_acceleration * pixels_per_world / 4.0)
.sqrt()
.ceil()
.clamp(2.0, 16.0) as u32
}
}
static GENERATION: AtomicU64 = AtomicU64::new(1);
fn fresh() -> u64 {
GENERATION.fetch_add(1, Ordering::Relaxed)
}
#[derive(Clone)]
pub(crate) struct Scene {
pub raw: Arc<Geometry>,
pub clusters: Arc<Geometry>,
pub ids: HashMap<String, usize>,
pub bounds: egui::Rect,
pub buckets: HashMap<(i32, i32), Vec<usize>>,
pub cluster_names: Vec<(String, usize)>,
pub hubs: Vec<usize>,
}
impl Scene {
pub fn new(graph: &Graph) -> Self {
let ids: HashMap<_, _> = graph
.nodes
.iter()
.enumerate()
.map(|(i, n)| (n.id.clone(), i))
.collect();
let points = graph
.nodes
.iter()
.map(|n| Point {
pos: n.position,
radius: if n.kind == 3 { 180.0 } else { 2.8 },
kind: n.kind as f32,
})
.collect();
let links = graph
.edges
.iter()
.enumerate()
.map(|(i, e)| Link {
from: ids[&e.from] as u32,
to: ids[&e.to] as u32,
phase: ((i.wrapping_mul(2654435761) & 65535) as f32) / 65536.0,
weight: 1.0,
})
.collect();
let raw = Arc::new(Geometry::new(points, links));
let mut buckets: HashMap<_, Vec<usize>> = HashMap::new();
let mut bounds = egui::Rect::NOTHING;
let mut groups = HashMap::new();
let mut sums: Vec<([f32; 2], usize, f32)> = Vec::new();
let mut cluster_names = Vec::new();
for (i, n) in graph.nodes.iter().enumerate() {
bounds.extend_with(egui::pos2(n.position[0], n.position[1]));
buckets.entry(bucket(n.position)).or_default().push(i);
let k = *groups.entry(n.group.as_str()).or_insert_with(|| {
let k = sums.len();
sums.push(([0.0; 2], 0, n.kind as f32));
cluster_names.push((n.group.clone(), 0));
k
});
sums[k].0[0] += n.position[0];
sums[k].0[1] += n.position[1];
sums[k].1 += 1;
cluster_names[k].1 += 1;
}
if graph.nodes.is_empty() {
bounds = egui::Rect::from_min_max(egui::pos2(-1.0, -1.0), egui::pos2(1.0, 1.0));
}
let points = sums
.iter()
.map(|(p, n, k)| Point {
pos: [p[0] / *n as f32, p[1] / *n as f32],
radius: if *k == 3.0 { 180.0 } else { 12.0 },
kind: *k,
})
.collect();
let mut pairs = std::collections::BTreeMap::new();
for e in &graph.edges {
let a = groups[graph.nodes[ids[&e.from]].group.as_str()] as u32;
let b = groups[graph.nodes[ids[&e.to]].group.as_str()] as u32;
if a != b {
*pairs.entry((a, b)).or_insert(0u32) += 1;
}
}
let links = pairs
.into_iter()
.enumerate()
.map(|(i, ((from, to), n))| Link {
from,
to,
phase: (i as f32 * 0.618034).fract(),
weight: n as f32,
})
.collect();
Self {
raw,
clusters: Arc::new(Geometry::new(points, links)),
ids,
bounds,
buckets,
cluster_names,
hubs: graph
.nodes
.iter()
.enumerate()
.filter(|(_, n)| n.kind == 3)
.map(|(i, _)| i)
.collect(),
}
}
pub fn centre(&self) -> [f32; 2] {
let p = self.bounds.center();
[p.x, p.y]
}
pub fn visible(&self, rect: egui::Rect, cap: usize) -> Vec<usize> {
let a = bucket([rect.min.x, rect.min.y]);
let b = bucket([rect.max.x, rect.max.y]);
let mut out = Vec::new();
// Bounded query: at low zoom no labels/cards are emitted, so this is only
// used for close zoom, picking, and the explicitly bounded CPU fallback.
if i64::from(b.0 - a.0 + 1) * i64::from(b.1 - a.1 + 1) > 200_000 {
return out;
}
for y in a.1..=b.1 {
for x in a.0..=b.0 {
if let Some(indices) = self.buckets.get(&(x, y)) {
for &i in indices {
let p = self.raw.points[i].pos;
if rect.contains(egui::pos2(p[0], p[1])) {
out.push(i);
if out.len() == cap {
return out;
}
}
}
}
}
}
out
}
}
fn bucket(p: [f32; 2]) -> (i32, i32) {
((p[0] / 64.0).floor() as i32, (p[1] / 64.0).floor() as i32)
}
#[cfg(test)]
mod lod_tests {
use super::*;
#[test]
fn curve_detail_tracks_zoom_and_bounds_screen_space_error() {
let a = egui::vec2(0.0, 0.0);
let b = egui::vec2(100.0, 60.0);
let c = egui::vec2(42.0, 0.0);
let d = b - c;
let geom = Geometry::new(
vec![
Point {
pos: [a.x, a.y],
radius: 1.0,
kind: 0.0,
},
Point {
pos: [b.x, b.y],
radius: 1.0,
kind: 1.0,
},
],
vec![Link {
from: 0,
to: 1,
phase: 0.0,
weight: 1.0,
}],
);
let curve = |t: f32| {
let s = 1.0 - t;
a * s.powi(3) + c * (3.0 * s * s * t) + d * (3.0 * s * t * t) + b * t.powi(3)
};
let mut previous = 0;
for scale in [0.01, 0.1, 0.5, 1.0, 2.0, 4.0, 16.0] {
let n = geom.curve_segments(scale);
assert!(n >= previous && (2..=16).contains(&n));
previous = n;
if n < 16 {
for segment in 0..n {
let lo = segment as f32 / n as f32;
let hi = (segment + 1) as f32 / n as f32;
for step in 0..=10 {
let t = step as f32 / 10.0;
let line = curve(lo) * (1.0 - t) + curve(hi) * t;
assert!((line - curve(lo + (hi - lo) * t)).length() * scale <= 0.501);
}
}
}
}
assert!(
geom.curve_segments(0.1) < 16,
"overview must submit fewer vertices"
);
assert_eq!(
geom.curve_segments(100.0),
16,
"detail preserves the previous ceiling"
);
}
}
#[cfg(any(feature = "host", feature = "compact-state"))]
pub(crate) fn hash_bytes(hash: &mut u64, bytes: &[u8]) {
for b in bytes {
*hash ^= u64::from(*b);
*hash = hash.wrapping_mul(0x100000001b3);
}
}
#[cfg(any(feature = "host", feature = "compact-state"))]
pub(crate) fn digest(graph: &Graph, key: u64) -> u64 {
let mut hash = 0xcbf29ce484222325_u64;
hash_bytes(&mut hash, &key.to_le_bytes());
let string = |hash: &mut u64, s: &str| {
hash_bytes(hash, &(s.len() as u64).to_le_bytes());
hash_bytes(hash, s.as_bytes());
};
hash_bytes(&mut hash, &(graph.nodes.len() as u64).to_le_bytes());
for n in &graph.nodes {
string(&mut hash, &n.id);
string(&mut hash, &n.label);
string(&mut hash, &n.group);
for p in n.position {
hash_bytes(&mut hash, &p.to_bits().to_le_bytes());
}
hash_bytes(&mut hash, &n.kind.to_le_bytes());
}
hash_bytes(&mut hash, &(graph.edges.len() as u64).to_le_bytes());
for e in &graph.edges {
string(&mut hash, &e.from);
string(&mut hash, &e.to);
}
hash
}
#[cfg(feature = "layered")]
#[derive(Clone, Copy, Debug)]
pub struct LayeredOpts {
pub align: facett_core::layered::Align,
pub rows: facett_core::layered::RowOrder,
pub pitch: [f32; 2],
}
#[cfg(feature = "layered")]
impl Default for LayeredOpts {
fn default() -> Self {
Self {
align: facett_core::layered::Align::Sources,
rows: facett_core::layered::RowOrder::Input,
pitch: [240.0, 120.0],
}
}
}
#[cfg(feature = "layered")]
impl Graph {
pub fn layered(&mut self, opts: LayeredOpts) -> Result<(), String> {
self.validate()?;
if opts.pitch.iter().any(|p| !p.is_finite() || *p <= 0.0) {
return Err("Layout pitch must be finite and positive".into());
}
let ids: HashMap<_, _> = self
.nodes
.iter()
.enumerate()
.map(|(i, n)| (n.id.as_str(), i as u32))
.collect();
let edges: Vec<_> = self
.edges
.iter()
.map(|e| (ids[e.from.as_str()], ids[e.to.as_str()]))
.collect();
let layout = facett_core::layered::layered(
self.nodes.len(),
&edges,
facett_core::layered::LayeredOpts {
align: opts.align,
rows: opts.rows,
},
);
let positions = facett_core::layered::place(&layout, opts.pitch, [0.0, 0.0]);
if positions
.iter()
.flatten()
.any(|v| !v.is_finite() || v.abs() > 1.0e7)
{
return Err("Layout exceeds graph coordinate budget".into());
}
for (n, p) in self.nodes.iter_mut().zip(positions) {
n.position = p;
}
Ok(())
}
}
#[cfg(test)]
mod plain_witness {
use super::*;
#[test]
fn graph_layout_and_serde_remain_plain() {
assert_eq!(std::mem::size_of::<Point>(), 16);
assert_eq!(std::mem::size_of::<Link>(), 16);
let g = Graph::hub_demo(1);
let value = serde_json::to_value(&g.nodes[0]).unwrap();
let keys: std::collections::BTreeSet<_> = value
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
assert_eq!(
keys,
["id", "label", "position", "group", "kind"]
.into_iter()
.collect()
);
}
}