02 — Neuro-Symbolic Compiler: Petri Nets → Stochastic Neurons

SCPN Control includes a neuro-symbolic compiler that converts Petri net control logic into stochastic LIF (leaky integrate-and-fire) neuron networks. This tutorial walks through the full pipeline:

1. Define a Petri net
2. Compile it to a stochastic neural network
3. Run inference (dense float-path)
4. Export/import artifacts for deployment

License: © 1998–2026 Miroslav Šotek. GNU AGPL v3.

---

import numpy as np
import matplotlib.pyplot as plt
from scpn_control.scpn import StochasticPetriNet, FusionCompiler

sc_neurocore not installed — numpy float-path only

Step 1: Define a Plasma Control Petri Net

We model a simplified tokamak position controller with:

• 4 input places (sensor observations: R_high, R_low, Z_high, Z_low)
• 4 transitions (decision logic)
• 4 output places (actuator commands: PF_up, PF_down, PF_in, PF_out)

net = StochasticPetriNet()

# Input places (sensor observations)
net.add_place("R_high", initial_tokens=0.0)
net.add_place("R_low", initial_tokens=0.0)
net.add_place("Z_high", initial_tokens=0.0)
net.add_place("Z_low", initial_tokens=0.0)

# Output places (actuator commands)
net.add_place("PF_up", initial_tokens=0.0)
net.add_place("PF_down", initial_tokens=0.0)
net.add_place("PF_in", initial_tokens=0.0)
net.add_place("PF_out", initial_tokens=0.0)

# Transitions (control logic)
net.add_transition("T_correct_R_high", threshold=0.5)
net.add_transition("T_correct_R_low", threshold=0.5)
net.add_transition("T_correct_Z_high", threshold=0.5)
net.add_transition("T_correct_Z_low", threshold=0.5)

# Arcs: if R is too high → move plasma inward
net.add_arc("R_high", "T_correct_R_high", weight=1.0)
net.add_arc("T_correct_R_high", "PF_in", weight=1.0)

# If R is too low → move plasma outward
net.add_arc("R_low", "T_correct_R_low", weight=1.0)
net.add_arc("T_correct_R_low", "PF_out", weight=1.0)

# If Z is too high → push plasma down
net.add_arc("Z_high", "T_correct_Z_high", weight=1.0)
net.add_arc("T_correct_Z_high", "PF_down", weight=1.0)

# If Z is too low → push plasma up
net.add_arc("Z_low", "T_correct_Z_low", weight=1.0)
net.add_arc("T_correct_Z_low", "PF_up", weight=1.0)

net.compile()
print(net.summary())

StochasticPetriNet  P=8  T=4  Arcs=8  compiled=True

Places:
  [0] R_high                tokens=0.000
  [1] R_low                 tokens=0.000
  [2] Z_high                tokens=0.000
  [3] Z_low                 tokens=0.000
  [4] PF_up                 tokens=0.000
  [5] PF_down               tokens=0.000
  [6] PF_in                 tokens=0.000
  [7] PF_out                tokens=0.000

Transitions:
  [0] T_correct_R_high      threshold=0.500  delay_ticks=0
  [1] T_correct_R_low       threshold=0.500  delay_ticks=0
  [2] T_correct_Z_high      threshold=0.500  delay_ticks=0
  [3] T_correct_Z_low       threshold=0.500  delay_ticks=0

Arcs:
  R_high --(1.000)--> T_correct_R_high
  T_correct_R_high --(1.000)--> PF_in
  R_low --(1.000)--> T_correct_R_low
  T_correct_R_low --(1.000)--> PF_out
  Z_high --(1.000)--> T_correct_Z_high
  T_correct_Z_high --(1.000)--> PF_down
  Z_low --(1.000)--> T_correct_Z_low
  T_correct_Z_low --(1.000)--> PF_up

W_in  (nT=4, nP=8)  nnz=4
W_out (nP=8, nT=4)  nnz=4

Step 2: Compile to Stochastic Neural Network

The compiler maps each transition to a stochastic LIF neuron. If sc-neurocore is installed, it uses hardware-accurate bitstream encoding. Otherwise, it falls back to NumPy float computation.

compiler = FusionCompiler(bitstream_length=1024, seed=42)
compiled = compiler.compile(net)

print(f"Places:      {compiled.n_places}")
print(f"Transitions: {compiled.n_transitions}")
print(f"Stochastic:  {compiled.has_stochastic_path}")
print(f"Firing mode: {compiled.firing_mode}")
print()
print(compiled.summary())

Places:      8
Transitions: 4
Stochastic:  False
Firing mode: binary

CompiledNet  P=8  T=4  L=1024  mode=float-only

Step 3: Run Inference

We simulate a scenario where the plasma is displaced to R_high and Z_low. The compiled network should activate PF_in (radial correction) and PF_up (vertical correction).

# Inject observation: plasma displaced R_high + Z_low
marking = np.zeros(compiled.n_places)
marking[0] = 0.8  # R_high active
marking[3] = 0.9  # Z_low active

W_in = compiled.W_in.toarray() if hasattr(compiled.W_in, "toarray") else np.asarray(compiled.W_in)
W_out = compiled.W_out.toarray() if hasattr(compiled.W_out, "toarray") else np.asarray(compiled.W_out)

print("Initial marking:", dict(zip(net.place_names, marking)))

# Step: compute transition firing
currents = W_in @ marking
fired = (currents >= compiled.thresholds).astype(float)
consumed = W_in.T @ fired
produced = W_out @ fired
new_marking = np.clip(marking - consumed + produced, 0.0, 1.0)

print("\nFired transitions:", dict(zip(net.transition_names, fired)))
print("\nNew marking:", dict(zip(net.place_names, new_marking)))
print("\n→ PF_in activated:", new_marking[6] > 0)  # PF_in
print("→ PF_up activated:", new_marking[4] > 0)  # PF_up

Initial marking: {'R_high': np.float64(0.8), 'R_low': np.float64(0.0), 'Z_high': np.float64(0.0), 'Z_low': np.float64(0.9), 'PF_up': np.float64(0.0), 'PF_down': np.float64(0.0), 'PF_in': np.float64(0.0), 'PF_out': np.float64(0.0)}

Fired transitions: {'T_correct_R_high': np.float64(1.0), 'T_correct_R_low': np.float64(0.0), 'T_correct_Z_high': np.float64(0.0), 'T_correct_Z_low': np.float64(1.0)}

New marking: {'R_high': np.float64(0.0), 'R_low': np.float64(0.0), 'Z_high': np.float64(0.0), 'Z_low': np.float64(0.0), 'PF_up': np.float64(1.0), 'PF_down': np.float64(0.0), 'PF_in': np.float64(1.0), 'PF_out': np.float64(0.0)}

→ PF_in activated: True
→ PF_up activated: True

Step 4: Multi-Step Evolution

Run the network for 30 steps with a time-varying disturbance signal.

n_steps = 30
history = np.zeros((n_steps + 1, compiled.n_places))
marking = np.zeros(compiled.n_places)
history[0] = marking

for k in range(n_steps):
    # Time-varying disturbance
    t = k / n_steps
    marking[0] = 0.6 * np.sin(2 * np.pi * t) ** 2  # R_high oscillation
    marking[3] = 0.5 * np.cos(2 * np.pi * t) ** 2  # Z_low oscillation

    currents = W_in @ marking
    fired = (currents >= compiled.thresholds).astype(float)
    consumed = W_in.T @ fired
    produced = W_out @ fired
    marking = np.clip(marking - consumed + produced, 0.0, 1.0)
    history[k + 1] = marking

fig, axes = plt.subplots(2, 1, figsize=(10, 6), sharex=True)
for i in range(4):
    axes[0].plot(history[:, i], label=net.place_names[i])
axes[0].set_ylabel("Input Places")
axes[0].legend(loc="upper right")
axes[0].set_title("Petri Net Token Evolution (30 steps)")

for i in range(4, 8):
    axes[1].plot(history[:, i], label=net.place_names[i])
axes[1].set_ylabel("Output Places")
axes[1].set_xlabel("Step")
axes[1].legend(loc="upper right")
plt.tight_layout()
plt.show()

Step 5: Artifact Export / Import

The compiled network can be serialised as a JSON artifact for deployment on embedded hardware or real-time controllers.

import tempfile
import os
from scpn_control.scpn import load_artifact, save_artifact

# Export
artifact = compiled.export_artifact(
    name="position_controller_v1",
    dt_control_s=0.001,
    readout_config={
        "actions": [
            {"name": "dI_vertical_A", "pos_place": 4, "neg_place": 5},
            {"name": "dI_radial_A", "pos_place": 7, "neg_place": 6},
        ],
        "gains": [1000.0, 500.0],
        "abs_max": [5000.0, 3000.0],
        "slew_per_s": [1e5, 5e4],
    },
    injection_config=[
        {"place_id": 0, "source": "R_high", "scale": 1.0, "offset": 0.0, "clamp_0_1": True},
        {"place_id": 1, "source": "R_low", "scale": 1.0, "offset": 0.0, "clamp_0_1": True},
        {"place_id": 2, "source": "Z_high", "scale": 1.0, "offset": 0.0, "clamp_0_1": True},
        {"place_id": 3, "source": "Z_low", "scale": 1.0, "offset": 0.0, "clamp_0_1": True},
    ],
)

fd, path = tempfile.mkstemp(suffix=".scpnctl.json")
os.close(fd)
save_artifact(artifact, path)
print(f"Saved artifact to: {path}")
print(f"File size: {os.path.getsize(path)} bytes")

# Reload
loaded = load_artifact(path)
print(f"\nReloaded: {loaded.meta.name}")
print(f"Places: {loaded.nP}, Transitions: {loaded.nT}")
os.unlink(path)

Saved artifact to: /tmp/tmpa25on1mc.scpnctl.json
File size: 3867 bytes

Reloaded: position_controller_v1
Places: 8, Transitions: 4

Performance Benchmarks

Timing the key computations in this notebook:

1. Petri net compilation (FusionCompiler.compile)
2. Single inference step (matrix multiply + threshold)
3. 30-step evolution loop (multi-step token propagation)
4. Artifact export/import round-trip

import timeit


# 1. Petri net compilation
def bench_compile():
    c = FusionCompiler(bitstream_length=1024, seed=42)
    c.compile(net)


t_compile = timeit.repeat(bench_compile, number=10, repeat=5)
print("FusionCompiler.compile (10 calls):")
print(f"  Mean: {np.mean(t_compile) * 1000:.1f} ms +/- {np.std(t_compile) * 1000:.1f} ms")
print(f"  Per call: {np.mean(t_compile) / 10 * 1000:.2f} ms")

# 2. Single inference step (matrix multiply + threshold)
W_in_dense = compiled.W_in.toarray() if hasattr(compiled.W_in, "toarray") else np.asarray(compiled.W_in)
W_out_dense = compiled.W_out.toarray() if hasattr(compiled.W_out, "toarray") else np.asarray(compiled.W_out)
test_marking = np.zeros(compiled.n_places)
test_marking[0] = 0.8
test_marking[3] = 0.9


def bench_single_step():
    currents = W_in_dense @ test_marking
    fired = (currents >= compiled.thresholds).astype(float)
    consumed = W_in_dense.T @ fired
    produced = W_out_dense @ fired
    np.clip(test_marking - consumed + produced, 0.0, 1.0)


t_step = timeit.repeat(bench_single_step, number=10000, repeat=5)
print("\nSingle inference step (10000 calls):")
print(f"  Mean: {np.mean(t_step) * 1000:.1f} ms +/- {np.std(t_step) * 1000:.1f} ms")
print(f"  Per call: {np.mean(t_step) / 10000 * 1e6:.2f} us")


# 3. 30-step evolution loop
def bench_evolution():
    m = np.zeros(compiled.n_places)
    for k in range(30):
        t = k / 30
        m[0] = 0.6 * np.sin(2 * np.pi * t) ** 2
        m[3] = 0.5 * np.cos(2 * np.pi * t) ** 2
        currents = W_in_dense @ m
        fired = (currents >= compiled.thresholds).astype(float)
        consumed = W_in_dense.T @ fired
        produced = W_out_dense @ fired
        m = np.clip(m - consumed + produced, 0.0, 1.0)


t_evo = timeit.repeat(bench_evolution, number=100, repeat=5)
print("\n30-step evolution (100 runs):")
print(f"  Mean: {np.mean(t_evo) * 1000:.1f} ms +/- {np.std(t_evo) * 1000:.1f} ms")
print(f"  Per run: {np.mean(t_evo) / 100 * 1000:.2f} ms")

# 4. Artifact export round-trip
import tempfile
import os


def bench_export_import():
    art = compiled.export_artifact(
        name="bench_test",
        dt_control_s=0.001,
        readout_config={"actions": [], "gains": [], "abs_max": [], "slew_per_s": []},
        injection_config=[],
    )
    fd, p = tempfile.mkstemp(suffix=".scpnctl.json")
    os.close(fd)
    save_artifact(art, p)
    load_artifact(p)
    os.unlink(p)


t_io = timeit.repeat(bench_export_import, number=10, repeat=3)
print("\nArtifact export/import round-trip (10 calls):")
print(f"  Mean: {np.mean(t_io) * 1000:.1f} ms +/- {np.std(t_io) * 1000:.1f} ms")
print(f"  Per call: {np.mean(t_io) / 10 * 1000:.2f} ms")

FusionCompiler.compile (10 calls):
  Mean: 0.1 ms +/- 0.0 ms
  Per call: 0.01 ms

Single inference step (10000 calls):
  Mean: 72.2 ms +/- 5.0 ms
  Per call: 7.22 us

30-step evolution (100 runs):
  Mean: 27.2 ms +/- 1.6 ms
  Per run: 0.27 ms

Artifact export/import round-trip (10 calls):
  Mean: 28.6 ms +/- 3.8 ms
  Per call: 2.86 ms

Summary

The neuro-symbolic compiler pipeline:

1. Define plasma control logic as a Stochastic Petri Net
2. Compile to stochastic LIF neurons (with optional SC-NeuroCore)
3. Run inference: inject observations → fire transitions → read actuator commands
4. Export as JSON artifact for deployment

This architecture enables sub-millisecond real-time plasma control with formally verifiable logic.

Next: See 03_flight_simulator.ipynb for integration with the tokamak flight simulator.