scpn_fusion.scpn – Neuro-Symbolic

The SCPN subpackage implements the Petri net to spiking neural network compilation pipeline – the core innovation of SCPN-Fusion-Core.

Petri Net Data Structures

Packet A — Stochastic Petri Net structure definition.

Pure Python + numpy + scipy. No sc_neurocore dependency.

Builds two sparse matrices that encode the net topology:

W_in : (n_transitions, n_places) — input arc weights W_out : (n_places, n_transitions) — output arc weights

class scpn_fusion.scpn.structure.StochasticPetriNet[source]

Bases: object

Stochastic Petri Net with sparse matrix representation.

Usage:

net = StochasticPetriNet()
net.add_place("P_red",   initial_tokens=1.0)
net.add_place("P_green", initial_tokens=0.0)
net.add_transition("T_r2g", threshold=0.5)
net.add_arc("P_red", "T_r2g", weight=1.0)
net.add_arc("T_r2g", "P_green", weight=1.0)
net.compile()
add_place(name, initial_tokens=0.0)[source]

Add a place (state variable) with token density in [0, 1].

Return type:

None

Parameters:
add_transition(name, threshold=0.5, delay_ticks=0)[source]

Add a transition (logic gate) with a firing threshold.

Return type:

None

Parameters:
add_arc(source, target, weight=1.0, inhibitor=False)[source]

Add a directed arc between a Place and a Transition (either direction).

Valid arcs:

Place -> Transition (input arc, stored in W_in) Transition -> Place (output arc, stored in W_out)

Parameters:

  • inhibitor (bool, default False) – If True, arc is encoded as a negative weight inhibitor input arc. Only valid for Place -> Transition edges.

  • nodes. (Raises ValueError for same-kind connections or unknown)

  • source (str)

  • target (str)

  • weight (float)

Return type:

None

compile(validate_topology=False, strict_validation=False, allow_inhibitor=False)[source]

Build sparse W_in and W_out matrices from the arc list.

Parameters:

  • validate_topology (bool, default False) – Compute and store topology diagnostics in last_validation_report.

  • strict_validation (bool, default False) – If True, raise ValueError when topology diagnostics contain dead nodes or unseeded place cycles.

  • allow_inhibitor (bool, default False) – Allow negative Place -> Transition weights for inhibitor arcs. If False, compile fails when inhibitor arcs are present.

Return type:

None

validate_topology()[source]

Return topology diagnostics without mutating compiled matrices.

Diagnostics: - dead_places: places with zero total degree. :rtype: Dict[str, object]

  • dead_transitions: transitions with zero total degree.

  • unseeded_place_cycles: place-level SCC cycles with no initial token mass.

  • input_weight_overflow_transitions: transitions whose positive input arc-weight sum exceeds 1.0.

Return type:

Dict[str, object]

property n_places: int
property n_transitions: int
property place_names: List[str]
property transition_names: List[str]
property is_compiled: bool
property last_validation_report: Dict[str, object] | None
get_initial_marking()[source]

Return (n_places,) float64 vector of initial token densities.

Return type:

ndarray[Any, dtype[float64]]

get_thresholds()[source]

Return (n_transitions,) float64 vector of firing thresholds.

Return type:

ndarray[Any, dtype[float64]]

get_delay_ticks()[source]

Return (n_transitions,) int64 vector of transition delay ticks.

Return type:

ndarray[Any, dtype[int64]]

summary()[source]

Human-readable summary of the net.

Return type:

str

verify_boundedness(n_steps=500, n_trials=100)[source]

Verify that markings stay in [0, 1] under random firing.

Runs n_trials random trajectories of n_steps each, checking that all markings remain in [0, 1] at every step.

Returns:

{“bounded”: bool, “max_marking”: float, “min_marking”: float,

”n_trials”: int, “n_steps”: int}

Return type:

dict

Parameters:

  • n_steps (int)

  • n_trials (int)

verify_liveness(n_steps=200, n_trials=1000)[source]

Verify that all transitions fire at least once in random campaigns.

Returns:

{“live”: bool, “transition_fire_pct”: dict[str, float],

”min_fire_pct”: float}

Return type:

dict

Parameters:

  • n_steps (int)

  • n_trials (int)

Formal Contracts

Data contracts for the SCPN Fusion-Core Control API.

Observation / action TypedDicts, feature extraction (obs → unipolar [0,1]), action decoding (marking → slew-limited actuator commands).

class scpn_fusion.scpn.contracts.ControlObservation[source]

Bases: TypedDict

Plant observation at a single control tick.

R_axis_m: float
Z_axis_m: float
class scpn_fusion.scpn.contracts.ControlAction[source]

Bases: TypedDict

Actuator command output for a single control tick.

Keys are dynamic and depend on the Petri Net readout configuration.

dI_PF3_A: float
dI_PF_topbot_A: float
class scpn_fusion.scpn.contracts.ControlTargets(R_target_m=6.2, Z_target_m=0.0)[source]

Bases: object

Setpoint targets for the control loop.

Parameters:
R_target_m: float = 6.2
Z_target_m: float = 0.0
class scpn_fusion.scpn.contracts.ControlScales(R_scale_m=0.5, Z_scale_m=0.5)[source]

Bases: object

Normalisation scales (error → [-1, 1] range).

Parameters:
R_scale_m: float = 0.5
Z_scale_m: float = 0.5
class scpn_fusion.scpn.contracts.FeatureAxisSpec(obs_key, target, scale, pos_key, neg_key)[source]

Bases: object

Configurable feature axis mapping from observation -> unipolar features.

Parameters:

  • obs_key (observation key to read.)

  • target (target/setpoint value for this axis.)

  • scale (normalisation scale for signed error (target - obs) / scale.)

  • pos_key (output feature key for positive error component.)

  • neg_key (output feature key for negative error component.)

obs_key: str
target: float
scale: float
pos_key: str
neg_key: str
scpn_fusion.scpn.contracts.extract_features(obs, targets, scales, feature_axes=None, passthrough_keys=None)[source]

Map observation → unipolar [0, 1] feature sources.

By default returns keys x_R_pos, x_R_neg, x_Z_pos, x_Z_neg. Custom feature mappings can be supplied via feature_axes for arbitrary observation dictionaries.

Return type:

Dict[str, float]

Parameters:
class scpn_fusion.scpn.contracts.ActionSpec(name, pos_place, neg_place)[source]

Bases: object

One action channel: positive/negative place differencing.

Parameters:
name: str
pos_place: int
neg_place: int
scpn_fusion.scpn.contracts.decode_actions(marking, actions_spec, gains, abs_max, slew_per_s, dt, prev)[source]

Decode marking → actuator commands with gain, slew-rate, and abs clamp.

Parameters:

  • marking (current marking vector (len >= max place index).)

  • actions_spec (per-action pos/neg place definitions.)

  • gains (per-action gain multiplier.)

  • abs_max (per-action absolute saturation.)

  • slew_per_s (per-action max change rate (units/s).)

  • dt (control tick period (s).)

  • prev (previous action outputs (same length as actions_spec).)

Return type:

dict mapping action name → clamped value. Also mutates prev in-place.

class scpn_fusion.scpn.contracts.PhysicsInvariant(name, description, threshold, comparator)[source]

Bases: object

A hard physics constraint the controller loop must respect.

Parameters:

  • name (str) – Short identifier for the invariant (e.g. "q_min", "beta_N").

  • description (str) – Human-readable description including the physics origin.

  • threshold (float) – Threshold value for the invariant condition.

  • comparator (str) – One of "gt", "lt", "gte", "lte" — the relationship that the measured value must satisfy with respect to threshold for the invariant to hold.

name: str
description: str
threshold: float
comparator: str
class scpn_fusion.scpn.contracts.PhysicsInvariantViolation(invariant, actual_value, margin, severity)[source]

Bases: object

Record of a physics invariant violation.

Parameters:

  • invariant (PhysicsInvariant) – The invariant that was violated.

  • actual_value (float) – The measured/computed value that violated the invariant.

  • margin (float) – Absolute distance between actual_value and the invariant threshold (always >= 0).

  • severity (str) – "warning" if margin <= 20 % of abs(threshold), "critical" otherwise.

invariant: PhysicsInvariant
actual_value: float
margin: float
severity: str
scpn_fusion.scpn.contracts.check_physics_invariant(invariant, value)[source]

Check a single physics invariant against a measured value.

Returns None if the invariant is satisfied, otherwise returns a PhysicsInvariantViolation with computed margin and severity.

Return type:

Optional[PhysicsInvariantViolation]

Parameters:

Severity classification

  • "critical" — margin exceeds 20 % of abs(threshold) (or 20 % of 1.0 when threshold == 0).

  • "warning" — violated but within the 20 % band.

type invariant:

PhysicsInvariant

param invariant:

The invariant to check.

type invariant:

PhysicsInvariant

type value:

float

param value:

Current measured / computed value for the quantity.

type value:

float

scpn_fusion.scpn.contracts.check_all_invariants(values, invariants=None)[source]

Check every invariant whose name appears in values.

Parameters:

  • values (dict) – Mapping from invariant name to the current measured value. Names not present in the invariant list are silently ignored.

  • invariants (list, optional) – The invariant set to check. Defaults to DEFAULT_PHYSICS_INVARIANTS.

Returns:

All detected violations (empty list when everything is nominal).

Return type:

list[PhysicsInvariantViolation]

scpn_fusion.scpn.contracts.should_trigger_mitigation(violations)[source]

Return True if any violation has severity == "critical".

This is the top-level disruption-mitigation gate: a single critical violation means the controller must engage protective actions (e.g. massive gas injection, current quench, or safe ramp-down).

Return type:

bool

Parameters:

violations (List[PhysicsInvariantViolation])

class scpn_fusion.scpn.contracts.SafetyContract(safety_place, control_transition)[source]

Bases: object

Symbolic safety contract linking a limit place to a control transition.

When safety_place has tokens > 0, control_transition must be disabled by inhibitor-arc semantics.

Parameters:

  • safety_place (str)

  • control_transition (str)

safety_place: str
control_transition: str
scpn_fusion.scpn.contracts.verify_safety_contracts(*, safety_tokens, transition_enabled, contracts=(SafetyContract(safety_place='thermal_limit', control_transition='heat_ramp'), SafetyContract(safety_place='density_limit', control_transition='density_ramp'), SafetyContract(safety_place='beta_limit', control_transition='power_ramp'), SafetyContract(safety_place='current_limit', control_transition='current_ramp'), SafetyContract(safety_place='vertical_limit', control_transition='position_move')))[source]

Return textual violations of inhibitor safety contracts.

A contract is violated iff: safety_tokens[safety_place] > 0 and transition_enabled[control_transition] is True.

Return type:

list[str]

Parameters:

Fusion Compiler

Packet B — FusionCompiler and CompiledNet.

Compiles a StochasticPetriNet into sc_neurocore artifacts:

  • One StochasticLIFNeuron per transition (pure threshold comparator).

  • Pre-packed uint64 weight bitstreams for AND+popcount forward pass.

  • Float-path compatibility route when sc_neurocore is not installed.

class scpn_fusion.scpn.compiler.CompiledNet(n_places, n_transitions, place_names, transition_names, W_in, W_out, W_in_packed=None, W_out_packed=None, neurons=<factory>, bitstream_length=1024, thresholds=<factory>, transition_delay_ticks=<factory>, initial_marking=<factory>, seed=42, firing_mode='binary', firing_margin=0.05, lif_tau_mem=1000000.0, lif_noise_std=0.0, lif_dt=1.0, lif_resistance=1.0, lif_refractory_period=0)[source]

Bases: object

Compiled Petri Net ready for sc_neurocore execution.

Holds both the dense float matrices (for validation / compatibility) and pre-packed uint64 weight bitstreams (for the stochastic path).

Parameters:
n_places: int
n_transitions: int
place_names: List[str]
transition_names: List[str]
W_in: ndarray[Any, dtype[float64]]
W_out: ndarray[Any, dtype[float64]]
W_in_packed: Optional[ndarray[Any, dtype[uint64]]] = None
W_out_packed: Optional[ndarray[Any, dtype[uint64]]] = None
neurons: List[Any]
bitstream_length: int = 1024
thresholds: ndarray[Any, dtype[float64]]
transition_delay_ticks: ndarray[Any, dtype[int64]]
initial_marking: ndarray[Any, dtype[float64]]
seed: int = 42
firing_mode: str = 'binary'
firing_margin: float = 0.05
lif_tau_mem: float = 1000000.0
lif_noise_std: float = 0.0
lif_dt: float = 1.0
lif_resistance: float = 1.0
lif_refractory_period: int = 0
property has_stochastic_path: bool
dense_forward(W_packed, input_probs)[source]

Stochastic matrix-vector product via AND + popcount.

Parameters:

  • W_packed ((n_out, n_in, n_words) uint64 — pre-packed weight bitstreams.)

  • input_probs ((n_in,) float64 — input probabilities in [0, 1].)

Returns:

output

Return type:

(n_out,) float64 — stochastic estimate of W @ input_probs.

dense_forward_float(W, inputs)[source]

Float-path validation: simple W @ inputs.

Return type:

ndarray[Any, dtype[float64]]

Parameters:
lif_fire(currents)[source]

Run LIF threshold detection on all transitions.

Binary mode: f_t = 1 if current >= threshold else 0 Fractional mode: f_t = clip((current - threshold) / margin, 0, 1)

Parameters:

currents ((n_transitions,) float64 — weighted-sum activations.)

Returns:

fired – Binary mode → values in {0.0, 1.0}. Fractional mode → values in [0.0, 1.0].

Return type:

(n_transitions,) float64 vector.

summary()[source]
Return type:

str

export_artifact(name='controller', dt_control_s=0.001, readout_config=None, injection_config=None)[source]

Build an Artifact from compiled state + user-provided config.

Parameters:

  • name (artifact name.)

  • dt_control_s (control tick period (s).)

  • readout_config (dict with actions, gains, abs_max,) – slew_per_s lists. Required for a complete artifact.

  • injection_config (list of place-injection dicts.)

Return type:

Any

class scpn_fusion.scpn.compiler.FusionCompiler(bitstream_length=1024, seed=42, *, lif_tau_mem=1000000.0, lif_noise_std=0.0, lif_dt=1.0, lif_resistance=1.0, lif_refractory_period=0)[source]

Bases: object

Compiles a StochasticPetriNet into a CompiledNet.

Parameters:

  • bitstream_length (Number of bits per stochastic stream (default 1024).)

  • seed (Base RNG seed for reproducibility.)

  • lif_tau_mem (float)

  • lif_noise_std (float)

  • lif_dt (float)

  • lif_resistance (float)

  • lif_refractory_period (int)

classmethod with_reactor_lif_defaults(bitstream_length=1024, seed=42, *, lif_dt=1.0, lif_resistance=1.0, lif_refractory_period=0)[source]

Create a compiler with reactor-like LIF defaults.

Keeps the legacy constructor defaults untouched for compatibility while exposing a realistic preset for new control experiments.

Return type:

FusionCompiler

Parameters:

  • bitstream_length (int)

  • seed (int)

  • lif_dt (float)

  • lif_resistance (float)

  • lif_refractory_period (int)

static traceable_runtime_kwargs(*, runtime_backend='auto')[source]

Recommended controller kwargs for traceable runtime loops.

Return type:

Dict[str, Any]

Parameters:

runtime_backend (str)

compile(net, firing_mode='binary', firing_margin=0.05, *, allow_inhibitor=False, validate_topology=False, strict_topology=False)[source]

Compile the Petri Net into sc_neurocore artifacts.

Parameters:

  • net (compiled StochasticPetriNet.)

  • firing_mode ("binary" (default) or "fractional".)

  • firing_margin (margin for fractional firing (ignored in binary mode).)

  • allow_inhibitor (enable inhibitor arc compilation.)

  • validate_topology (run topology diagnostics during compile.)

  • strict_topology (raise if topology diagnostics detect issues.)

  • Steps –

    1. Extract dense W_in (nT x nP) and W_out (nP x nT).

    2. Create one LIF neuron per transition (pure threshold comparator).

    3. Pre-encode weight matrices as packed uint64 bitstreams.

    4. Return CompiledNet with all artifacts.

Return type:

CompiledNet

SCPN Controller

Neuro-Symbolic Controller — oracle + SC dual paths.

Loads a .scpnctl.json artifact and provides deterministic step(obs, k) → ControlAction with JSONL logging.

class scpn_fusion.scpn.controller.NeuroSymbolicController(artifact, seed_base, targets, scales, sc_n_passes=8, sc_bitflip_rate=0.0, sc_binary_margin=None, sc_antithetic=True, enable_oracle_diagnostics=True, feature_axes=None, runtime_profile='adaptive', runtime_backend='auto', rust_backend_min_problem_size=1, sc_antithetic_chunk_size=2048)[source]

Bases: NeuroSymbolicControllerFeaturesMixin, NeuroSymbolicControllerBackendMixin

Reference controller with oracle float and stochastic paths.

Parameters:

  • artifact (loaded .scpnctl.json artifact.)

  • seed_base (64-bit base seed for deterministic stochastic execution.)

  • targets (control setpoint targets.)

  • scales (normalisation scales.)

  • sc_n_passes (int)

  • sc_bitflip_rate (float)

  • sc_binary_margin (Optional[float])

  • sc_antithetic (bool)

  • enable_oracle_diagnostics (bool)

  • feature_axes (Optional[Sequence[FeatureAxisSpec]])

  • runtime_profile (str)

  • runtime_backend (str)

  • rust_backend_min_problem_size (int)

  • sc_antithetic_chunk_size (int)

reset()[source]

Restore initial marking and zero previous actions.

Return type:

None

property runtime_backend_name: str
property runtime_profile_name: str
property marking: List[float]
step(obs, k, log_path=None)[source]

Execute one control tick.

Return type:

ControlAction

Parameters:
Steps:

  1. extract_features(obs) → 4 unipolar features

  2. _inject_places(features)

  3. _oracle_step() — float path (optional)

  4. _sc_step(k) — deterministic stochastic path

  5. _decode_actions() — gain × differencing, slew + abs clamp

  6. Optional JSONL logging

step_traceable(obs_vector, k, log_path=None)[source]

Execute one control tick from a fixed-order observation vector.

The vector order is self._axis_obs_keys. This avoids per-step key lookups/dict allocation in tight control loops.

Return type:

ndarray[Any, dtype[float64]]

Parameters:

Compilation Artifacts

SCPN Controller Artifact (.scpnctl.json) loader / saver.

Defines the Artifact dataclass that mirrors the JSON schema sections (meta, topology, weights, readout, initial_state) and provides lightweight validation on load.

class scpn_fusion.scpn.artifact.FixedPoint(data_width, fraction_bits, signed)[source]

Bases: object

Parameters:

  • data_width (int)

  • fraction_bits (int)

  • signed (bool)

data_width: int
fraction_bits: int
signed: bool
class scpn_fusion.scpn.artifact.SeedPolicy(id, hash_fn, rng_family)[source]

Bases: object

Parameters:
id: str
hash_fn: str
rng_family: str
class scpn_fusion.scpn.artifact.CompilerInfo(name, version, git_sha)[source]

Bases: object

Parameters:
name: str
version: str
git_sha: str
class scpn_fusion.scpn.artifact.ArtifactMeta(artifact_version, name, dt_control_s, stream_length, fixed_point, firing_mode, seed_policy, created_utc, compiler, notes=None)[source]

Bases: object

Parameters:
artifact_version: str
name: str
dt_control_s: float
stream_length: int
fixed_point: FixedPoint
firing_mode: str
seed_policy: SeedPolicy
created_utc: str
compiler: CompilerInfo
notes: Optional[str] = None
class scpn_fusion.scpn.artifact.PlaceSpec(id, name)[source]

Bases: object

Parameters:
id: int
name: str
class scpn_fusion.scpn.artifact.TransitionSpec(id, name, threshold, margin=None, delay_ticks=0)[source]

Bases: object

Parameters:
id: int
name: str
threshold: float
margin: Optional[float] = None
delay_ticks: int = 0
class scpn_fusion.scpn.artifact.Topology(places, transitions)[source]

Bases: object

Parameters:
places: List[PlaceSpec]
transitions: List[TransitionSpec]
class scpn_fusion.scpn.artifact.WeightMatrix(shape, data)[source]

Bases: object

Parameters:
shape: List[int]
data: List[float]
class scpn_fusion.scpn.artifact.PackedWeights(shape, data_u64)[source]

Bases: object

Parameters:
shape: List[int]
data_u64: List[int]
class scpn_fusion.scpn.artifact.PackedWeightsGroup(words_per_stream, w_in_packed, w_out_packed=None)[source]

Bases: object

Parameters:
words_per_stream: int
w_in_packed: PackedWeights
w_out_packed: Optional[PackedWeights] = None
class scpn_fusion.scpn.artifact.Weights(w_in, w_out, packed=None)[source]

Bases: object

Parameters:
w_in: WeightMatrix
w_out: WeightMatrix
packed: Optional[PackedWeightsGroup] = None
class scpn_fusion.scpn.artifact.ActionReadout(id, name, pos_place, neg_place)[source]

Bases: object

Parameters:
id: int
name: str
pos_place: int
neg_place: int
class scpn_fusion.scpn.artifact.Readout(actions, gains, abs_max, slew_per_s)[source]

Bases: object

Parameters:
actions: List[ActionReadout]
gains: List[float]
abs_max: List[float]
slew_per_s: List[float]
class scpn_fusion.scpn.artifact.PlaceInjection(place_id, source, scale, offset, clamp_0_1)[source]

Bases: object

Parameters:
place_id: int
source: str
scale: float
offset: float
clamp_0_1: bool
class scpn_fusion.scpn.artifact.InitialState(marking, place_injections)[source]

Bases: object

Parameters:
marking: List[float]
place_injections: List[PlaceInjection]
class scpn_fusion.scpn.artifact.Artifact(meta, topology, weights, readout, initial_state)[source]

Bases: object

Full SCPN controller artifact (.scpnctl.json).

Parameters:
meta: ArtifactMeta
topology: Topology
weights: Weights
readout: Readout
initial_state: InitialState
property nP: int
property nT: int
exception scpn_fusion.scpn.artifact.ArtifactValidationError[source]

Bases: ValueError, FusionCoreError

Raised when an artifact fails lightweight validation.

scpn_fusion.scpn.artifact.encode_u64_compact(data_u64)[source]

Public compact codec helper for deterministic uint64 payload encoding.

Return type:

Dict[str, Any]

Parameters:

data_u64 (List[int])

scpn_fusion.scpn.artifact.decode_u64_compact(encoded)[source]

Public compact codec helper for deterministic uint64 payload decoding.

Return type:

List[int]

Parameters:

encoded (Dict[str, Any])

scpn_fusion.scpn.artifact.load_artifact(path)[source]

Parse a .scpnctl.json file into an Artifact dataclass.

Return type:

Artifact

Parameters:

path (str | Path)

scpn_fusion.scpn.artifact.get_artifact_json_schema()[source]

Return a formal JSON schema for .scpnctl.json artifacts.

Return type:

Dict[str, Any]

scpn_fusion.scpn.artifact.save_artifact(artifact, path, compact_packed=False)[source]

Serialize an Artifact to indented JSON.

Return type:

None

Parameters: