From ebe47c45c7b98bc6c30669e5cc08697d64c4850f Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:35:40 +0300 Subject: [PATCH 01/19] added a local gitignore --- libs/@hashintel/petrinaut-opt/.gitignore | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/.gitignore diff --git a/libs/@hashintel/petrinaut-opt/.gitignore b/libs/@hashintel/petrinaut-opt/.gitignore new file mode 100644 index 00000000000..361e2ace7bd --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/.gitignore @@ -0,0 +1,9 @@ +# Python +__pycache__/ +*.pyc + +# Desktop store +*.DS_Store + +# Virtual environments +uv.lock \ No newline at end of file From 9f8efa08bed552c9db0699e5ec657ec55ec73347 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:36:06 +0300 Subject: [PATCH 02/19] added pyproject for dependencies and python version --- libs/@hashintel/petrinaut-opt/pyproject.toml | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/pyproject.toml diff --git a/libs/@hashintel/petrinaut-opt/pyproject.toml b/libs/@hashintel/petrinaut-opt/pyproject.toml new file mode 100644 index 00000000000..8a2d1baaaa1 --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/pyproject.toml @@ -0,0 +1,11 @@ +[project] +name = "petrinaut-optimization" +version = "0.1.0" +description = "Petrinaut parameter and initial state optimisation using optuna" +readme = "README.md" +requires-python = ">=3.10.20" +dependencies = [ + "fastapi>=0.128.8", + "optuna>=3.6", + "uvicorn[standard]>=0.39.0", +] From 50f3cf6c121b59cd391e7e0a9939d228a4dff1d1 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:36:19 +0300 Subject: [PATCH 03/19] added src module init --- libs/@hashintel/petrinaut-opt/src/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 libs/@hashintel/petrinaut-opt/src/__init__.py diff --git a/libs/@hashintel/petrinaut-opt/src/__init__.py b/libs/@hashintel/petrinaut-opt/src/__init__.py new file mode 100644 index 00000000000..e69de29bb2d From 50e8880b12528cdbc8d00d3a46b43a1c9c1bac7b Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:37:03 +0300 Subject: [PATCH 04/19] created python wrapper for petrinaut-cli unix server requests based on a petrinaut specification --- .../petrinaut-opt/src/petrinaut_client.py | 209 ++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/src/petrinaut_client.py diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py new file mode 100644 index 00000000000..e21bd43606e --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Python Wrapper for Petri Net execution CLI. + +Inputs: all parameters and initial states to a Petri Net (both fixed and optimizable) +Output: metric computed inside Petrinaut CLI at the end of the Petri Net execution + + +Example: + python3 petrinaut_cli.py --produce-a 0.5 --produce-b 0.5 \ + --initial-tokens 100 --policy balance +""" + +import logging + +import re +import uuid +import json +import socket +import argparse +import subprocess + +from typing import Optional, Dict, Union, Sequence +from pydantic import BaseModel, Field + +log = logging.getLogger("pn_client") + +# The Petrinaut execution program parameters +DEFAULT_MODEL = "SIR" +DEFAULT_STRUCTURE = "" +DEFAULT_METRIC = "Infected Fraction" +DEFAULT_STEPS = 100 +DEFAULT_TIMESTEP = 0.1 +DEFAULT_SEED = 1234 +DEFAULT_STORE = ["metric"] +DEFAULT_OUTPATH = "" +DEFAULT_EVAL_TIMEOUT = None + +# Regex for extracting number from Petrinaut CLI +_NUMBER = re.compile(r"[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?") + + +# ───────────────────────────────────────────────────────────────────────────── +# Petrinaut CLI Python wrapper specification +# ───────────────────────────────────────────────────────────────────────────── + +class PetrinautModelSpec(BaseModel): + name: str = Field(default=DEFAULT_MODEL, description="Petri Net class name") + structure: str = Field(default=DEFAULT_STRUCTURE, description="Filepath of Petri Net model structure (places, transitions, arcs etc.)") + metric: str = Field(default=DEFAULT_METRIC, description="Metric name that will be computed at the end of execution") + steps: Optional[int] = Field(default=DEFAULT_STEPS, description="Number of steps in a single execution") + dt: Optional[float] = Field(default=DEFAULT_TIMESTEP, description="Step size for dynamics discretisation in a single execution") + seed: Optional[int] = Field(default=DEFAULT_SEED, description="Random number generator seed (fixed -> deterministic output)") + outpath: Optional[str] = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") + store: Optional[Sequence[str]] = Field(default=DEFAULT_STORE, description="Quantities to store/print inside execution") + eval_timeout: Optional[float] = Field(default=DEFAULT_EVAL_TIMEOUT, description="timeout threshold for CLI eval") + + +# ───────────────────────────────────────────────────────────────────────────── +# Petrinaut CLI Python wrapper +# ───────────────────────────────────────────────────────────────────────────── + +class PetrinautModel: + """Python wrapper for Petrinaut execution CLI.""" + + def __init__( + self, + pn_spec: PetrinautModelSpec, + **kwargs + ) -> None: + self.name = pn_spec.name + self.structure = pn_spec.structure + self.metric = pn_spec.metric + self.steps = pn_spec.steps + self.dt = pn_spec.dt + self.seed = pn_spec.seed + # The following are currently unused + # they are left here for future releases + self.outpath = pn_spec.outpath + self.store = pn_spec.store + self.eval_timeout = pn_spec.eval_timeout + + # Connect to Petrinaut-cli socket + self.stream = self.connect() + + def connect(self): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect("/tmp/petrinaut.sock") + stream = sock.makefile("rwb") + return stream + + # ── UNIX request submission ───────────────────────────────────────────────────────── + def _build_payload(self, params:dict, init_states:dict) -> Dict[str,Union[str,Dict]]: + + return { + "id": str(uuid.uuid4()), + "method": "run", + "params": { + "parameters": params, + "initialState": init_states, + "metrics": [self.metric], + "maxSteps": self.steps, + "dt": self.dt, + "seed": self.seed + } + } + + + def run(self, params:dict, init_states:dict): + """Submits request to Petrinaut-cli UNIT socket + + Args: + params (dict): All parameters (optimized and fixed) to Petrinaut CLI + init_states (dict): All initial states (optimized and fixed) to Petrinaut CLI + + Returns: + objective (float): The metric evaluated from petrinaut-cli execution + """ + + # Build payload first + payload = self._build_payload( + params=params, + init_states=init_states + ) + # Write to stream + self.stream.write((json.dumps(payload) + "\n").encode()) + self.stream.flush() + # Get Petrinaut-cli response + response = json.loads(self.stream.readline()) + + if "error" in response: + raise RuntimeError(response["error"]["message"]) + + # Return the objective (metric) from the execution + objective = response["result"]["metrics"][self.metric] + return objective + + # ── CLI invokation ───────────────────────────────────────────────────────── + def _cli_build_command(self, inputs: dict) -> list: + """Turn a {name: value} dict into the full argv list.""" + cmd = list(self.command) + for name, value in inputs.items(): + # Default flag per input: "initial_tokens" -> "--initial-tokens". + cmd += [f"--{name.replace('_', '-')}", str(value)] + return cmd + + def cli_run(self, inputs: dict): + """Invokes the Petrinaut Cli + + Args: + inputs (dict): All inputs (optimized and fixed) to Petrinaut + + Returns: + proc (CompletedProcess): A completed subprocess run + """ + # Build the CLI command invocation as a string + cmd = self._cli_build_command(inputs) + + # Call Petrinaut cli here + proc = subprocess.run( + cmd, capture_output=True, text=True, + timeout=self.eval_timeout, check=True, + ) + return proc + + @staticmethod + def _parse_output(stdout: str) -> float: + """Extract the objective: the last number printed on stdout.""" + matches = _NUMBER.findall(stdout) + if not matches: + raise ValueError(f"no number found in CLI output:\n{stdout!r}") + return float(matches[-1]) + + +if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(message)s", + datefmt="%H:%M:%S", + ) + + p = argparse.ArgumentParser(description="Toy Petri-net execution function.") + p.add_argument("--infection_rate", type=float, required=True, + help="infection rate in SIR Petri Net") + p.add_argument("--recovery_rate", type=float, required=True, + help="recovery rate in SIR Petri Net") + p.add_argument("--susceptible", type=int, required=True, + help="Number of susceptible individuals in SIR Petri Net") + p.add_argument("--infected", type=int, required=True, + help="Number of infected individuals in SIR Petri Net") + p.add_argument("--recovered", type=int, required=True, + help="Number of recovered individuals in SIR Petri Net") + p.add_argument("--steps", type=int, default=1000, + help="simulation steps (fixed; not optimized)") + p.add_argument("--seed", type=int, default=0, + help="RNG seed (fixed -> deterministic output)") + args = p.parse_args() + + # Create the petrinaut execution specification + pn_spec = PetrinautModelSpec() + # Build the Petri net from the client spec. + petrinet_model = PetrinautModel(pn_spec) + # Run petrinaut + result = petrinet_model.run( + params = {"infection_rate": args.infection_rate, "recovery_rate": args.recovery_rate}, + init_states = {"Susceptible": args.susceptible} + ) + # Read metric value from petrinaut execution + objective = result["metrics"]["Infected Fraction"] + log.info('objective',objective) From a036c57e305ab374438795f4e16cf164c3e6e984 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:37:45 +0300 Subject: [PATCH 05/19] created optuna-based python optimisation class for optimising petrinaut-cli params and init_states based on optimisation specification --- .../petrinaut-opt/src/petrinaut_optimizer.py | 439 ++++++++++++++++++ 1 file changed, 439 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py new file mode 100644 index 00000000000..0c43985a4f5 --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py @@ -0,0 +1,439 @@ +#!/usr/bin/env python3 +"""Black-box optimization of a CLI Petrinaut execution via Optuna. + +Wraps a command-line program that prints a single numeric objective to stdout. +`PetrinautOptimizer` proposes inputs, invokes the CLI, parses the result, and +reports it back to Optuna — maximising (or minimising) the output. + +Configure the DEFAULT_* constants below (or pass overrides to the constructor), +then either run this file directly for a one-off run, or instantiate +`PetrinautOptimizer` from another module (see optimization_api.py). +""" + +from __future__ import annotations + +import logging +import sys +import queue +import json +import threading +import asyncio + +from enum import Enum +from datetime import datetime +from dataclasses import dataclass +from typing import Annotated, Optional, Sequence, Literal, Union +from pydantic import BaseModel, Field, model_validator +from fastapi import Request + +import optuna + +from src.petrinaut_client import PetrinautModel,PetrinautModelSpec + +log = logging.getLogger("pn_optimize") + +# ───────────────────────────────────────────────────────────────────────────── +# CONFIG — defaults for the optimizer specification; override via constructor args. +# ───────────────────────────────────────────────────────────────────────────── + +# Search space comprising of parameters and initial state. One entry per optimizable input. +# name : Optuna param name and (by default) flag name; +# type : "float" | "int" | "categorical" +# float/int: low, high (+ optional step, log=True); categorical: choices=[...] +# Specification of all parameters + +@dataclass(frozen=True) +class Bounds: + low: float + high: float + log: bool = False + +BOUNDS: dict[str, dict[str, Bounds]] = { + "parameters":{ + "infection_rate": Bounds(0.01, 10.0), + "recovery_rate": Bounds(0.01, 10.0) + }, + "initial_states":{ + "Susceptible": Bounds(0.0, 1000), + "Infected": Bounds(0.0, 1000), + "Recovered": Bounds(0.0, 1000), + } +} + +class Parameters(BaseModel): + infection_rate: float | None = None + recovery_rate: float | None = None + +class InitialStates(BaseModel): + Susceptible: float | None = None + Infected: float | None = None + Recovered: float | None = None + + +# Hard-coded allowable optuna samplers +SAMPLERS = { + "tpe": optuna.samplers.TPESampler, + "random": optuna.samplers.RandomSampler, +} +# Enum for optuna sampler name +SamplerName = Enum( + "SamplerName", + {name.upper(): name for name in SAMPLERS}, + type=str, +) +# input space sampling algorithm - default: Tree-structured Parzen Estimator +DEFAULT_SAMPLER = "tpe" +# maximise the CLI's output +DEFAULT_DIRECTION = "maximize" +# evaluations per run +DEFAULT_N_TRIALS = 100 + + +class OptimizationSpec(BaseModel): + parameters: Parameters = Parameters() + initial_states: InitialStates = InitialStates() + study_name: Optional[str] = Field(default=None, description="Name of optimization study") + sampler: Optional[SamplerName] = Field(default=DEFAULT_SAMPLER, description="input sampling algorithm") + direction: Optional[Literal["maximize", "minimize"]] = Field(default=DEFAULT_DIRECTION, description="optimization direction") + n_trials: Optional[int] = Field(default=DEFAULT_N_TRIALS, description="number of evals to run") + + def fixed(self) -> dict[str, dict[str, float]]: + out: dict[str, dict[str, float]] = {} + for group_name in ("parameters", "initial_states"): + group = getattr(self, group_name) + out[group_name] = { + name: value for name, value in group if value is not None + } + return out + + @model_validator(mode="after") + def _check_bounds(self): + for group_name, fields in self.fixed().items(): + for name, value in fields.items(): + b = BOUNDS[group_name][name] + if not (b.low <= value <= b.high): + raise ValueError( + f"{group_name}.{name}={value} outside " + f"allowed range [{b.low}, {b.high}]" + ) + return self + +# ───────────────────────────────────────────────────────────────────────────── +# OptimizationModel +# ───────────────────────────────────────────────────────────────────────────── +_SENTINEL = object() + + +class PetrinautOptimizer: + """Optimize a Petrinaut CLI's stdout objective over a mixed input space.""" + + def __init__( + self, + opt_spec: OptimizationSpec, + pn_model: PetrinautModel, + **kwargs + ) -> None: + self.fixed = opt_spec.fixed() + self.params = opt_spec.parameters + self.init_states = opt_spec.initial_states + self.study_name = f"input_opt_{datetime.now().strftime('%m/%d/%Y-%H:%M:%S')}" + self.sampler = SAMPLERS[opt_spec.sampler.lower()](**kwargs) + self.direction = opt_spec.direction + self.n_trials = opt_spec.n_trials + self.study = optuna.create_study( + study_name=self.study_name, + storage=None, + load_if_exists=False, + direction=self.direction, + sampler=self.sampler, + ) + # Petrinaut model (Python wrapper) + self.pn_model = pn_model + # A lock so the same instance is never driven by two concurrent streams. + self.lock = threading.Lock() + + # ── search space ───────────────────────────────────────────────────────── + def suggest(self, trial: optuna.Trial) -> None: + """Ask Optuna for a value for one input, per its spec. + + Args: + trial (optuna.Trial): Optuna optimization trial + + Raises: + ValueError: Unknown input type found + + """ + values: dict[str, dict[str, float]] = {} + for group_name, fields in BOUNDS.items(): + values[group_name] = {} + for name, b in fields.items(): + if name in self.fixed[group_name]: + values[group_name][name] = self.fixed[group_name][name] + else: + values[group_name][name] = trial.suggest_float( + f"{group_name}.{name}", b.low, b.high, log=b.log + ) + return values + + + def objective(self, trial: optuna.Trial) -> float: + """One evaluation: suggest inputs, run the Petrinaut CLI, parse the result. + + Args: + trial (optuna.Trial): Single evaluation of objective function + + Raises: + optuna.TrialPruned: Early stopping optimization due to timeout + optuna.TrialPruned: Early stopping optimization due to process error + + Returns: + float: Evaluation of metric from Petrinaut execution + """ + # Suggest new set of params and init states + # while keeping fixed parameters fixed + params_and_init_states = self.suggest(trial) + params = params_and_init_states["parameters"] + init_states = params_and_init_states["initial_states"] + + try: + # Build and invoke the Petrinaut CLI command + value = self.pn_model.run( + params=params, + init_states=init_states + ) + except RuntimeError as r: + # This happens in case the Petrinaut execution takes too long to run + # as defined by the eval_timeout parameter in the PetrinautModelSpec + log.warning("trial %d runtime error %s — pruned", trial.number, str(r)) + raise optuna.TrialPruned() + except Exception as e: + # If Petrinaut execution fails for whatever other reason + # optuna prunes that run and continues the optimization + log.warning( + "trial %d failed — pruned\nstderr: %s", + trial.number, str(e), + ) + raise optuna.TrialPruned() + + # Log results + log.info("trial %d value=%.6g params=%s init_states=%s", trial.number, value, params, init_states) + + return value + + # ── runs for API ───────────────────────────────────────────────────────────────── + async def stream_all(self, request: Request, n_trials: int): + """Async generator yielding Server-side event frames, one per finished trial. + + Args: + request (Request): Optimization API generic request + n_trials (int): number of optimization steps + + Yields: + _type_: json line of data or event + """ + if not self.lock.acquire(blocking=False): + yield 'event: error\ndata: {"message": "already running"}\n\n' + return + + loop = asyncio.get_running_loop() + q: asyncio.Queue = asyncio.Queue() + stop_flag = threading.Event() + + # Callback for generating the payload from optuna optimize worker + def callback(study, trial): + params_and_init_states = {} + for k, v in trial.params.items(): + outer, inner = k.split('.', 1) + params_and_init_states.setdefault(outer, {})[inner] = v + payload = { + "step": trial.number, + "params": params_and_init_states["parameters"], + "init_states": params_and_init_states["initial_states"], + "metric": trial.value, + "state": trial.state.name, + } + # Pass payload to streamer + loop.call_soon_threadsafe(q.put_nowait, payload) + if stop_flag.is_set(): + study.stop() + + # Running the optuna optimize worker + def run(): + try: + self.study.optimize( + self.objective, n_trials=n_trials, callbacks=[callback] + ) + except Exception as exc: + # Pass error to streamer + loop.call_soon_threadsafe( + q.put_nowait, {"state": "ERROR", "message": str(exc)} + ) + finally: + # Pass error to streamer + loop.call_soon_threadsafe(q.put_nowait, _SENTINEL) + + threading.Thread(target=run, daemon=True).start() + + try: + while True: + item = await q.get() + if item is _SENTINEL: + yield "event: done\ndata: {}\n\n" + break + yield f"data: {json.dumps(item)}\n\n" + if await request.is_disconnected(): + stop_flag.set() + break + finally: + stop_flag.set() + self.lock.release() + + async def stream_best(self, request: Request, n_trials: int): + """Async generator yielding Server-side event frames, one per finished trial. + + Args: + request (Request): Optimization API generic request + n_trials (int): number of optimization steps + + Yields: + _type_: json line of data or event + """ + if not self.lock.acquire(blocking=False): + yield 'event: error\ndata: {"message": "already running"}\n\n' + return + + loop = asyncio.get_running_loop() + q: asyncio.Queue = asyncio.Queue() + stop_flag = threading.Event() + + # Callback for generating the payload from optuna optimize worker + def callback(study, trial): + # `best_params`/`best_value` raise if no trial has completed yet (e.g. + # the opening trials were all pruned). Skip emitting until there is a + # best to report, but still honour a pending stop request. + has_completed = any( + t.state == optuna.trial.TrialState.COMPLETE + for t in study.get_trials(deepcopy=False) + ) + if not has_completed: + if stop_flag.is_set(): + study.stop() + return + + best_params_and_init_states = {} + for k, v in study.best_params.items(): + outer, inner = k.split('.', 1) + best_params_and_init_states.setdefault(outer, {})[inner] = v + payload = { + "step": trial.number, + "params": best_params_and_init_states["parameters"], + "init_states": best_params_and_init_states["initial_states"], + "metric": study.best_value, + "state": "COMPLETE", + } + # Pass payload to streamer + loop.call_soon_threadsafe(q.put_nowait, payload) + if stop_flag.is_set(): + study.stop() + + # Running the optuna optimize worker + def run(): + try: + self.study.optimize( + self.objective, n_trials=n_trials, callbacks=[callback] + ) + except Exception as exc: + # Pass error to streamer + loop.call_soon_threadsafe( + q.put_nowait, {"state": "ERROR", "message": str(exc)} + ) + finally: + # Pass error to streamer + loop.call_soon_threadsafe(q.put_nowait, _SENTINEL) + + threading.Thread(target=run, daemon=True).start() + + try: + while True: + item = await q.get() + if item is _SENTINEL: + yield "event: done\ndata: {}\n\n" + break + yield f"data: {json.dumps(item)}\n\n" + if await request.is_disconnected(): + stop_flag.set() + break + finally: + stop_flag.set() + self.lock.release() + + # ── run for local testing /printing ───────────────────────────────────────────────────────────────── + def run_stream(self, study, objective, n_trials): + q = queue.Queue() + _DONE = object() + + def callback(study, trial): + params_and_init_states = {} + for k, v in trial.params.items(): + outer, inner = k.split('.', 1) + params_and_init_states.setdefault(outer, {})[inner] = v + q.put(( + trial.number, + params_and_init_states["parameters"], + params_and_init_states["initial_states"], + trial.value + )) + + def run(): + study.optimize(objective, n_trials=n_trials, callbacks=[callback]) + q.put(_DONE) + + threading.Thread(target=run, daemon=True).start() + while (item := q.get()) is not _DONE: + yield item + +# ───────────────────────────────────────────────────────────────────────────── +# Main function for testing the script +# ───────────────────────────────────────────────────────────────────────────── + +def main() -> None: + """Main method for testing the optimizer + """ + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(message)s", + datefmt="%H:%M:%S", + ) + + pn_spec = PetrinautModelSpec() + opt_spec = OptimizationSpec( + n_trials = 10, + parameters = { + "infection_rate": 0.1 + }, + initial_states = { + "Susceptible": 100 + } + ) + # Build the Petri net from the client spec. + petrinet_model = PetrinautModel(pn_spec) + + # Instantiate Petrinaut optimization class + optimizer = PetrinautOptimizer( + opt_spec = opt_spec, + pn_model = petrinet_model + ) + + for step, params, init_states, metric_value in optimizer.run_stream( + optimizer.study, + optimizer.objective, + optimizer.n_trials + ): + log.info(json.dumps({"step":step,"params":params,"init_states":init_states,"metric":metric_value})) + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\ninterrupted", file=sys.stderr) + sys.exit(130) From 96832c8928559b7f994d78196b305223e2e68fea Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:38:04 +0300 Subject: [PATCH 06/19] created python API for petrinaut UI communication --- .../petrinaut-opt/src/optimization_api.py | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/src/optimization_api.py diff --git a/libs/@hashintel/petrinaut-opt/src/optimization_api.py b/libs/@hashintel/petrinaut-opt/src/optimization_api.py new file mode 100644 index 00000000000..c528f8ba6cf --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/src/optimization_api.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""HTTP API for Petrinaut optimization. + +Endpoints +--------- +POST /init Initialise the Petri-net execution model (petrinet_model.initialize) + and start an optimization run in the background. +GET /optimize Server-Sent Events stream of objective evaluations, + emitted as each trial completes, ending with a summary. +GET /status Current run model and state. + +Run with: uv run uvicorn optimization_api:app --reload +""" + +from __future__ import annotations + +import time +import uuid +import json + +from contextlib import asynccontextmanager +from typing import Union, Generator + +import optuna +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import StreamingResponse + +from src.petrinaut_client import PetrinautModelSpec, PetrinautModel +from src.petrinaut_optimizer import OptimizationSpec, PetrinautOptimizer + +optuna.logging.set_verbosity(optuna.logging.WARNING) + +# Concurrency is scoped per session: each `PetrinautOptimizer` holds its own lock +# so a single session can't be driven by two concurrent streams (see +# `stream_all`/`stream_best`). Independent sessions run independently — there is +# deliberately no global single-run guard. + +@asynccontextmanager +async def lifespan(app: FastAPI): + app.state.sessions: dict[str, PetrinautOptimizer] = {} + yield + app.state.sessions.clear() + +app = FastAPI(title="Petrinaut optimization API", lifespan=lifespan) + + +# ───────────────────────────────────────────────────────────────────────────── +# Dummy functions +# ───────────────────────────────────────────────────────────────────────────── + +def dummy_stream() -> Generator[dict[str,Union[float,int]]]: + """Dummy data stream to check that API endpoint works + """ + from datetime import datetime + n = 0 + while n < 10: + time.sleep(2) + event = {"inputs":[1.2,3.4],"output":datetime.now().strftime('%H:%M:%S'),"step":n} + yield f"{json.dumps(event)}\n\n" + n +=1 + +# ───────────────────────────────────────────────────────────────────────────── +# Endpoints +# ───────────────────────────────────────────────────────────────────────────── +@app.post("/init") +async def init(opt_spec: OptimizationSpec, pn_spec: PetrinautModelSpec) -> dict: + """Initialises the Petrinaut model and its optimization model from the Petrinaut UI + + Args: + opt_spec (OptimizationSpec): Specification for Petri Net input optimization with respect to output + pn_spec (PetrinautModelSpec): Specification for Petri Net execution + + Raises: + HTTPException: _description_ + + Returns: + dict: API status and Petrinaut model name + """ + # Build the model + optimizer. Each session is independent, so a failure here + # only fails this request — it never wedges the service for future /init calls. + try: + # Build the Petri net from the client spec. + petrinet_model = PetrinautModel(pn_spec) + # Instantiate Petrinaut optimization class + optimizer = PetrinautOptimizer( + opt_spec = opt_spec, + pn_model = petrinet_model, + ) + except Exception as exc: + raise HTTPException(500, f"failed to initialise optimization: {exc}") + + session_id = str(uuid.uuid4()) + app.state.sessions[session_id] = optimizer + return {"session_id": session_id, "status": "initialised", "pn_model": petrinet_model.name, "opt_study": optimizer.study_name} + + +@app.get("/optimize/{session_id}/stream/all",response_class=StreamingResponse) +async def optimize_stream_all(session_id: str, request: Request, n_trials:Union[int,None]=None) -> StreamingResponse: + """Streams optimization results per optimization step (trial) to json line + + Args: + session_id (str): Optimization API curent session id + request (Request): Optimization API generic request + + Raises: + HTTPException: Unknown session id + + Returns: + StreamingResponse: + """ + optimizer = app.state.sessions.get(session_id) + if optimizer is None: + raise HTTPException(404, "unknown session_id — call /init first") + + # Set default trials if no argument passed + n_trials = n_trials if n_trials else optimizer.n_trials + + # The optimiser's SSE generator acquires/releases the session lock itself, so + # ending the stream (completion, error, or client disconnect) never leaves the + # session wedged. + return StreamingResponse( + optimizer.stream_all(request, n_trials=n_trials), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + +@app.get("/optimize/{session_id}/stream/best",response_class=StreamingResponse) +async def optimize_stream_best(session_id: str, request: Request, n_trials:Union[int,None]=None) -> StreamingResponse: + """Streams current best optimization results per optimization step (trial) to json line + + Args: + session_id (str): Optimization API curent session id + request (Request): Optimization API generic request + + Raises: + HTTPException: Unknown session id + + Returns: + StreamingResponse: + """ + optimizer = app.state.sessions.get(session_id) + if optimizer is None: + raise HTTPException(404, "unknown session_id — call /init first") + + # Set default trials if no argument passed + n_trials = n_trials if n_trials else optimizer.n_trials + + # The optimiser's SSE generator acquires/releases the session lock itself, so + # ending the stream (completion, error, or client disconnect) never leaves the + # session wedged. + return StreamingResponse( + optimizer.stream_best(request, n_trials=n_trials), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + +@app.delete("/optimize/{session_id}") +async def teardown(session_id: str): + if app.state.sessions.pop(session_id, None) is None: + raise HTTPException(404, "unknown session_id") + return {"status": "deleted"} + + +@app.get("/status") +async def status() -> dict: + """Lists the currently active optimization sessions.""" + return { + "sessions": [ + { + "session_id": session_id, + "pn_model": optimizer.pn_model.name, + "opt_study": optimizer.study_name, + } + for session_id, optimizer in app.state.sessions.items() + ] + } + +@app.get("/") +async def root() -> dict: + return {"message": "Welcome to Petrinaut optimization API"} From 86edca0683eb131d0df2976b5ce89fc84717a169 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 15:42:40 +0300 Subject: [PATCH 07/19] added README for python-based petrinaut parameter and initial state optimisation --- libs/@hashintel/petrinaut-opt/README.md | 171 ++++++++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 libs/@hashintel/petrinaut-opt/README.md diff --git a/libs/@hashintel/petrinaut-opt/README.md b/libs/@hashintel/petrinaut-opt/README.md new file mode 100644 index 00000000000..6555aabeed9 --- /dev/null +++ b/libs/@hashintel/petrinaut-opt/README.md @@ -0,0 +1,171 @@ +# Petrinaut optimization + +Black-box optimization of a **Petri-net execution**. A request is submitted to a +Petrinaut UNIX socket server that executes the Petri net for a set of parameters +and initial states and returns a single metric value per run. +[Optuna](https://optuna.org/) searches the input space to maximise (or minimise) +that metric. Results stream out one evaluation at a time over Server-Sent Events, +so a UI can watch the optimization live. + +The search space is **continuous and discrete**, and evaluations are treated as +expensive, so a sample-efficient sampler (TPE by default) is used. + +## How it connects to Petrinaut + +This package does **not** run the Petri net itself — it drives the +[`petrinaut-cli`](../petrinaut-cli) socket server. Follow the instructions +on [there](../petrinaut-cli/README.md) first. + + +## Components + +| File | Role | +|------|------| +| [petrinaut_client.py](src/petrinaut_client.py) | `PetrinautModel` — connects to the Petrinaut CLI socket, builds each `run` request, and returns the metric. `PetrinautModelSpec` configures the execution. | +| [petrinaut_optimizer.py](src/petrinaut_optimizer.py) | `PetrinautOptimizer` — drives the Optuna study: proposes inputs, runs the model, and streams evaluations (`stream_all` / `stream_best`) as Server-Sent Events. `OptimizationSpec` configures a run; `BOUNDS` defines the search space. | +| [optimization_api.py](src/optimization_api.py) | FastAPI service exposing `/init`, the two streaming endpoints, `/status`, and `/`. | + +## Setup + +This is a [uv](https://docs.astral.sh/uv/) project (Python ≥ 3.10.20): + +```bash +uv sync +``` + +Imports are package-qualified (`from src...`), so run everything from the package +root using module syntax. + +## Run the optimizer directly + +`main()` runs a short study against the socket and logs each evaluation: + +```bash +# 1. petrinaut serve --model <...> --socket /tmp/petrinaut.sock (in another shell) +# 2. from libs/@hashintel/petrinaut-opt: +uv run python -m src.petrinaut_optimizer +``` + +## Run the API + +```bash +uv run uvicorn src.optimization_api:app --reload +``` + +### 1. Initialise a run — `POST /init` + +The body carries **two** objects: `opt_spec` (what to optimize) and `pn_spec` +(the Petri-net execution model). It returns a `session_id`. + +```bash +curl -s -X POST localhost:8000/init -H 'content-type: application/json' -d '{ + "opt_spec": { + "parameters": {"infection_rate": 0.5}, + "initial_states": {"Susceptible": 990}, + "n_trials": 30, + "sampler": "tpe", + "direction": "minimize" + }, + "pn_spec": { + "name": "SIR", + "metric": "Infected Fraction", + "steps": 100, + "dt": 0.1, + "seed": 1234 + } +}' +# -> {"session_id": "...", "status": "initialised", "pn_model": "SIR", "opt_study": "input_opt_..."} +``` + +Here `infection_rate`, and `Susceptible` are **fixed** at the given +values, and every other input in the search space (`recovery_rate`, `Recovered`, `Infected`) +is **optimized** — see [Configuring a run](#configuring-a-run). + +### 2. Stream evaluations — `GET /optimize/{session_id}/stream/all` + +Opens a Server-Sent Events stream: one frame per finished trial, then a final +`event: done`. Disconnecting the client stops the underlying study. Each frame +reports the inputs that were **searched** this trial (fixed inputs are constant +and omitted) plus the resulting metric. + +```bash +curl -sN localhost:8000/optimize//stream/all +# data: {"step": 0, "params": {"recovery_rate": 3.21}, "init_states": {"Recovered": 412.7}, "metric": 0.87, "state": "COMPLETE"} +# data: {"step": 1, ...} +# event: done +# data: {} +``` + +`state` is the Optuna trial state (`COMPLETE`, `PRUNED`, `FAIL`); `metric` is +`null` for a pruned trial. Pass `?n_trials=` to override the run's `n_trials`. + +### 3. Stream the running best — `GET /optimize/{session_id}/stream/best` + +Same shape, but each frame reports the **best-so-far** inputs and metric rather +than the latest trial. Frames are suppressed until at least one trial has +completed. + +### Other endpoints + +- `GET /status` — lists the currently active sessions (`session_id`, `pn_model`, + `opt_study`). +- `DELETE /optimize/{session_id}` — drops a session. +- `GET /` — welcome message. + +## Configuring a run + +**Search space** — the universe of optimizable inputs is defined once in `BOUNDS` +at the top of [petrinaut_optimizer.py](src/petrinaut_optimizer.py): + +```python +BOUNDS = { + "parameters": { + "infection_rate": Bounds(0.01, 10.0), + "recovery_rate": Bounds(0.01, 10.0), + }, + "initial_states": { + "Susceptible": Bounds(0.0, 1000), + "Infected": Bounds(0.0, 1000), + "Recovered": Bounds(0.0, 1000), + }, +} +``` + +**`OptimizationSpec`** ([petrinaut_optimizer.py](src/petrinaut_optimizer.py)) +partitions those inputs per run: + +- `parameters` / `initial_states` — any input you give a value here is **held + fixed** at that value; any input you omit (leave `null`) is **optimized** over + its `BOUNDS` range. Provided values must fall within `BOUNDS` (validated on + `/init`). +- `sampler` — `tpe` or `random`. +- `direction` — `maximize` or `minimize`. +- `n_trials` — number of evaluations (overridable per stream via the `n_trials` + query parameter). +- `study_name` — optional label for the Optuna study. + +**`PetrinautModelSpec`** ([petrinaut_client.py](src/petrinaut_client.py)) +configures the execution sent to the CLI: + +- `name` — Petri-net class name (default `SIR`). +- `metric` — metric name computed at the end of a run and used as the objective + (default `Infected Fraction`); must match a metric defined in the loaded model. +- `steps` — number of steps per run (sent as `maxSteps`). +- `dt` — timestep for the dynamics. +- `seed` — RNG seed (fixed → deterministic runs). +- `structure`, `outpath`, `store`, `eval_timeout` — accepted but currently unused + (the model structure is loaded server-side via the CLI's `--model`). + +## Notes + +- **Failures/timeouts**: any error returned by the CLI (or other exception during + a trial) marks that trial pruned; the study continues. +- **Streaming model**: evaluations run in a background thread and are pushed to + the SSE client through an `asyncio.Queue`. Each optimizer instance holds a lock, + so one session can't be driven by two concurrent streams — a second concurrent + stream on the same session receives `event: error` (`already running`). +- **Sessions are independent**: there is no global single-run guard; multiple + `/init` sessions can exist and run at once. +- **Leave one input free per group**: each group (`parameters`, `initial_states`) + should leave at least one input unfixed so there is something to optimize. + From e5d73b7d663237b7e3dd4c89e4f23ce8d4246f90 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:18:01 +0300 Subject: [PATCH 08/19] removed relative import to allow local python runs --- libs/@hashintel/petrinaut-opt/src/optimization_api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/optimization_api.py b/libs/@hashintel/petrinaut-opt/src/optimization_api.py index c528f8ba6cf..233dbdbf50c 100644 --- a/libs/@hashintel/petrinaut-opt/src/optimization_api.py +++ b/libs/@hashintel/petrinaut-opt/src/optimization_api.py @@ -25,8 +25,8 @@ from fastapi import FastAPI, HTTPException, Request from fastapi.responses import StreamingResponse -from src.petrinaut_client import PetrinautModelSpec, PetrinautModel -from src.petrinaut_optimizer import OptimizationSpec, PetrinautOptimizer +from petrinaut_client import PetrinautModelSpec, PetrinautModel +from petrinaut_optimizer import OptimizationSpec, PetrinautOptimizer optuna.logging.set_verbosity(optuna.logging.WARNING) From 37f8979b02bbd0b903a27129a064083d37e31cf4 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:30:34 +0300 Subject: [PATCH 09/19] brought back src imports --- libs/@hashintel/petrinaut-opt/src/optimization_api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/optimization_api.py b/libs/@hashintel/petrinaut-opt/src/optimization_api.py index 233dbdbf50c..c528f8ba6cf 100644 --- a/libs/@hashintel/petrinaut-opt/src/optimization_api.py +++ b/libs/@hashintel/petrinaut-opt/src/optimization_api.py @@ -25,8 +25,8 @@ from fastapi import FastAPI, HTTPException, Request from fastapi.responses import StreamingResponse -from petrinaut_client import PetrinautModelSpec, PetrinautModel -from petrinaut_optimizer import OptimizationSpec, PetrinautOptimizer +from src.petrinaut_client import PetrinautModelSpec, PetrinautModel +from src.petrinaut_optimizer import OptimizationSpec, PetrinautOptimizer optuna.logging.set_verbosity(optuna.logging.WARNING) From e143266a1128a93a1a8f8ffe948c5cab3ec034e0 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:32:02 +0300 Subject: [PATCH 10/19] added command from petrinautmodel spec and fixed __main__ --- .../@hashintel/petrinaut-opt/src/petrinaut_client.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py index e21bd43606e..6836234017e 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py @@ -33,6 +33,7 @@ DEFAULT_SEED = 1234 DEFAULT_STORE = ["metric"] DEFAULT_OUTPATH = "" +DEFAULT_COMMAND = "" DEFAULT_EVAL_TIMEOUT = None # Regex for extracting number from Petrinaut CLI @@ -50,8 +51,9 @@ class PetrinautModelSpec(BaseModel): steps: Optional[int] = Field(default=DEFAULT_STEPS, description="Number of steps in a single execution") dt: Optional[float] = Field(default=DEFAULT_TIMESTEP, description="Step size for dynamics discretisation in a single execution") seed: Optional[int] = Field(default=DEFAULT_SEED, description="Random number generator seed (fixed -> deterministic output)") - outpath: Optional[str] = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") store: Optional[Sequence[str]] = Field(default=DEFAULT_STORE, description="Quantities to store/print inside execution") + outpath: Optional[str] = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") + command: Optional[str] = Field(default=DEFAULT_METRIC, description="Petrinaut CLI command to invoke") eval_timeout: Optional[float] = Field(default=DEFAULT_EVAL_TIMEOUT, description="timeout threshold for CLI eval") @@ -77,6 +79,7 @@ def __init__( # they are left here for future releases self.outpath = pn_spec.outpath self.store = pn_spec.store + self.command = pn_spec.command self.eval_timeout = pn_spec.eval_timeout # Connect to Petrinaut-cli socket @@ -199,11 +202,10 @@ def _parse_output(stdout: str) -> float: pn_spec = PetrinautModelSpec() # Build the Petri net from the client spec. petrinet_model = PetrinautModel(pn_spec) - # Run petrinaut - result = petrinet_model.run( + # Run petrinaut once to get metric value + metric_value = petrinet_model.run( params = {"infection_rate": args.infection_rate, "recovery_rate": args.recovery_rate}, init_states = {"Susceptible": args.susceptible} ) # Read metric value from petrinaut execution - objective = result["metrics"]["Infected Fraction"] - log.info('objective',objective) + log.info('metric_value',metric_value) From b5e10c1eeeb480f3548c2b57d826576eeb4e1c80 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:32:31 +0300 Subject: [PATCH 11/19] passed study_name from opt_spec and fixed edge case of no learned parameters and intitial states --- .../petrinaut-opt/src/petrinaut_optimizer.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py index 0c43985a4f5..e5c19cf16aa 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py @@ -136,7 +136,7 @@ def __init__( self.fixed = opt_spec.fixed() self.params = opt_spec.parameters self.init_states = opt_spec.initial_states - self.study_name = f"input_opt_{datetime.now().strftime('%m/%d/%Y-%H:%M:%S')}" + self.study_name = f"{opt_spec.study_name}_{datetime.now().strftime('%m/%d/%Y-%H:%M:%S')}" self.sampler = SAMPLERS[opt_spec.sampler.lower()](**kwargs) self.direction = opt_spec.direction self.n_trials = opt_spec.n_trials @@ -247,8 +247,8 @@ def callback(study, trial): params_and_init_states.setdefault(outer, {})[inner] = v payload = { "step": trial.number, - "params": params_and_init_states["parameters"], - "init_states": params_and_init_states["initial_states"], + "params": params_and_init_states.get("parameters",dict()), + "init_states": params_and_init_states.get("initial_states",dict()), "metric": trial.value, "state": trial.state.name, } @@ -326,8 +326,8 @@ def callback(study, trial): best_params_and_init_states.setdefault(outer, {})[inner] = v payload = { "step": trial.number, - "params": best_params_and_init_states["parameters"], - "init_states": best_params_and_init_states["initial_states"], + "params": best_params_and_init_states.get("parameters",dict()), + "init_states": best_params_and_init_states.get("initial_states",dict()), "metric": study.best_value, "state": "COMPLETE", } From ec4a42cd746389daa835354bdf9f2c2752c0cbfc Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:42:54 +0300 Subject: [PATCH 12/19] changed command default value --- libs/@hashintel/petrinaut-opt/src/petrinaut_client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py index 6836234017e..bc9c51b3953 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py @@ -53,7 +53,7 @@ class PetrinautModelSpec(BaseModel): seed: Optional[int] = Field(default=DEFAULT_SEED, description="Random number generator seed (fixed -> deterministic output)") store: Optional[Sequence[str]] = Field(default=DEFAULT_STORE, description="Quantities to store/print inside execution") outpath: Optional[str] = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") - command: Optional[str] = Field(default=DEFAULT_METRIC, description="Petrinaut CLI command to invoke") + command: Optional[str] = Field(default=DEFAULT_COMMAND, description="Petrinaut CLI command to invoke") eval_timeout: Optional[float] = Field(default=DEFAULT_EVAL_TIMEOUT, description="timeout threshold for CLI eval") @@ -180,7 +180,7 @@ def _parse_output(stdout: str) -> float: format="%(asctime)s %(message)s", datefmt="%H:%M:%S", ) - + p = argparse.ArgumentParser(description="Toy Petri-net execution function.") p.add_argument("--infection_rate", type=float, required=True, help="infection rate in SIR Petri Net") From dfbf3db23b9d651953d76e1a5c62879b8f4129c5 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:44:00 +0300 Subject: [PATCH 13/19] fixed broken logging --- libs/@hashintel/petrinaut-opt/src/petrinaut_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py index bc9c51b3953..0ac378f8f4f 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py @@ -208,4 +208,4 @@ def _parse_output(stdout: str) -> float: init_states = {"Susceptible": args.susceptible} ) # Read metric value from petrinaut execution - log.info('metric_value',metric_value) + log.info(f'metric_value = {metric_value}') From 614a7dcda4729f28f8fceb59cab15a1ba706fe0c Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:02:32 +0300 Subject: [PATCH 14/19] fixed linting errors in README --- libs/@hashintel/petrinaut-opt/README.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/README.md b/libs/@hashintel/petrinaut-opt/README.md index 6555aabeed9..38a970b6b7e 100644 --- a/libs/@hashintel/petrinaut-opt/README.md +++ b/libs/@hashintel/petrinaut-opt/README.md @@ -13,10 +13,9 @@ expensive, so a sample-efficient sampler (TPE by default) is used. ## How it connects to Petrinaut This package does **not** run the Petri net itself — it drives the -[`petrinaut-cli`](../petrinaut-cli) socket server. Follow the instructions +[`petrinaut-cli`](../petrinaut-cli) socket server. Follow the instructions on [there](../petrinaut-cli/README.md) first. - ## Components | File | Role | From bde5db85091d0f1be3b7044b8124d4278816e2e8 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:06:58 +0300 Subject: [PATCH 15/19] fixed n_trials default behaviour --- libs/@hashintel/petrinaut-opt/src/optimization_api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/optimization_api.py b/libs/@hashintel/petrinaut-opt/src/optimization_api.py index c528f8ba6cf..252697777d2 100644 --- a/libs/@hashintel/petrinaut-opt/src/optimization_api.py +++ b/libs/@hashintel/petrinaut-opt/src/optimization_api.py @@ -113,7 +113,7 @@ async def optimize_stream_all(session_id: str, request: Request, n_trials:Union[ raise HTTPException(404, "unknown session_id — call /init first") # Set default trials if no argument passed - n_trials = n_trials if n_trials else optimizer.n_trials + n_trials = n_trials if (n_trials is not None) else optimizer.n_trials # The optimiser's SSE generator acquires/releases the session lock itself, so # ending the stream (completion, error, or client disconnect) never leaves the @@ -143,7 +143,7 @@ async def optimize_stream_best(session_id: str, request: Request, n_trials:Union raise HTTPException(404, "unknown session_id — call /init first") # Set default trials if no argument passed - n_trials = n_trials if n_trials else optimizer.n_trials + n_trials = n_trials if (n_trials is not None) else optimizer.n_trials # The optimiser's SSE generator acquires/releases the session lock itself, so # ending the stream (completion, error, or client disconnect) never leaves the From 528a47e4ff7cfee42fcc9dc21fa2e72494650250 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:31:25 +0300 Subject: [PATCH 16/19] added safety for edge case of no parameter optimisation --- libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py index e5c19cf16aa..bce7ce4882e 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py @@ -378,9 +378,9 @@ def callback(study, trial): outer, inner = k.split('.', 1) params_and_init_states.setdefault(outer, {})[inner] = v q.put(( - trial.number, - params_and_init_states["parameters"], - params_and_init_states["initial_states"], + trial.number, + params_and_init_states.get("parameters", dict()), + params_and_init_states.get("initial_states", dict()), trial.value )) From c692fc8488be8604f9e1193457d29f1ab7e88996 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:19:08 +0300 Subject: [PATCH 17/19] more lint fixes --- libs/@hashintel/petrinaut-opt/README.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/README.md b/libs/@hashintel/petrinaut-opt/README.md index 38a970b6b7e..e59243e7d85 100644 --- a/libs/@hashintel/petrinaut-opt/README.md +++ b/libs/@hashintel/petrinaut-opt/README.md @@ -18,11 +18,11 @@ on [there](../petrinaut-cli/README.md) first. ## Components -| File | Role | -|------|------| -| [petrinaut_client.py](src/petrinaut_client.py) | `PetrinautModel` — connects to the Petrinaut CLI socket, builds each `run` request, and returns the metric. `PetrinautModelSpec` configures the execution. | +| File | Role | +| ---------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| [petrinaut_client.py](src/petrinaut_client.py) | `PetrinautModel` — connects to the Petrinaut CLI socket, builds each `run` request, and returns the metric. `PetrinautModelSpec` configures the execution. | | [petrinaut_optimizer.py](src/petrinaut_optimizer.py) | `PetrinautOptimizer` — drives the Optuna study: proposes inputs, runs the model, and streams evaluations (`stream_all` / `stream_best`) as Server-Sent Events. `OptimizationSpec` configures a run; `BOUNDS` defines the search space. | -| [optimization_api.py](src/optimization_api.py) | FastAPI service exposing `/init`, the two streaming endpoints, `/status`, and `/`. | +| [optimization_api.py](src/optimization_api.py) | FastAPI service exposing `/init`, the two streaming endpoints, `/status`, and `/`. | ## Setup @@ -167,4 +167,4 @@ configures the execution sent to the CLI: `/init` sessions can exist and run at once. - **Leave one input free per group**: each group (`parameters`, `initial_states`) should leave at least one input unfixed so there is something to optimize. - + From 1a069079ec6e651990ca8bc84dc9977e391330b6 Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:40:07 +0300 Subject: [PATCH 18/19] created default study name and made all spec params compulsory with defaults --- .../petrinaut-opt/src/petrinaut_optimizer.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py index bce7ce4882e..866e05a93eb 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_optimizer.py @@ -81,6 +81,8 @@ class InitialStates(BaseModel): {name.upper(): name for name in SAMPLERS}, type=str, ) +# optuna study name prefix +DEFAULT_STUDY_NAME = "opt_study" # input space sampling algorithm - default: Tree-structured Parzen Estimator DEFAULT_SAMPLER = "tpe" # maximise the CLI's output @@ -92,10 +94,10 @@ class InitialStates(BaseModel): class OptimizationSpec(BaseModel): parameters: Parameters = Parameters() initial_states: InitialStates = InitialStates() - study_name: Optional[str] = Field(default=None, description="Name of optimization study") - sampler: Optional[SamplerName] = Field(default=DEFAULT_SAMPLER, description="input sampling algorithm") - direction: Optional[Literal["maximize", "minimize"]] = Field(default=DEFAULT_DIRECTION, description="optimization direction") - n_trials: Optional[int] = Field(default=DEFAULT_N_TRIALS, description="number of evals to run") + study_name: str = Field(default=DEFAULT_STUDY_NAME, description="Name of optimization study") + sampler: SamplerName = Field(default=DEFAULT_SAMPLER, description="input sampling algorithm") + direction: Literal["maximize", "minimize"] = Field(default=DEFAULT_DIRECTION, description="optimization direction") + n_trials: int = Field(default=DEFAULT_N_TRIALS, description="number of evals to run") def fixed(self) -> dict[str, dict[str, float]]: out: dict[str, dict[str, float]] = {} From b3ace774b76f303c0df0cf88f9e0e6223da912ee Mon Sep 17 00:00:00 2001 From: Yannis Zachos <43412884+YannisZa@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:40:30 +0300 Subject: [PATCH 19/19] made all petrinautmodel spec params compulsory and refactored main function --- .../petrinaut-opt/src/petrinaut_client.py | 36 ++++++------------- 1 file changed, 10 insertions(+), 26 deletions(-) diff --git a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py index 0ac378f8f4f..177a8924e18 100644 --- a/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py +++ b/libs/@hashintel/petrinaut-opt/src/petrinaut_client.py @@ -48,13 +48,13 @@ class PetrinautModelSpec(BaseModel): name: str = Field(default=DEFAULT_MODEL, description="Petri Net class name") structure: str = Field(default=DEFAULT_STRUCTURE, description="Filepath of Petri Net model structure (places, transitions, arcs etc.)") metric: str = Field(default=DEFAULT_METRIC, description="Metric name that will be computed at the end of execution") - steps: Optional[int] = Field(default=DEFAULT_STEPS, description="Number of steps in a single execution") - dt: Optional[float] = Field(default=DEFAULT_TIMESTEP, description="Step size for dynamics discretisation in a single execution") - seed: Optional[int] = Field(default=DEFAULT_SEED, description="Random number generator seed (fixed -> deterministic output)") - store: Optional[Sequence[str]] = Field(default=DEFAULT_STORE, description="Quantities to store/print inside execution") - outpath: Optional[str] = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") - command: Optional[str] = Field(default=DEFAULT_COMMAND, description="Petrinaut CLI command to invoke") - eval_timeout: Optional[float] = Field(default=DEFAULT_EVAL_TIMEOUT, description="timeout threshold for CLI eval") + steps: int = Field(default=DEFAULT_STEPS, description="Number of steps in a single execution") + dt: float = Field(default=DEFAULT_TIMESTEP, description="Step size for dynamics discretisation in a single execution") + seed: int = Field(default=DEFAULT_SEED, description="Random number generator seed (fixed -> deterministic output)") + store: Sequence[str] = Field(default=DEFAULT_STORE, description="Quantities to store/print inside execution") + outpath: str = Field(default=DEFAULT_OUTPATH, description="Filepath to execution trace") + command: str = Field(default=DEFAULT_COMMAND, description="Petrinaut CLI command to invoke") + eval_timeout: float = Field(default=DEFAULT_EVAL_TIMEOUT, description="timeout threshold for CLI eval") # ───────────────────────────────────────────────────────────────────────────── @@ -181,31 +181,15 @@ def _parse_output(stdout: str) -> float: datefmt="%H:%M:%S", ) - p = argparse.ArgumentParser(description="Toy Petri-net execution function.") - p.add_argument("--infection_rate", type=float, required=True, - help="infection rate in SIR Petri Net") - p.add_argument("--recovery_rate", type=float, required=True, - help="recovery rate in SIR Petri Net") - p.add_argument("--susceptible", type=int, required=True, - help="Number of susceptible individuals in SIR Petri Net") - p.add_argument("--infected", type=int, required=True, - help="Number of infected individuals in SIR Petri Net") - p.add_argument("--recovered", type=int, required=True, - help="Number of recovered individuals in SIR Petri Net") - p.add_argument("--steps", type=int, default=1000, - help="simulation steps (fixed; not optimized)") - p.add_argument("--seed", type=int, default=0, - help="RNG seed (fixed -> deterministic output)") - args = p.parse_args() - # Create the petrinaut execution specification pn_spec = PetrinautModelSpec() # Build the Petri net from the client spec. petrinet_model = PetrinautModel(pn_spec) # Run petrinaut once to get metric value metric_value = petrinet_model.run( - params = {"infection_rate": args.infection_rate, "recovery_rate": args.recovery_rate}, - init_states = {"Susceptible": args.susceptible} + params = {"infection_rate": 1.2, + "recovery_rate": 0.6}, + init_states = {"Susceptible": 1000.0} ) # Read metric value from petrinaut execution log.info(f'metric_value = {metric_value}')