petri_net_nn.adapter

adapter

Config-driven adapter from external data into the framework.

Each scenario — BPMN process, biological cascade, network protocol, manufacturing line — needs roughly the same pipeline: load a Petri net, load traces, train, optionally extract rules, optionally score anomalies. The boilerplate is identical across scenarios, only the data differs.

This module reads a TOML config file describing the data sources and training parameters, and returns a ScenarioContext that exposes the standard pipeline as methods. A scenario then becomes one config file plus a data file (BPMN or XES), with no per-scenario glue code.

Config schema (TOML):

[scenario]
name = "..."
description = "..."

# Net source — exactly one of:
[net]
source = "bpmn_file"
path = "process.bpmn"
# OR
[net]
source = "inline"
# then declare places/transitions/arcs as TOML tables below

[[net.places]]
id = "p_x"
tokens = 1            # optional, default 0
label = "..."         # optional

[[net.transitions]]
id = "t_x"
label = "..."

[[net.arcs]]
src = "p_x"
dst = "t_x"

# Trace source — exactly one of:
[traces]
source = "xes_file"
path = "traces.xes"
# OR
[traces]
source = "inline"
# with [[traces.inline]] entries below

[[traces.inline]]
attributes = { signal_strength = "0.9" }
events = ["t_a", "t_b"]

# Map per-trace attribute (or constant) onto a place's input
# activation. Each key is a place id; each value is either
# { attribute = "name" } or { constant = 0.5 }.
[training.input_marking]
p_signal = { attribute = "signal_strength" }

[training]
steps = 1500
lr = 0.1
sharpness = 1.0
firing = "sigmoid"          # or "ste"
routing = "independent"     # or "softmax"
num_steps = 0
seed = 0

Missing optional sections fall back to library defaults.

ScenarioContext dataclass

ScenarioContext(name, description, net, traces, training, input_marking_spec, input_values_spec=dict(), config=dict(), config_dir=Path())

Self-contained, config-driven scenario.

Use compile() + train() for the standard pipeline, or attribute_to_marking / anomaly_score if you want to mix in your own scoring code. Carrying the original config dict lets consumers read scenario-specific extras (e.g. anomaly-generator settings) without re-parsing the file.

attribute_to_values

attribute_to_values(trace)

Coloured-Petri-net analogue of attribute_to_marking: returns the scalar value carried by the token at each source place. Reads the same spec form ({attribute = "..."} or {constant = ...}) from the [training.input_values] TOML section.

Source code in petri_net_nn/adapter.py

def attribute_to_values(self, trace: XESTrace) -> dict[str, float]:
    """Coloured-Petri-net analogue of ``attribute_to_marking``:
    returns the scalar value carried by the token at each source
    place. Reads the same spec form (``{attribute = "..."}`` or
    ``{constant = ...}``) from the
    ``[training.input_values]`` TOML section."""
    return _resolve_attribute_spec(
        trace, self.input_values_spec, "input_values"
    )

load_scenario

load_scenario(config_path)

Read a TOML scenario config and produce a ScenarioContext.

Relative path entries inside the config are resolved against the config file's directory, so configs can sit next to their data files without absolute paths.

Source code in petri_net_nn/adapter.py

def load_scenario(config_path: str | Path) -> ScenarioContext:
    """Read a TOML scenario config and produce a ``ScenarioContext``.

    Relative ``path`` entries inside the config are resolved against
    the config file's directory, so configs can sit next to their
    data files without absolute paths."""
    path = Path(config_path)
    if not path.exists():
        raise FileNotFoundError(path)
    with path.open("rb") as f:
        config = tomllib.load(f)

    config_dir = path.parent

    scenario = config.get("scenario", {})
    name = scenario.get("name", path.stem)
    description = scenario.get("description", "")

    # Traces load first because `[net] source = "discover"` needs them
    # as input to the Inductive Miner. For every other net source the
    # two are independent, so the reordering is transparent.
    traces = _load_traces(config.get("traces", {}), config_dir)
    net = _load_net(config.get("net", {}), config_dir, traces=traces)
    training_dict = config.get("training", {}) or {}
    input_marking_spec = training_dict.pop("input_marking", {}) or {}
    input_values_spec = training_dict.pop("input_values", {}) or {}
    training = TrainingConfig(**{k: v for k, v in training_dict.items() if k in TrainingConfig.__dataclass_fields__})

    return ScenarioContext(
        name=name,
        description=description,
        net=net,
        traces=traces,
        training=training,
        input_marking_spec=input_marking_spec,
        input_values_spec=input_values_spec,
        config=config,
        config_dir=config_dir,
    )