dsp-graph 0.1.6
A Python DSL for defining DSP signal graphs with C++ compilation and optimization
License: MIT
Requires Python: >=3.10
Keywords: dsp, audio, signal-processing, codegen, c++, pydantic, graph

# dsp-graph

A Python DSL for defining DSP signal graphs, compiling them to standalone C++, simulating in Python, and optimizing the result.

Define audio processing graphs using Pydantic models, validate them, compile to C++, simulate in Python with numpy, and serialize to/from JSON. Zero runtime dependencies beyond Pydantic (numpy optional for simulation).

## Install

```bash
pip install dsp-graph

# With simulation support (numpy):
pip install dsp-graph[sim]
```

For development:

```bash
git clone https://github.com/shakfu/dsp-graph.git
cd dsp-graph
make install-dev
```

## Quick Start

```python
from dsp_graph import (
    AudioInput, AudioOutput, BinOp, Graph, History, Param,
    compile_graph, validate_graph,
)

graph = Graph(
    name="onepole",
    inputs=[AudioInput(id="in1")],
    outputs=[AudioOutput(id="out1", source="result")],
    params=[Param(name="coeff", min=0.0, max=0.999, default=0.5)],
    nodes=[
        BinOp(id="inv_coeff", op="sub", a=1.0, b="coeff"),
        BinOp(id="dry", op="mul", a="in1", b="inv_coeff"),
        History(id="prev", init=0.0, input="result"),
        BinOp(id="wet", op="mul", a="prev", b="coeff"),
        BinOp(id="result", op="add", a="dry", b="wet"),
    ],
)

errors = validate_graph(graph)
assert errors == []

code = compile_graph(graph)  # standalone C++ string
print(graph.model_dump_json(indent=2))
```

## Simulation

Run graphs in Python without C++ compilation. Useful for prototyping, unit-testing DSP algorithms, and verifying correctness. Requires numpy (`pip install dsp-graph[sim]`).

```python
import numpy as np
from dsp_graph.simulate import simulate, SimState

# Simulate with audio input
inp = np.ones(100, dtype=np.float32)
result = simulate(graph, inputs={"in1": inp}, params={"coeff": 0.3})

# Output arrays keyed by output ID
output = result.outputs["out1"]  # NDArray[np.float32]

# State persists across calls for streaming
result2 = simulate(graph, inputs={"in1": inp}, state=result.state)

# Generators (no inputs) require explicit n_samples
from dsp_graph import Graph, AudioOutput, SinOsc
gen = Graph(
    name="tone",
    outputs=[AudioOutput(id="out1", source="osc")],
    nodes=[SinOsc(id="osc", freq=440.0)],
    sample_rate=44100.0,
)
result = simulate(gen, n_samples=44100)

# Buffer and peek access via SimState
state = SimState(graph)
state.set_param("coeff", 0.7)
state.set_buffer("wt", np.sin(np.linspace(0, 2*np.pi, 1024)).astype(np.float32))
result = simulate(graph, inputs={"in1": inp}, state=state)
print(state.get_peek("monitor"))
```

The simulator executes a per-sample Python loop that mirrors the C++ codegen exactly: same topo-sorted node order, same deferred History write-backs, same interpolation formulas.

## Multi-Rate Processing

Nodes can be explicitly assigned to run at control rate (once per block) instead of audio rate (per sample). This is useful for parameter smoothing, coefficient computation, and other operations that don't need per-sample updates.

```python
from dsp_graph import Graph, AudioInput, AudioOutput, BinOp, Param, SmoothParam

graph = Graph(
    name="smooth_gain",
    sample_rate=48000.0,
    control_interval=64,          # control block = 64 samples
    control_nodes=["smoother"],   # these nodes run once per block
    inputs=[AudioInput(id="in0")],
    outputs=[AudioOutput(id="out0", source="scaled")],
    params=[Param(name="vol", default=0.5)],
    nodes=[
        SmoothParam(id="smoother", a="vol", coeff=0.99),   # control-rate
        BinOp(id="scaled", op="mul", a="in0", b="smoother"),  # audio-rate
    ],
)
```

When `control_interval > 0`, the generated C++ uses a two-tier loop structure:

```cpp
for (int _cb = 0; _cb < n; _cb += 64) {
    int _block_end = (_cb + 64 < n) ? _cb + 64 : n;
    // Control-rate nodes (once per block)
    float smoother = ...;
    for (int i = _cb; i < _block_end; i++) {
        // Audio-rate nodes (per sample)
        float scaled = in0[i] * smoother;
        out0[i] = scaled;
    }
}
```

Nodes are classified into three tiers:

1. **Invariant** (LICM): pure nodes depending only on params/literals -- hoisted before both loops.
2. **Control-rate**: nodes listed in `control_nodes` -- computed once per control block.
3. **Audio-rate**: everything else -- computed per sample.

The simulator mirrors this behavior: control-rate nodes compute at block boundaries and hold their values between updates. Setting `control_interval = 0` (the default) preserves the existing single-loop behavior.

Validation enforces that control-rate nodes cannot depend on audio inputs or audio-rate nodes. Dependencies on params, other control-rate nodes, and invariant nodes are allowed.

## Graph Algebra

FAUST-style block diagram combinators for composing graphs without manually wiring `Subgraph` nodes. Four combinators build new `Graph` objects from existing ones:

```python
from dsp_graph import Graph, AudioInput, AudioOutput, OnePole, Param
from dsp_graph.algebra import series, parallel, split, merge

lpf = Graph(
    name="lpf",
    inputs=[AudioInput(id="in")],
    outputs=[AudioOutput(id="out", source="filt")],
    params=[Param(name="coeff", default=0.5)],
    nodes=[OnePole(id="filt", a="in", coeff="coeff")],
)
hpf = Graph(name="hpf", inputs=[AudioInput(id="in")],
            outputs=[AudioOutput(id="out", source="filt")],
            params=[Param(name="coeff", default=0.3)],
            nodes=[OnePole(id="filt", a="in", coeff="coeff")])

# Series: pipe outputs -> inputs (requires matching counts)
chain = series(lpf, hpf)  # 1-in, 1-out, params: lpf_coeff, hpf_coeff

# Parallel: stack side by side (independent I/O)
stack = parallel(lpf, hpf)  # 2-in (lpf_in, hpf_in), 2-out (lpf_out, hpf_out)

# Split: fan-out (cyclic distribution, requires len(b.inputs) % len(a.outputs) == 0)
duo = split(lpf, stack)  # 1-in, 2-out (lpf output duplicated to both)

# Merge: fan-in (grouped summing, requires len(a.outputs) % len(b.inputs) == 0)
mono = merge(stack, lpf)  # 2-in, 1-out (stack outputs summed into lpf input)
```

Operator overloading provides concise syntax -- `>>` for series, `//` for parallel:

```python
chain = lpf >> hpf           # series
stack = lpf // hpf           # parallel
multi = (lpf // hpf) >> out  # parallel then series
```

Params are namespaced with the subgraph ID prefix at each level of nesting. For `series(lpf, hpf)`, inner param `coeff` becomes `lpf_coeff` and `hpf_coeff`. This compounds with deeper nesting: `series(series(a, b), c)` produces params like `a__b_a_coeff`.

All composed graphs work with `expand_subgraphs()`, `compile_graph()`, `validate_graph()`, and `simulate()`.

## CLI

The `dsp-graph` command-line tool compiles, validates, and visualizes graph JSON files.

```bash
# Compile graph to C++ (stdout)
dsp-graph compile graph.json

# Compile to directory with optimization
dsp-graph compile graph.json -o build/ --optimize

# Compile with gen-dsp adapter for a specific platform
dsp-graph compile graph.json --gen-dsp chuck -o build/

# Validate graph JSON
dsp-graph validate graph.json

# Generate DOT visualization (stdout or directory)
dsp-graph dot graph.json -o build/
```

## Node Types (38)

### Arithmetic / Math

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `BinOp` | `add`, `sub`, `mul`, `div`, `min`, `max`, `mod`, `pow` | `a`, `b` | Binary arithmetic |
| `UnaryOp` | `sin`, `cos`, `tanh`, `exp`, `log`, `abs`, `sqrt`, `neg`, `floor`, `ceil`, `round`, `sign`, `atan`, `asin`, `acos` | `a` | Unary math functions |
| `Clamp` | `clamp` | `a`, `lo`, `hi` | Saturate to `[lo, hi]` |
| `Constant` | `constant` | `value` | Literal float value |
| `Compare` | `gt`, `lt`, `gte`, `lte`, `eq` | `a`, `b` | Comparison (returns 0.0 or 1.0) |
| `Select` | `select` | `cond`, `a`, `b` | Conditional: `a` if `cond > 0`, else `b` |
| `Wrap` | `wrap` | `a`, `lo`, `hi` | Wrap value into range |
| `Fold` | `fold` | `a`, `lo`, `hi` | Fold (reflect) value into range |
| `Mix` | `mix` | `a`, `b`, `t` | Linear interpolation: `a + (b - a) * t` |
| `Scale` | `scale` | `a`, `in_lo`, `in_hi`, `out_lo`, `out_hi` | Linear range mapping |

### Delay

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `DelayLine` | `delay` | `max_samples` | Circular buffer declaration |
| `DelayRead` | `delay_read` | `delay`, `tap`, `interp` | Read from delay line (none/linear/cubic) |
| `DelayWrite` | `delay_write` | `delay`, `value` | Write to delay line |
| `History` | `history` | `input`, `init` | Single-sample delay (z^-1 feedback) |

### Buffer / Table

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `Buffer` | `buffer` | `size` | Random-access data buffer |
| `BufRead` | `buf_read` | `buffer`, `index`, `interp` | Read from buffer (none/linear/cubic, clamped) |
| `BufWrite` | `buf_write` | `buffer`, `index`, `value` | Write to buffer at index |
| `BufSize` | `buf_size` | `buffer` | Returns buffer length as float |

### Filters

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `Biquad` | `biquad` | `a`, `b0`, `b1`, `b2`, `a1`, `a2` | Generic biquad (user supplies coefficients) |
| `SVF` | `svf` | `a`, `freq`, `q`, `mode` | State-variable filter (lp/hp/bp/notch) |
| `OnePole` | `onepole` | `a`, `coeff` | One-pole lowpass |
| `DCBlock` | `dcblock` | `a` | DC blocking filter |
| `Allpass` | `allpass` | `a`, `coeff` | First-order allpass |

### Oscillators / Sources

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `Phasor` | `phasor` | `freq` | Ramp oscillator 0..1 |
| `SinOsc` | `sinosc` | `freq` | Sine oscillator |
| `TriOsc` | `triosc` | `freq` | Triangle wave |
| `SawOsc` | `sawosc` | `freq` | Bipolar saw (-1..1) |
| `PulseOsc` | `pulseosc` | `freq`, `width` | Pulse/square with variable duty cycle |
| `Noise` | `noise` | -- | White noise source |

### State / Timing

| Node | `op` | Fields | Purpose |
|------|------|--------|---------|
| `Delta` | `delta` | `a` | Sample-to-sample difference |
| `Change` | `change` | `a` | 1.0 if value changed, else 0.0 |
| `SampleHold` | `sample_hold` | `a`, `trig` | Latch on any zero crossing |
| `Latch` | `latch` | `a`, `trig` | Latch on rising edge only |
| `Accum` | `accum` | `incr`, `reset` | Running sum, resets when `reset > 0` |
| `Counter` | `counter` | `trig`, `max` | Integer counter, wraps at max |
| `RateDiv` | `rate_div` | `a`, `divisor` | Output every N-th sample, hold between |
| `SmoothParam` | `smooth` | `a`, `coeff` | One-pole smoothing for param changes |
| `Peek` | `peek` | `a` | Debug pass-through, readable externally |

## C++ Compilation

`compile_graph()` generates a single self-contained `.cpp` file with:

- A state struct (`{Name}State`)
- `create(sr)` / `destroy(self)` / `reset(self)` lifecycle
- `perform(self, ins, outs, n)` sample-processing loop
- Param introspection: `num_params`, `param_name`, `param_min`, `param_max`, `set_param`, `get_param`
- Buffer introspection: `num_buffers`, `buffer_name`, `buffer_size`, `get_buffer`, `set_buffer`
- Peek introspection: `num_peeks`, `peek_name`, `get_peek`

```python
from dsp_graph import compile_graph, compile_graph_to_file

code = compile_graph(graph)           # returns C++ string
path = compile_graph_to_file(graph, "build/")  # writes build/{name}.cpp
```

## gen-dsp Integration

dsp-graph graphs can be compiled into buildable audio plugin projects via [gen-dsp](https://github.com/shakfu/gen-dsp), which supports 11 platforms: ChucK, CLAP, AudioUnit, VST3, LV2, SuperCollider, VCV Rack, Daisy, and more.

`compile_for_gen_dsp()` generates the three files needed to drop into any gen-dsp platform backend:

```python
from dsp_graph import compile_for_gen_dsp

# Generates: test_synth.cpp, _ext_chuck.cpp, manifest.json
compile_for_gen_dsp(graph, "build/", platform="chuck")
```

The adapter replaces gen-dsp's genlib-side code while reusing its platform-side code unchanged. The generated `manifest.json` is compatible with `gen_dsp.core.manifest.Manifest`.

For a fully assembled project (requires gen-dsp installed):

```python
from dsp_graph.gen_dsp_adapter import assemble_project

# Copies platform templates + generates adapter + manifest
assemble_project(graph, "build/chuck_project", platform="chuck")
```

## Optimization

```python
from dsp_graph import optimize_graph, constant_fold, eliminate_cse, eliminate_dead_nodes

optimized = optimize_graph(graph)  # constant folding + CSE + dead node elimination
```

- **Constant folding**: pure nodes with all-constant inputs are replaced by `Constant` nodes
- **Common subexpression elimination**: duplicate pure nodes with identical inputs are merged
- **Dead node elimination**: nodes not reachable from any output are removed (respects side-effecting writers for delay lines and buffers)
- **Loop-invariant code motion**: param-only expressions are hoisted before the sample loop
- **Multi-rate processing**: control-rate nodes run once per block in an outer loop, reducing per-sample overhead for smoothing/coefficient computation
- **SIMD hints**: `__restrict` on I/O pointers; vectorization pragmas for pure-only graphs

## Validation

`validate_graph()` checks:

1. Unique node IDs (no collisions with inputs or params)
2. All string references resolve to existing IDs
3. Output sources reference existing nodes
4. DelayRead/DelayWrite reference existing DelayLine nodes
5. BufRead/BufWrite/BufSize reference existing Buffer nodes
6. Control-rate consistency: `control_nodes` reference existing nodes, don't depend on audio inputs or audio-rate nodes
7. No pure cycles (cycles must pass through History or delay)

## Visualization

```python
from dsp_graph import graph_to_dot, graph_to_dot_file

dot_str = graph_to_dot(graph)                  # DOT string
dot_path = graph_to_dot_file(graph, "build/")  # writes .dot, renders .pdf if `dot` is on PATH
```

## Examples

See `examples/` for complete working graphs:

- `stereo_gain.py` -- stateless stereo gain
- `onepole.py` -- one-pole lowpass with History feedback
- `fbdelay.py` -- feedback delay with dry/wet mix
- `wavetable.py` -- wavetable oscillator using Buffer + Phasor + BufRead
- `smooth_gain.py` -- control-rate parameter smoothing (multi-rate)
- `multirate_synth.py` -- control-rate envelope with audio-rate oscillator (multi-rate)
- `filter_chain.py` -- two filters in series using graph algebra

## Development

```bash
make install-dev  # install with dev deps
make test         # run tests
make lint         # ruff check
make typecheck    # mypy --strict
make qa           # all of the above
```