# SPDX-License-Identifier: AGPL-3.0-or-later | Commercial license available
# © Concepts 1996–2026 Miroslav Šotek. All rights reserved.
# © Code 2020–2026 Miroslav Šotek. All rights reserved.
# ORCID: 0009-0009-3560-0851
# Contact: www.anulum.li | protoscience@anulum.li
# SCPN Fusion Core — Neuro-Symbolic Logic Compiler
"""
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.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from scpn_fusion.exceptions import FusionCoreError as _FusionCoreError
from typing import Any, Dict, List, Optional, Union
from scpn_fusion.scpn.artifact_codec import (
decode_u64_compact_payload,
encode_u64_compact_payload,
)
from scpn_fusion.scpn.artifact_validation import validate_artifact
ARTIFACT_SCHEMA_VERSION = "1.0.0"
MAX_PACKED_WORDS = 10_000_000
MAX_DECOMPRESSED_BYTES = MAX_PACKED_WORDS * 8
MAX_COMPRESSED_BYTES = 50_000_000
# ── Sub-structures ──────────────────────────────────────────────────────────
[docs]
@dataclass
class FixedPoint:
data_width: int
fraction_bits: int
signed: bool
[docs]
@dataclass
class SeedPolicy:
id: str
hash_fn: str
rng_family: str
[docs]
@dataclass
class CompilerInfo:
name: str
version: str
git_sha: str
[docs]
@dataclass
class PlaceSpec:
id: int
name: str
[docs]
@dataclass
class TransitionSpec:
id: int
name: str
threshold: float
margin: Optional[float] = None
delay_ticks: int = 0
[docs]
@dataclass
class Topology:
places: List[PlaceSpec]
transitions: List[TransitionSpec]
[docs]
@dataclass
class WeightMatrix:
shape: List[int] # [rows, cols]
data: List[float] # row-major
[docs]
@dataclass
class PackedWeights:
shape: List[int] # [rows, cols, words]
data_u64: List[int]
[docs]
@dataclass
class PackedWeightsGroup:
words_per_stream: int
w_in_packed: PackedWeights
w_out_packed: Optional[PackedWeights] = None
[docs]
@dataclass
class Weights:
w_in: WeightMatrix
w_out: WeightMatrix
packed: Optional[PackedWeightsGroup] = None
[docs]
@dataclass
class ActionReadout:
id: int
name: str
pos_place: int
neg_place: int
[docs]
@dataclass
class Readout:
actions: List[ActionReadout]
gains: List[float]
abs_max: List[float]
slew_per_s: List[float]
[docs]
@dataclass
class PlaceInjection:
place_id: int
source: str
scale: float
offset: float
clamp_0_1: bool
[docs]
@dataclass
class InitialState:
marking: List[float]
place_injections: List[PlaceInjection]
# ── Artifact ────────────────────────────────────────────────────────────────
[docs]
@dataclass
class Artifact:
"""Full SCPN controller artifact (``.scpnctl.json``)."""
meta: ArtifactMeta
topology: Topology
weights: Weights
readout: Readout
initial_state: InitialState
@property
def nP(self) -> int:
return len(self.topology.places)
@property
def nT(self) -> int:
return len(self.topology.transitions)
# ── Validation ──────────────────────────────────────────────────────────────
[docs]
class ArtifactValidationError(ValueError, _FusionCoreError):
"""Raised when an artifact fails lightweight validation."""
def _encode_u64_compact(data_u64: List[int]) -> Dict[str, Any]:
"""Encode uint64 list as zlib-compressed base64 little-endian payload."""
return encode_u64_compact_payload(data_u64)
def _decode_u64_compact(encoded: Dict[str, Any]) -> List[int]:
"""Decode compact uint64 payload generated by ``_encode_u64_compact``."""
return decode_u64_compact_payload(
encoded,
error_type=ArtifactValidationError,
max_packed_words=MAX_PACKED_WORDS,
max_compressed_bytes=MAX_COMPRESSED_BYTES,
max_decompressed_bytes=MAX_DECOMPRESSED_BYTES,
)
[docs]
def encode_u64_compact(data_u64: List[int]) -> Dict[str, Any]:
"""Public compact codec helper for deterministic uint64 payload encoding."""
return _encode_u64_compact(list(map(int, data_u64)))
[docs]
def decode_u64_compact(encoded: Dict[str, Any]) -> List[int]:
"""Public compact codec helper for deterministic uint64 payload decoding."""
return _decode_u64_compact(encoded)
def _validate(artifact: Artifact) -> None:
"""Lightweight checks: required fields, ranges, shape consistency."""
validate_artifact(artifact, error_type=ArtifactValidationError)
# ── Load / Save ─────────────────────────────────────────────────────────────
[docs]
def load_artifact(path: Union[str, Path]) -> Artifact:
"""Parse a ``.scpnctl.json`` file into an ``Artifact`` dataclass."""
with open(path, "r", encoding="utf-8") as f:
obj = json.load(f)
# Meta
m = obj["meta"]
meta = ArtifactMeta(
artifact_version=m["artifact_version"],
name=m["name"],
dt_control_s=m["dt_control_s"],
stream_length=m["stream_length"],
fixed_point=FixedPoint(
data_width=m["fixed_point"]["data_width"],
fraction_bits=m["fixed_point"]["fraction_bits"],
signed=m["fixed_point"]["signed"],
),
firing_mode=m["firing_mode"],
seed_policy=SeedPolicy(
id=m["seed_policy"]["id"],
hash_fn=m["seed_policy"]["hash_fn"],
rng_family=m["seed_policy"]["rng_family"],
),
created_utc=m["created_utc"],
compiler=CompilerInfo(
name=m["compiler"]["name"],
version=m["compiler"]["version"],
git_sha=m["compiler"]["git_sha"],
),
notes=m.get("notes"),
)
# Topology
places = [PlaceSpec(id=p["id"], name=p["name"]) for p in obj["topology"]["places"]]
transitions = [
TransitionSpec(
id=t["id"],
name=t["name"],
threshold=t["threshold"],
margin=t.get("margin"),
delay_ticks=t.get("delay_ticks", 0),
)
for t in obj["topology"]["transitions"]
]
topology = Topology(places=places, transitions=transitions)
# Weights
w_in = WeightMatrix(
shape=obj["weights"]["w_in"]["shape"],
data=list(map(float, obj["weights"]["w_in"]["data"])),
)
w_out = WeightMatrix(
shape=obj["weights"]["w_out"]["shape"],
data=list(map(float, obj["weights"]["w_out"]["data"])),
)
packed = None
if "packed" in obj["weights"]:
pw = obj["weights"]["packed"]
w_in_obj = pw["w_in_packed"]
if "data_u64" in w_in_obj:
w_in_data = list(map(int, w_in_obj["data_u64"]))
else:
w_in_data = _decode_u64_compact(w_in_obj)
pw_in = PackedWeights(
shape=w_in_obj["shape"],
data_u64=w_in_data,
)
pw_out = None
if "w_out_packed" in pw:
w_out_obj = pw["w_out_packed"]
if "data_u64" in w_out_obj:
w_out_data = list(map(int, w_out_obj["data_u64"]))
else:
w_out_data = _decode_u64_compact(w_out_obj)
pw_out = PackedWeights(
shape=w_out_obj["shape"],
data_u64=w_out_data,
)
packed = PackedWeightsGroup(
words_per_stream=int(pw["words_per_stream"]),
w_in_packed=pw_in,
w_out_packed=pw_out,
)
weights = Weights(w_in=w_in, w_out=w_out, packed=packed)
# Readout
actions = [
ActionReadout(
id=a["id"],
name=a["name"],
pos_place=a["pos_place"],
neg_place=a["neg_place"],
)
for a in obj["readout"]["actions"]
]
readout = Readout(
actions=actions,
gains=obj["readout"]["gains"]["per_action"],
abs_max=obj["readout"]["limits"]["per_action_abs_max"],
slew_per_s=obj["readout"]["limits"]["slew_per_s"],
)
# Initial state
injections = [
PlaceInjection(
place_id=inj["place_id"],
source=inj["source"],
scale=inj["scale"],
offset=inj["offset"],
clamp_0_1=inj["clamp_0_1"],
)
for inj in obj["initial_state"]["place_injections"]
]
initial_state = InitialState(
marking=list(map(float, obj["initial_state"]["marking"])),
place_injections=injections,
)
artifact = Artifact(
meta=meta,
topology=topology,
weights=weights,
readout=readout,
initial_state=initial_state,
)
_validate(artifact)
return artifact
[docs]
def get_artifact_json_schema() -> Dict[str, Any]:
"""Return a formal JSON schema for ``.scpnctl.json`` artifacts."""
return {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "SCPN Controller Artifact",
"type": "object",
"required": ["meta", "topology", "weights", "readout", "initial_state"],
"properties": {
"meta": {
"type": "object",
"required": ["artifact_version", "name", "stream_length"],
"properties": {
"artifact_version": {"type": "string"},
"name": {"type": "string"},
"dt_control_s": {"type": "number"},
"stream_length": {"type": "integer"},
"fixed_point": {
"type": "object",
"properties": {
"data_width": {"type": "integer"},
"fraction_bits": {"type": "integer"},
"signed": {"type": "boolean"},
},
},
},
},
"topology": {
"type": "object",
"properties": {
"places": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
},
},
},
"transitions": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"threshold": {"type": "number"},
},
},
},
},
},
"weights": {
"type": "object",
"properties": {
"w_in": {"$ref": "#/definitions/weight_matrix"},
"w_out": {"$ref": "#/definitions/weight_matrix"},
"packed": {
"type": "object",
"properties": {
"shape": {"type": "array", "items": {"type": "integer"}},
"data_b64": {"type": "string"},
},
},
},
},
"readout": {
"type": "object",
"properties": {
"actions": {"type": "array"},
"gains": {"type": "object"},
"limits": {"type": "object"},
},
},
"initial_state": {
"type": "object",
"properties": {
"marking": {"type": "array", "items": {"type": "number"}},
"place_injections": {"type": "array"},
},
},
},
"definitions": {
"weight_matrix": {
"type": "object",
"properties": {
"shape": {"type": "array", "items": {"type": "integer"}},
"data": {"type": "array", "items": {"type": "number"}},
},
},
},
}
[docs]
def save_artifact(
artifact: Artifact,
path: Union[str, Path],
compact_packed: bool = False,
) -> None:
"""Serialize an ``Artifact`` to indented JSON."""
def _weight_matrix_dict(wm: WeightMatrix) -> Dict[str, Any]:
return {"shape": wm.shape, "data": wm.data}
packed_dict: Optional[Dict[str, Any]] = None
if artifact.weights.packed is not None:
pg = artifact.weights.packed
if compact_packed:
pw_in_d = {"shape": pg.w_in_packed.shape}
pw_in_d.update(_encode_u64_compact(pg.w_in_packed.data_u64))
else:
pw_in_d = {"shape": pg.w_in_packed.shape, "data_u64": pg.w_in_packed.data_u64}
pw_out_d = None
if pg.w_out_packed is not None:
if compact_packed:
pw_out_d = {"shape": pg.w_out_packed.shape}
pw_out_d.update(_encode_u64_compact(pg.w_out_packed.data_u64))
else:
pw_out_d = {
"shape": pg.w_out_packed.shape,
"data_u64": pg.w_out_packed.data_u64,
}
packed_dict = {
"words_per_stream": pg.words_per_stream,
"w_in_packed": pw_in_d,
}
if pw_out_d is not None:
packed_dict["w_out_packed"] = pw_out_d
obj: Dict[str, Any] = {
"meta": {
"artifact_version": artifact.meta.artifact_version,
"name": artifact.meta.name,
"dt_control_s": artifact.meta.dt_control_s,
"stream_length": artifact.meta.stream_length,
"fixed_point": {
"data_width": artifact.meta.fixed_point.data_width,
"fraction_bits": artifact.meta.fixed_point.fraction_bits,
"signed": artifact.meta.fixed_point.signed,
},
"firing_mode": artifact.meta.firing_mode,
"seed_policy": {
"id": artifact.meta.seed_policy.id,
"hash_fn": artifact.meta.seed_policy.hash_fn,
"rng_family": artifact.meta.seed_policy.rng_family,
},
"created_utc": artifact.meta.created_utc,
"compiler": {
"name": artifact.meta.compiler.name,
"version": artifact.meta.compiler.version,
"git_sha": artifact.meta.compiler.git_sha,
},
},
"topology": {
"places": [{"id": p.id, "name": p.name} for p in artifact.topology.places],
"transitions": [
{
"id": t.id,
"name": t.name,
"threshold": t.threshold,
**({"margin": t.margin} if t.margin is not None else {}),
"delay_ticks": int(t.delay_ticks),
}
for t in artifact.topology.transitions
],
},
"weights": {
"w_in": _weight_matrix_dict(artifact.weights.w_in),
"w_out": _weight_matrix_dict(artifact.weights.w_out),
},
"readout": {
"actions": [
{
"id": a.id,
"name": a.name,
"pos_place": a.pos_place,
"neg_place": a.neg_place,
}
for a in artifact.readout.actions
],
"gains": {"per_action": artifact.readout.gains},
"limits": {
"per_action_abs_max": artifact.readout.abs_max,
"slew_per_s": artifact.readout.slew_per_s,
},
},
"initial_state": {
"marking": artifact.initial_state.marking,
"place_injections": [
{
"place_id": inj.place_id,
"source": inj.source,
"scale": inj.scale,
"offset": inj.offset,
"clamp_0_1": inj.clamp_0_1,
}
for inj in artifact.initial_state.place_injections
],
},
}
if artifact.meta.notes is not None:
obj["meta"]["notes"] = artifact.meta.notes
if packed_dict is not None:
obj["weights"]["packed"] = packed_dict
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, indent=2)
f.write("\n")