1pub mod nodes;
2pub mod osc;
3pub mod offline;
4pub mod freeze;
5pub mod jit;
6#[cfg(test)]
7mod tests {
8 mod null_test;
9}
10
11use dirtydata_core::ir::{Graph, EdgeKind};
12use dirtydata_core::types::{StableId, NodeKind, PortDirection};
13use dirtydata_core::graph_utils::topological_sort;
14use crate::nodes::*;
15use crate::nodes::legacy::{
16 MidiEvent, EnvelopeNode, AutomationNode, SequencerNode, WavefolderNode,
17 AddNode, MultiplyNode, ClipNode, TriggerNode, DelayNode,
18 LorenzNode, MackeyGlassNode, GrayScottNode, SlewLimiterNode, SampleHoldNode,
19 ClockNode, ProbabilityGateNode, ReverbNode, SpringReverbNode, GranularNode,
20 LogicNode, SpectralFreezeNode, FFTConvolveNode, ZdfLadderNode, SvfNode,
21 DiodeClipperNode, BbdDelayNode, KarplusStrongNode, ModalResonatorNode,
22 ChuaCircuitNode, TapeMachineNode, MatrixMixerNode, EuclideanSequencerNode,
23};
24use crate::osc::OscHandler;
25
26use std::collections::HashMap;
27use std::sync::atomic::{AtomicBool, Ordering};
28use std::sync::Arc;
29use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
30use crossbeam_channel::{Sender, Receiver};
31
32pub struct ParameterUpdate {
33 pub node_id: StableId,
34 pub param: String,
35 pub value: f32,
36 pub provenance: Vec<String>,
37}
38
39pub use offline::OfflineRenderer;
40
41pub enum EngineCommand {
42 UpdateParameter(ParameterUpdate),
43 ReplaceGraph(Graph, Option<jit::JitProgram>),
44}
45
46#[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize)]
47pub struct SignalMetrics {
48 pub rms: f32,
49 pub peak: f32,
50 pub dc_offset: f32,
51 pub dominant_freq: f32,
52 pub activity_score: f32,
53 pub saturation: f32,
54}
55
56#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
57pub struct DiagnosticRecord {
58 pub message: String,
59 pub severity: DiagnosticSeverity,
60 pub timestamp: u64,
61}
62
63#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
64pub enum DiagnosticSeverity {
65 Info,
66 Warning,
67 Error,
68}
69
70pub struct SharedState {
71 node_metrics: Arc<dashmap::DashMap<StableId, SignalMetrics>>,
72 scope_buffer: Arc<crossbeam_queue::ArrayQueue<f32>>,
73 probe_buffers: Arc<dashmap::DashMap<StableId, Arc<crossbeam_queue::ArrayQueue<f32>>>>,
74 convergence_info: Arc<dashmap::DashMap<StableId, usize>>,
75 circuit_instability: Arc<dashmap::DashMap<StableId, f32>>,
76 parameter_provenance: Arc<dashmap::DashMap<StableId, HashMap<String, Vec<String>>>>,
77 node_diagnostics: Arc<dashmap::DashMap<StableId, DiagnosticRecord>>,
78 engine_logs: Arc<crossbeam_queue::ArrayQueue<String>>,
79}
80
81impl SharedState {
82 pub fn new() -> Self {
83 Self {
84 node_metrics: Arc::new(dashmap::DashMap::new()),
85 scope_buffer: Arc::new(crossbeam_queue::ArrayQueue::new(4096)),
86 probe_buffers: Arc::new(dashmap::DashMap::new()),
87 convergence_info: Arc::new(dashmap::DashMap::new()),
88 circuit_instability: Arc::new(dashmap::DashMap::new()),
89 parameter_provenance: Arc::new(dashmap::DashMap::new()),
90 node_diagnostics: Arc::new(dashmap::DashMap::new()),
91 engine_logs: Arc::new(crossbeam_queue::ArrayQueue::new(100)),
92 }
93 }
94
95 pub fn log(&self, msg: impl Into<String>) {
96 let _ = self.engine_logs.push(msg.into());
97 }
98
99 pub fn get_node_metrics(&self, node_id: &StableId) -> Option<SignalMetrics> {
100 self.node_metrics.get(node_id).map(|m| *m)
101 }
102
103 pub fn get_diagnostic(&self, node_id: &StableId) -> Option<DiagnosticRecord> {
104 self.node_diagnostics.get(node_id).map(|d: dashmap::mapref::one::Ref<'_, StableId, DiagnosticRecord>| d.clone())
105 }
106
107 pub fn scope_buffer(&self) -> Arc<crossbeam_queue::ArrayQueue<f32>> {
108 self.scope_buffer.clone()
109 }
110
111 pub fn get_circuit_instability(&self, node_id: &StableId) -> Option<f32> {
112 self.circuit_instability.get(node_id).map(|i| *i)
113 }
114}
115
116pub struct ModulationMapping {
117 pub source_node_id: StableId,
118 pub source_port_idx: usize,
119 pub target_node_idx: usize,
120 pub target_param: String,
121 pub amount: f32,
122}
123
124pub struct DspRunner {
125 nodes: Vec<(StableId, Box<dyn base::DspNode>)>,
126 node_outputs: HashMap<StableId, Vec<[f32; 2]>>,
127 graph: Graph,
128 feedback_latches: Vec<[f32; 2]>,
129 feedback_reads: Vec<Vec<(usize, usize)>>,
130 feedback_writes: Vec<Vec<(usize, usize)>>,
131 modulation_mappings: Vec<ModulationMapping>,
132 node_saturation: HashMap<StableId, f32>,
133 jit_program: Option<jit::JitProgram>,
134 parameter_provenance: HashMap<StableId, HashMap<String, Vec<String>>>,
135}
136
137impl DspRunner {
138 #[tracing::instrument(skip(graph, midi_rx))]
139 pub fn new(graph: Graph, midi_rx: Option<Receiver<MidiEvent>>, sample_rate: f32) -> Self {
140 tracing::debug!("Creating DspRunner with {} nodes", graph.nodes.len());
141 let (sorted_ids, _) = topological_sort(&graph);
142 let mut nodes: Vec<(StableId, Box<dyn base::DspNode>)> = Vec::new();
143 let mut node_outputs = HashMap::new();
144
145 for &id in &sorted_ids {
146 if let Some(node) = graph.nodes.get(&id) {
147 let dsp_node: Box<dyn base::DspNode> = match &node.kind {
148 NodeKind::Foreign(plugin_name) => {
149 Box::new(legacy::ForeignNode::new(plugin_name.clone(), 256))
150 }
151 _ => {
152 let name = node.config.get("name").and_then(|v| v.as_string());
153 match name.map(|s| s.as_str()).unwrap_or("Unknown") {
154 "Oscillator" | "Sine" => Box::new(OscillatorNode::new()),
155 "Noise" => Box::new(NoiseNode::new(format!("{}", id).as_bytes().len() as u64)),
156 "Gain" => Box::new(GainNode::new()),
157 "Add" => Box::new(AddNode::new()),
158 "Multiply" => Box::new(MultiplyNode::new()),
159 "Clip" => Box::new(ClipNode::new()),
160 "Filter" | "Biquad" => Box::new(BiquadFilterNode::new()),
161 "Compressor" | "Dynamics" => Box::new(CompressorNode::new()),
162 "Delay" => Box::new(DelayNode::new(sample_rate as usize)),
163 "Sampler" | "AssetReader" => {
164 let path_val = node.config.get("path").and_then(|v| v.as_string());
165 let data = if let Some(path) = path_val {
166 match hound::WavReader::open(path) {
167 Ok(mut reader) => {
168 let samples: Vec<f32> = reader.samples::<f32>().map(|s: Result<f32, hound::Error>| s.unwrap_or(0.0)).collect::<Vec<f32>>();
169 Arc::new(samples)
170 }
171 Err(e) => {
172 tracing::error!("Failed to load asset {}: {}", path, e);
173 Arc::new(vec![])
174 }
175 }
176 } else {
177 Arc::new(vec![])
178 };
179 Box::new(AssetReaderNode::new(data))
180 }
181 "Trigger" => Box::new(TriggerNode::new()),
182 "Envelope" | "ADSR" => Box::new(EnvelopeNode::new()),
183 "Automation" => Box::new(AutomationNode::new()),
184 "MidiIn" => {
185 if let Some(rx) = &midi_rx {
186 Box::new(MidiInNode::new(rx.clone()))
187 } else {
188 Box::new(GainNode::new())
189 }
190 }
191 "Sequencer" => Box::new(SequencerNode::new()),
192 "Wavefolder" => Box::new(WavefolderNode::new()),
193 "Lorenz" => Box::new(LorenzNode::new()),
194 "MackeyGlass" => Box::new(MackeyGlassNode::new(10.0, sample_rate)),
195 "GrayScott" | "ReactionDiffusion" => Box::new(GrayScottNode::new(256)),
196
197 "BitCrush" => Box::new(BitCrushNode::new()),
199 "WaveShaper" => Box::new(WaveShaperNode::new()),
200 "Pll" => Box::new(PllNode::new()),
201
202 "SlewLimiter" | "Slew" => Box::new(SlewLimiterNode::new()),
204 "SampleHold" | "S&H" => Box::new(SampleHoldNode::new()),
205 "Clock" => Box::new(ClockNode::new()),
206 "ProbabilityGate" => Box::new(ProbabilityGateNode::new()),
207 "Reverb" => Box::new(ReverbNode::new(sample_rate)),
208 "SpringReverb" => Box::new(SpringReverbNode::new(sample_rate)),
209 "Granular" => Box::new(GranularNode::new(sample_rate)),
210 "Logic" => Box::new(LogicNode::new()),
211 "SpectralFreeze" => Box::new(SpectralFreezeNode::new(1024)),
212 "FFTConvolve" => Box::new(FFTConvolveNode::new(1024)),
213 "ZdfLadder" | "TB303Ladder" | "Ladder" => Box::new(ZdfLadderNode::new(sample_rate)),
214 "Svf" | "SVFFilter" => Box::new(SvfNode::new(sample_rate)),
215 "DiodeClipper" => Box::new(DiodeClipperNode::new()),
216 "BbdDelay" => Box::new(BbdDelayNode::new(sample_rate)),
217 "KarplusStrong" => Box::new(KarplusStrongNode::new(sample_rate)),
218 "ModalResonator" => Box::new(ModalResonatorNode::new(sample_rate)),
219 "ChuaCircuit" | "Chua" => Box::new(ChuaCircuitNode::new(sample_rate)),
220 "TapeMachine" | "Tape" => Box::new(TapeMachineNode::new(sample_rate)),
221 "MatrixMixer" => Box::new(MatrixMixerNode::new(8, 8)),
222 "Euclidean" | "EuclideanSequencer" => Box::new(EuclideanSequencerNode::new()),
223 _ => Box::new(GainNode::new()),
224 }
225 }
226 };
227 nodes.push((id, dsp_node));
228 let port_count = node.ports.iter().filter(|p| p.direction == PortDirection::Output).count().max(1);
229 node_outputs.insert(id, vec![[0.0, 0.0]; port_count]);
230 }
231 }
232
233 let mut feedback_latches = Vec::new();
234 let mut feedback_reads = vec![Vec::new(); nodes.len()];
235 let mut feedback_writes = vec![Vec::new(); nodes.len()];
236
237 for edge in graph.edges.values() {
238 if edge.kind == EdgeKind::Feedback {
239 let latch_idx = feedback_latches.len();
240 feedback_latches.push([0.0, 0.0]);
241 if let Some(src_idx) = nodes.iter().position(|(id, _)| *id == edge.source.node_id) {
242 feedback_writes[src_idx].push((0, latch_idx));
243 }
244 if let Some(tgt_idx) = nodes.iter().position(|(id, _)| *id == edge.target.node_id) {
245 feedback_reads[tgt_idx].push((0, latch_idx));
246 }
247 }
248 }
249
250 let mut modulation_mappings = Vec::new();
251 for m in graph.modulations.values() {
252 if let Some(target_idx) = nodes.iter().position(|(id, _)| *id == m.target_node) {
253 modulation_mappings.push(ModulationMapping {
254 source_node_id: m.source.node_id,
255 source_port_idx: 0,
256 target_node_idx: target_idx,
257 target_param: m.target_param.clone(),
258 amount: m.amount,
259 });
260 }
261 }
262
263 let mut node_saturation = HashMap::new();
264 for id in graph.nodes.keys() {
265 node_saturation.insert(*id, 0.0);
266 }
267
268 Self { nodes, node_outputs, graph, feedback_latches, feedback_reads, feedback_writes, modulation_mappings, node_saturation, jit_program: None, parameter_provenance: HashMap::new() }
269 }
270
271 pub fn process_sample(&mut self, ctx: &ProcessContext) -> [f32; 2] {
272 if let Some(jit) = &mut self.jit_program {
273 return jit.execute(ctx);
274 }
275
276 for m in &self.modulation_mappings {
277 if let Some(outputs) = self.node_outputs.get(&m.source_node_id) {
278 let val = (outputs[m.source_port_idx][0] + outputs[m.source_port_idx][1]) * 0.5;
279 let (_, node) = &mut self.nodes[m.target_node_idx];
280 node.update_parameter(&m.target_param, val * m.amount);
281 }
282 }
283
284 for (i, (id, node)) in self.nodes.iter_mut().enumerate() {
285 let mut inputs = Vec::new();
286 for edge in self.graph.edges.values() {
287 if edge.kind == EdgeKind::Normal && edge.target.node_id == *id {
288 if let Some(prev_outputs) = self.node_outputs.get(&edge.source.node_id) {
289 let val = prev_outputs[0];
290 inputs.push(val[0]);
291 inputs.push(val[1]);
292 }
293 }
294 }
295
296 for (_, latch_idx) in &self.feedback_reads[i] {
297 let latch = self.feedback_latches[*latch_idx];
298 if inputs.is_empty() {
299 inputs.push(latch[0]);
300 inputs.push(latch[1]);
301 }
302 }
303
304 let outputs = self.node_outputs.get_mut(id).unwrap();
305
306 let ctx = ProcessContext {
307 sample_rate: ctx.sample_rate,
308 global_sample_index: ctx.global_sample_index,
309 crash_flag: ctx.crash_flag,
310 osc_tx: ctx.osc_tx,
311 convergence_info: ctx.convergence_info,
312 node_diagnostics: ctx.node_diagnostics,
313 node_id: Some(*id),
314 };
315 node.process(&inputs, &mut outputs[..], &self.graph.nodes.get(id).unwrap().config, &ctx);
316
317 let mut sat_accum = 0.0;
319 for port_out in outputs.iter_mut() {
320 for sample in port_out.iter_mut() {
321 let s: &mut f32 = sample;
322 let x = *s;
323 if x.abs() > 1.2 {
324 *s = 1.2 * x.signum();
325 } else {
326 *s = x - (x.powi(3) * 0.23);
327 }
328 sat_accum += (x - *s).abs();
329 }
330 }
331 if let Some(entry) = self.node_saturation.get_mut(id) {
332 *entry += sat_accum;
333 }
334
335 for (_, latch_idx) in &self.feedback_writes[i] {
336 let val: [f32; 2] = outputs[0];
337 self.feedback_latches[*latch_idx] = val;
338 }
339 }
340
341 let mut final_out = [0.0, 0.0];
342 for (id, _) in &self.nodes {
343 if let Some(node) = self.graph.nodes.get(id) {
344 if node.kind == NodeKind::Sink {
345 let out = self.node_outputs.get(id).unwrap()[0];
346 final_out[0] += out[0];
347 final_out[1] += out[1];
348 }
349 }
350 }
351 final_out
352 }
353
354 pub fn get_graph(&self) -> &Graph {
355 &self.graph
356 }
357
358 pub fn nodes_mut(&mut self) -> &mut Vec<(StableId, Box<dyn DspNode>)> {
359 &mut self.nodes
360 }
361
362 pub fn update_parameter(&mut self, node_id: StableId, param: &str, value: f32, provenance: Vec<String>) {
363 self.parameter_provenance.entry(node_id).or_default().insert(param.to_string(), provenance);
364 if let Some((_, node)) = self.nodes.iter_mut().find(|(id, _)| *id == node_id) {
365 node.update_parameter(param, value);
366 }
367 }
368
369 pub fn extract_all_states(&self) -> HashMap<StableId, NodeState> {
370 self.nodes.iter().map(|(id, node)| (*id, node.extract_state())).collect()
371 }
372
373 pub fn inject_all_states(&mut self, states: &HashMap<StableId, NodeState>) {
374 for (id, node) in &mut self.nodes {
375 if let Some(state) = states.get(id) {
376 node.inject_state(state);
377 }
378 }
379 }
380
381 pub fn get_node_outputs(&self, id: &StableId) -> Option<&Vec<[f32; 2]>> {
382 self.node_outputs.get(id)
383 }
384}
385
386impl std::fmt::Debug for DspRunner {
387 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
388 f.debug_struct("DspRunner")
389 .field("nodes", &self.nodes.len())
390 .field("feedback_latches", &self.feedback_latches.len())
391 .field("modulation_mappings", &self.modulation_mappings.len())
392 .finish()
393 }
394}
395
396pub struct AudioEngine {
397 _stream: cpal::Stream,
398 command_tx: Sender<EngineCommand>,
399 shared_state: Arc<SharedState>,
400}
401
402impl AudioEngine {
403 #[tracing::instrument(skip(shared_state, midi_rx))]
404 pub fn new(shared_state: Arc<SharedState>, midi_rx: Receiver<MidiEvent>) -> Self {
405 tracing::info!("Initializing AudioEngine");
406 let host = cpal::default_host();
407 let device = host.default_output_device().expect("no output device available");
408 let config = device.default_output_config().unwrap();
409 let sample_rate = config.sample_rate().0 as f32;
410 let channels = config.channels() as usize;
411
412 let (command_tx, command_rx) = crossbeam_channel::unbounded::<EngineCommand>();
413 let shared_state_for_audio = shared_state.clone();
414 let crash_flag_for_audio = Arc::new(AtomicBool::new(false));
415
416 let osc_handler = OscHandler::new(command_tx.clone());
418 osc_handler.spawn_input_thread("127.0.0.1:8000");
419
420 let (osc_tx, osc_rx) = crossbeam_channel::bounded::<OscMessage>(1024);
421 OscHandler::spawn_output_thread(osc_rx, "127.0.0.1:9001".to_string());
422
423 let mut current_runner: Option<DspRunner> = None;
424 let mut global_sample_index: u64 = 0;
425 let midi_rx_internal = midi_rx.clone();
426 let crash_flag_callback = crash_flag_for_audio.clone();
427
428 let mut metrics_acc: HashMap<StableId, (f32, f32, f32, f32, usize)> = HashMap::new();
430
431 let stream = match config.sample_format() {
432 cpal::SampleFormat::F32 => device.build_output_stream(
433 &config.into(),
434 move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
435 while let Ok(cmd) = command_rx.try_recv() {
436 match cmd {
437 EngineCommand::UpdateParameter(update) => {
438 shared_state_for_audio.parameter_provenance.entry(update.node_id).or_default().insert(update.param.clone(), update.provenance.clone());
439 if let Some(runner) = &mut current_runner {
440 runner.update_parameter(update.node_id, &update.param, update.value, update.provenance);
441 }
442 }
443 EngineCommand::ReplaceGraph(graph, jit_prog) => {
444 shared_state_for_audio.log(format!("Replacing Graph: {} nodes", graph.nodes.len()));
445 let mut new_runner = DspRunner::new(graph, Some(midi_rx_internal.clone()), sample_rate);
446 if let Some(old_runner) = ¤t_runner {
447 let states = old_runner.extract_all_states();
448 new_runner.inject_all_states(&states);
449 }
450 new_runner.jit_program = jit_prog;
451 current_runner = Some(new_runner);
452 shared_state_for_audio.log("Graph Replacement (JIT enabled) OK.");
453 }
454 }
455 }
456
457 let Some(runner) = &mut current_runner else { data.fill(0.0); return; };
458
459 for frame in data.chunks_mut(channels) {
460 let ctx = ProcessContext {
461 sample_rate,
462 global_sample_index,
463 crash_flag: Some(&crash_flag_callback),
464 osc_tx: Some(&osc_tx),
465 convergence_info: Some(&shared_state_for_audio.convergence_info),
466 node_diagnostics: Some(&shared_state_for_audio.node_diagnostics),
467 node_id: None,
468 };
469 let out = runner.process_sample(&ctx);
470 for (node_id, ports) in &runner.node_outputs {
471 let val = (ports[0][0] + ports[0][1]) * 0.5;
472 let acc = metrics_acc.entry(*node_id).or_insert((0.0, 0.0, 0.0, 0.0, 0));
473 acc.0 += val; acc.1 += val * val; acc.2 = acc.2.max(val.abs()); let sat = runner.node_saturation.get_mut(node_id).map(|s| {
478 let v = *s;
479 *s = 0.0;
480 v
481 }).unwrap_or(0.0);
482 acc.3 += sat; acc.4 += 1;
484
485 if acc.4 >= 128 {
486 let (sum, sum_sq, peak, sat_sum, count) = *acc;
487 let f_count = count as f32;
488 let dc = sum / f_count;
489 let rms = (sum_sq / f_count).sqrt();
490
491 shared_state_for_audio.node_metrics.insert(*node_id, SignalMetrics {
492 rms,
493 peak,
494 dc_offset: dc,
495 dominant_freq: 0.0, activity_score: (rms * 10.0).min(1.0),
497 saturation: sat_sum / f_count,
498 });
499 *acc = (0.0, 0.0, 0.0, 0.0, 0);
500 }
501
502 let probes = &shared_state_for_audio.probe_buffers;
503 if let Some(buf_ref) = probes.get(node_id) {
504 let _ = buf_ref.value().push(val);
505 }
506 if val.is_nan() { crash_flag_callback.store(true, Ordering::SeqCst); }
507 }
508 frame[0] = out[0];
509 if channels > 1 { frame[1] = out[1]; }
510 global_sample_index += 1;
511 }
512 },
513 |err| {
514 tracing::error!("an error occurred on stream: {}", err);
515 },
516 None
517 ).unwrap(),
518 _ => panic!("unsupported sample format"),
519 };
520
521 stream.play().unwrap();
522 Self { _stream: stream, command_tx, shared_state }
523 }
524
525 pub fn shared_state(&self) -> Arc<SharedState> {
526 self.shared_state.clone()
527 }
528
529 pub fn update_parameter(&self, node_id: StableId, param: String, value: f32) {
530 let _ = self.command_tx.send(EngineCommand::UpdateParameter(ParameterUpdate {
531 node_id,
532 param,
533 value,
534 provenance: vec!["host".to_string()],
535 }));
536 }
537
538 pub fn replace_graph(&self, graph: Graph, jit_prog: Option<jit::JitProgram>) {
539 let _ = self.command_tx.send(EngineCommand::ReplaceGraph(graph, jit_prog));
540 }
541}