script

py3r.behaviour.script lets you run and sweep parameterised Python pipeline scripts without modifying them.

Annotate any scalar script variable with Param and any intermediate result with Output. Both are transparent during normal execution — the script runs unchanged on its own. When invoked via run or sensitivity, parameter values are injected into each subprocess and outputs are captured and returned in a ScriptResults container.

Each iteration runs in a fresh subprocess, so process state never leaks between runs. stop_after_outputs=True combined with outputs= terminates the subprocess as soon as the last requested output is captured, allowing you to short-circuit long pipelines when only early-stage results are needed.

Annotation

py3r.behaviour.script.param.Param

Param(
    default: bool | int | float | str | None = _MISSING,
    *,
    name: str,
) -> bool | int | float | str

Mark a script variable as a runner parameter.

In normal script execution, returns default. If no default is given, raises RuntimeError to signal that the script must be run via :func:run or :func:sensitivity with a value supplied for this parameter.

When run via :func:run or :func:sensitivity, returns the injected value.

Parameters:

Name	Type	Description	Default
default	bool | int | float | str | None	Value used during normal execution. Omit to mark the parameter as required (no default — must always be supplied by the runner).	_MISSING
name	str	Parameter name, matched against keys passed to run or sensitivity.	required

Examples

>>> window = Param(5, name="window")
>>> window
5
>>> type(window)
<class 'int'>

py3r.behaviour.script.param.Output

Output(value: object, *, name: str) -> object

Mark a script value as a runner output to capture.

In normal script execution, returns value unchanged. When run via :func:run or :func:sensitivity, captures the value under name in the results.

Parameters:

Name	Type	Description	Default
value	object	The value to capture. Any type is accepted.	required
name	str	Output name, used to key results in ScriptResults.	required

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({"a": [1, 2]})
>>> result = Output(df, name="summary")
>>> result is df
True

Running and sweeping

py3r.behaviour.script.runner.run

run(
    script_path: str | Path,
    params: dict[str, Any] | None = None,
    *,
    outputs: list[str] | None = None,
    stop_after_outputs: bool = False,
) -> ScriptResults

Run a parameterised script exactly once with the given parameter values.

Unspecified parameters use their script default. Parameters with no default must be provided.

Parameters:

Name	Type	Description	Default
script_path	str | Path	Path to a Python script containing Param and/or Output calls.	required
params	dict[str, Any] | None	Parameter values to inject, keyed by name. Unspecified params use their script default.	None
outputs	list[str] | None	Names of Output values to capture. Defaults to all outputs.	None
stop_after_outputs	bool	If True, the subprocess is terminated immediately after the last requested output is captured.	False

Returns:

Type	Description
ScriptResults	Single-entry results container. Access outputs via
ScriptResults	sr[params]["output_name"].

Examples

from py3r.behaviour.script import run

sr = run("pipeline.py", {"data_path": "/data/session1", "threshold": 0.8})
summary = sr[{"data_path": "/data/session1", "threshold": 0.8}]["summary"]

Capture only an early output and stop the subprocess before downstream stages run:

sr = run("pipeline.py", {"threshold": 0.8}, outputs=["tracking"], stop_after_outputs=True)

py3r.behaviour.script.runner.sensitivity

sensitivity(
    script_path: str | Path,
    params: dict[str, list],
    *,
    nominal: dict[str, Any] | None = None,
    outputs: list[str] | None = None,
    stop_after_outputs: bool = False,
    mode: Literal["independent", "grid"] = "independent",
) -> ScriptResults

Run a parameterised script repeatedly, sweeping flagged :func:Param values.

Each iteration runs in a subprocess. :func:Output values are collected and returned in a :class:ScriptResults container. Failed iterations are recorded but do not stop the run.

For each swept parameter, its script default is automatically included in the sweep (deduplicated silently) and used as the nominal value for independent-mode sweeps. Use nominal to override this.

Parameters:

Name	Type	Description	Default
script_path	str | Path	Path to a Python script containing Param and Output calls.	required
params	dict[str, list]	Mapping of parameter name to list of values to sweep.	required
nominal	dict[str, Any] | None	Nominal (baseline) values for independent-mode sweeps. Overrides script defaults. Required for any swept parameter that has no script default.	None
outputs	list[str] | None	Names of Output values to capture. Defaults to all outputs.	None
stop_after_outputs	bool	If True, each subprocess is terminated after the last requested output is captured. Useful when outputs appear early in a long pipeline.	False
mode	Literal['independent', 'grid']	"independent" varies one parameter at a time, holding the others at their nominal value. "grid" tests every combination.	'independent'

Returns:

Type	Description
ScriptResults	Results container keyed by parameter combination. Access individual
ScriptResults	outputs via sr[{"param": value}]["output_name"], or flatten
ScriptResults	scalar outputs with sr.to_dataframe().

Examples

Independent sweep (default mode) — one parameter varies at a time:

from py3r.behaviour.script import sensitivity

sr = sensitivity("pipeline.py", {"smooth_window": [3, 5, 7], "threshold": [0.5, 0.8]})
df = sr.to_dataframe()

Grid sweep — every combination of parameter values:

sr = sensitivity(
    "pipeline.py",
    {"smooth_window": [3, 5, 7], "threshold": [0.5, 0.8]},
    mode="grid",
)

Required parameter not being swept must be supplied via nominal:

sr = sensitivity(
    "pipeline.py",
    {"smooth_window": [3, 5, 7]},
    nominal={"threshold": 0.8},
)

Early termination — stop each subprocess after the first output is captured, skipping downstream pipeline stages:

sr = sensitivity(
    "pipeline.py",
    {"smooth_window": [3, 5, 7]},
    outputs=["tracking"],
    stop_after_outputs=True,
)

py3r.behaviour.script.runner.inspect

inspect(script_path: str | Path) -> None

Print a summary of the :func:Param and :func:Output calls in a script.

Useful for quickly checking what parameters a script exposes, which have defaults, and what outputs it produces — without actually running it.

Parameters:

Name	Type	Description	Default
script_path	str | Path	Path to a Python script containing Param and/or Output calls.	required

Examples

from py3r.behaviour.script import inspect
inspect("pipeline.py")
# Script: pipeline.py
#
# Parameters:
#   data_path            default='/my/data'  type=str
#   smooth_window        default=5           type=int
#   threshold            required
#
# Outputs:
#   tracking
#   summary

Results

py3r.behaviour.script.results.ScriptResults

ScriptResults(
    param_names: list[str], output_names: list[str]
)

Container for results returned by :func:run and :func:sensitivity.

Keyed by parameter combination; holds captured :func:Output values per iteration. Access individual outputs by passing a parameter dict:

sr[{"window": 5}]["summary"]

Flatten scalar outputs across all iterations to a DataFrame:

df = sr.to_dataframe()

Attributes

errors : list[tuple[dict, str]] Failed iterations as (param_dict, error_message) pairs.

errors property

errors: list[tuple[dict, str]]

Failed iterations as a list of (param_dict, error_message) pairs.

to_dataframe

to_dataframe() -> pd.DataFrame

Flatten scalar outputs into a DataFrame (one row per iteration).

Each row contains the parameter values for that iteration plus one column per captured scalar output. Raises TypeError if any output is not a scalar — access those directly via indexing instead.

Returns

pd.DataFrame One row per completed iteration; columns are parameter names followed by output names.