pyantz.jobs.branching package

pyantz.jobs.branching package#

Submodules#

pyantz.jobs.branching.case_matrix module#

Jobs for running a pipeline over a grid of parameter combinations.

create_case_matrix() produces a Parquet case-matrix file from a set of VariableDefinition ranges, then pipeline_expansion() reads that file and submits one copy of a pipeline_template per row, with each row’s columns injected as variables.

pipeline_expansion_with_output_dir() is a convenience wrapper that also creates per-run output directories and injects output_dir and pipeline_id variables.

class CaseMatrixCreator(*, variables: Mapping[str, VariableDefinition], save_file: str)[source]#

Bases: BaseModel

Parameters to create a case matrix from a set of variables.

Produces the cartesian products (aka all possible permutations) of the variables to create an expansive dense case matrix.

Parameters:

variables (Annotated[Mapping[str, VariableDefinition], MinLen(min_length=1)])
save_file (str)

_abc_impl = <_abc._abc_data object>#

save_file: str#: Parquet file to save the case matrix to.

variables: Mapping[str, VariableDefinition]#: The variables, by name; each becomes a column of the case matrix.

class CaseMatrixExpansionParams(*, case_matrix_parquet: Path, cols_to_exclude: tuple[str, ...] | None = None, pipeline_template: JobPipeline)[source]#

Bases: BaseModel

Parameters to create many pipelines from a case matrix.

Parameters:

case_matrix_parquet (Path)
cols_to_exclude (tuple[str, ...] | None)
pipeline_template (JobPipeline)

_abc_impl = <_abc._abc_data object>#

case_matrix_parquet: Path#

one row per pipeline, one column per variable.

Type:: Parquet file of the case matrix

cols_to_exclude: tuple[str, ...] | None#: If provided, these columns will not be read by the job; useful for large columns which are only there for human readability.

pipeline_template: JobPipeline#: The jobs to copy for each row of the case matrix.

class CaseMatrixRunSetup(*, case_matrix_parquet: Path, cols_to_exclude: tuple[str, ...] | None = None, pipeline_template: JobPipeline, output_dir: str)[source]#

Bases: CaseMatrixExpansionParams

Create a case matrix and provide each sub-pipeline with an output directory.

Parameters:

case_matrix_parquet (Path)
cols_to_exclude (tuple[str, ...] | None)
pipeline_template (JobPipeline)
output_dir (str)

_abc_impl = <_abc._abc_data object>#

output_dir: str#: Root path for all the output directories.

class ContinuousRange[S: int | float](*, min_value: S, max_value: S, step: S)[source]#

Bases: BaseModel, Generic[S]

Range of values like numpy arange: min_value up to, excluding, the max.

Parameters:

min_value (S)
max_value (S)
step (S)

_abc_impl = <_abc._abc_data object>#

_check_step_is_positive()[source]#

Ensure that the range makes progress, instead of looping forever.

Parameters:: None.
Returns:: the validated range.
Return type:: Self
Raises:: ValueError – if step isn’t greater than 0.

get_values()[source]#

Generate values from this range.

Each value is computed as min_value + i * step (rather than by repeatedly adding the step) so floating point errors don’t build up.

Parameters:: None.
Yields:: S – each value from min_value up to, excluding, max_value; float values are rounded to FLOAT_DECIMALS places.
Returns:: the values of the range.
Return type:: Generator[S]
Raises:: None. –

Example

from pyantz.jobs.branching.case_matrix import ContinuousRange

ints = ContinuousRange(min_value=0, max_value=10, step=3)
floats = ContinuousRange(min_value=0.0, max_value=0.3, step=0.1)
print(list(ints.get_values()), list(floats.get_values()))

[0, 3, 6, 9] [0.0, 0.1, 0.2]

max_value: S#: End of the range, which is not included.

min_value: S#: First value of the range.

step: S#: Difference between consecutive values; it must be greater than 0.

class DiscreteRange[S](*, possible_values: list[S])[source]#

Bases: BaseModel, Generic[S]

Range of values defined by the user.

Parameters:: possible_values (list[S])

_abc_impl = <_abc._abc_data object>#

classmethod _remove_repeats(values: list[S])[source]#

Drop repeated values keeping the first of each (works for unhashable).

Parameters:: values (list[S]) – the values given by the user.
Returns:: the values without repeats, in their original order.
Return type:: list[S]
Raises:: None. –

get_values()[source]#

Generate values from the user defined range.

Parameters:: None.
Yields:: S – each of possible_values, in order.
Returns:: the values of the range.
Return type:: Generator[S]
Raises:: None. –

Example

from pyantz.jobs.branching.case_matrix import DiscreteRange

values = DiscreteRange(possible_values=["b", "a", "b"]).get_values()
print(list(values))

['b', 'a']

possible_values: list[S]#: The values, used in the order given, with repeated values removed.

FLOAT_DECIMALS = 12#: Values of float ranges are rounded to this many places to hide float noise.

type PipelineSubmitter = Callable[[SubmissionFnType], None]#: Function which accepts a submit function and submits a pipeline.

class VariableDefinition(*, range: ContinuousRange[Any] | DiscreteRange[Any])[source]#

Bases: BaseModel

Defines how to create a range of values for the case matrix.

Parameters:: range (ContinuousRange[Any] | DiscreteRange[Any])

_abc_impl = <_abc._abc_data object>#

range: ContinuousRange[Any] | DiscreteRange[Any]#

a ContinuousRange (min_value, max_value, step) or a DiscreteRange (possible_values).

Type:: The values of the variable

_nested_jobs(value: Any)[source]#

Find the job configs (objects or raw dicts) inside value, at any depth.

A raw dict counts as a job if it has a function key.

Parameters:: value (Any) – the value to search, such as a job’s parameters.
Returns:: the jobs found, each followed by the jobs nested in it.
Return type:: list[Any]
Raises:: None. –

Example

from pyantz.jobs.branching.case_matrix import _nested_jobs

branch = {"function": "x.y.z", "job_id": "inner", "parameters": {}}
found = _nested_jobs({"if_true": branch, "others": [1, "two"]})
print([job["job_id"] for job in found])

['inner']

_pipeline_factory_factory(pipeline_template: list[JobConfig])[source]#

Return a factory that clones a pipeline template for a set of variables.

Each call to the returned factory produces a fresh copy of pipeline_template with new UUIDs, chained dependencies, and the supplied variable mapping injected into every job.

Parameters:: pipeline_template (list[JobConfig]) – The list of JobConfig objects to clone for each case.
Return type:: Callable[[Mapping[str, Any]], PipelineSubmitter]
Returns:: A callable that accepts a variable mapping and returns a PipelineSubmitter closure. Calling that closure with a SubmissionFnType enqueues the cloned pipeline.
Raises:: None. –

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.jobs.branching.case_matrix import _pipeline_factory_factory

template = [
    JobConfig(function="pyantz.jobs.testing.nop.do_nothing", parameters={}),
    JobConfig(function="pyantz.jobs.testing.nop.do_nothing", parameters={}),
]
create_pipeline = _pipeline_factory_factory(template)
submitted = []
create_pipeline({"x": 1})(submitted.append)
first, second = submitted
print(second.depends_on == {first.job_id}, first.variables)

True {'x': 1, 'pipeline_id': 0}

_regenerate_nested_ids(parameters: Any)[source]#

Give every job nested in parameters a new id, keeping their dependencies.

Jobs in a template are cloned for every row, so nested jobs with fixed ids would collide in the queue. Dependencies between the nested jobs follow their new ids. Ids written as templates (containing %{) are left alone as variables make them unique per clone.

Parameters:: parameters (Any) – the parameters of a job of the template.
Returns:: the parameters with new ids (parameters itself if it has no nested jobs with fixed ids).
Return type:: Any
Raises:: None. –

Example

from pyantz.jobs.branching.case_matrix import _regenerate_nested_ids

params = {"if_true": {"function": "x.y.z", "job_id": "a", "parameters": {}}}
print(_regenerate_nested_ids(params)["if_true"]["job_id"] != "a")

True

_rewrite_ids(value: Any, id_map: Mapping[str, str])[source]#

Replace job ids and the dependencies on them, in the jobs nested in value.

Parameters:

value (Any) – the value to rewrite, such as a job’s parameters.
id_map (Mapping[str, str]) – old job ids mapped to their new ids; ids not in it are kept.

Returns:

a copy of value with the ids replaced.

Return type:

Any

Raises:

None. –

Example

from pyantz.jobs.branching.case_matrix import _rewrite_ids

jobs = [
    {"function": "x.y.z", "job_id": "a", "parameters": {}},
    {"function": "x.y.z", "depends_on": ["a"], "parameters": {}},
]
for job in _rewrite_ids(jobs, {"a": "new"}):
    print(job.get("job_id"), job.get("depends_on"))

new None
None ['new']

create_case_matrix(params: CaseMatrixCreator)[source]#

Create a case matrix as a cartesian product of the variables.

The case matrix has a column for each variable and a row for each combination of their values. The first variable changes slowest.

Parameters:

params (CaseMatrixCreator) – the job’s parameters, validated as CaseMatrixCreator.

Job parameters:

variables (Mapping[str, VariableDefinition]) – the variables, by name, each with a range: either min_value, max_value and step or possible_values.
save_file (str) – parquet file to save the case matrix to.

Returns:

True once the case matrix is saved.

Return type:

bool

Raises:

polars.exceptions.PolarsError – if the parquet can’t be written.
OSError – if the parquet’s directory doesn’t exist.

Example

import polars as pl

from pyantz.jobs.branching.case_matrix import create_case_matrix

create_case_matrix(
    {
        "variables": {
            "size": {"range": {"possible_values": ["small", "large"]}},
            "speed": {
                "range": {"min_value": 1, "max_value": 4, "step": 1},
            },
        },
        "save_file": "cases.parquet",
    },
    lambda job: None,
)
for row in pl.read_parquet("cases.parquet").rows():
    print(row)

('small', 1)
('small', 2)
('small', 3)
('large', 1)
('large', 2)
('large', 3)

pipeline_expansion(params: CaseMatrixExpansionParams, submit_fn: SubmissionFnType)[source]#

Submit a set of pipelines based on the case matrix.

For each row of the case matrix, a copy of pipeline_template is submitted with the row’s columns as variables, plus pipeline_id (0, 1, …). The copies’ jobs get new ids. If the template’s jobs don’t depend on each other, each copy runs its jobs in the order given.

Parameters:

params (CaseMatrixExpansionParams) – the job’s parameters, validated as CaseMatrixExpansionParams.
submit_fn (SubmissionFnType) – function to submit the jobs of the pipelines with.

Job parameters:

case_matrix_parquet (Path) – parquet file of the case matrix.
cols_to_exclude (tuple[str, …] | None) – columns not to read. Defaults to None.
pipeline_template (JobPipeline) – the jobs to copy for each row.

Returns:

True once every pipeline is submitted.

Return type:

bool

Raises:

polars.exceptions.PolarsError – if the case matrix can’t be read, e.g., it doesn’t exist or lacks an excluded column.

Example

import polars as pl

from pyantz.jobs.branching.case_matrix import pipeline_expansion

pl.DataFrame({"speed": [1, 2]}).write_parquet("cases.parquet")
submitted = []
pipeline_expansion(
    {
        "case_matrix_parquet": "cases.parquet",
        "pipeline_template": [
            {
                "function": "pyantz.jobs.testing.nop.do_nothing",
                "parameters": {"out": "speed_%{speed}.txt"},
            },
        ],
    },
    submitted.append,
)
for job in submitted:
    print(job.variables)

{'speed': 1, 'pipeline_id': 0}
{'speed': 2, 'pipeline_id': 1}

pipeline_expansion_with_output_dir(params: CaseMatrixRunSetup, submit_fn: SubmissionFnType)[source]#

Add variables output_dir and pipeline_id for each child pipeline.

Like pipeline_expansion(), but each pipeline also gets its own output directory <output_dir>/run_<pipeline_id>, created before its jobs are submitted and given to them in the output_dir variable. When this job runs inside another expanded pipeline, the ids of the enclosing pipelines are kept as parent_pipeline_id, parent_parent_pipeline_id and so on.

Parameters:

params (CaseMatrixRunSetup) – the job’s parameters, validated as CaseMatrixRunSetup.
submit_fn (SubmissionFnType) – function to submit the jobs of the pipelines with.

Job parameters:

case_matrix_parquet (Path) – parquet file of the case matrix.
cols_to_exclude (tuple[str, …] | None) – columns not to read. Defaults to None.
pipeline_template (JobPipeline) – the jobs to copy for each row.
output_dir (str) – root directory of the pipelines’ output directories.

Returns:

True once every pipeline is submitted.

Return type:

bool

Raises:

OSError – if an output directory can’t be created.
polars.exceptions.PolarsError – if the case matrix can’t be read, e.g., it doesn’t exist.

Example

import os

import polars as pl

from pyantz.jobs.branching.case_matrix import (
    pipeline_expansion_with_output_dir,
)

pl.DataFrame({"speed": [1, 2]}).write_parquet("cases.parquet")
submitted = []
pipeline_expansion_with_output_dir(
    {
        "case_matrix_parquet": "cases.parquet",
        "output_dir": "results",
        "pipeline_template": [
            {
                "function": "pyantz.jobs.testing.nop.do_nothing",
                "parameters": {},
            },
        ],
    },
    submitted.append,
)
print(sorted(os.listdir("results")))
print([job.variables["output_dir"] for job in submitted])

['run_0', 'run_1']
['results/run_%{pipeline_id}', 'results/run_%{pipeline_id}']

pyantz.jobs.branching.if_else module#

Conditional job execution based on a comparison between two values.

if_else() submits one of two jobs depending on whether left_side <comparator> right_side is true.

conditionally_execute() is a simpler variant that only executes a job when the condition is true; it does nothing on the false branch.

Both jobs support template variables in left_side and right_side, making them useful for dynamic branching based on runtime state.

type Coercion = Literal['auto', 'str', 'float']#: How the sides are read before comparing: auto (numbers if both can be), str (text) or float (numbers).

class ConditionalExecution(*, left_side: str | float, right_side: str | float, comparator: ValidOperators, rel_tol: float = 1e-06, abs_tol: float = 0.0, coerce: Coercion = 'auto', if_true: JobConfig)[source]#

Bases: BaseModel

Execute a job iff a condition is true.

Parameters:

left_side (str | float)
right_side (str | float)
comparator (ValidOperators)
rel_tol (float)
abs_tol (float)
coerce (Coercion)
if_true (JobConfig)

_abc_impl = <_abc._abc_data object>#

abs_tol: float#: Absolute tolerance of the ~= (approximately equal) comparator, see math.isclose().

coerce: Coercion#: How the sides are read before comparing, see IfElseParams.coerce.

comparator: ValidOperators#: How to compare the sides, see ValidOperators.

if_true: JobConfig#: Job submitted when the comparison is true.

left_side: str | float#: Left side of the comparison.

rel_tol: float#: Relative tolerance of the ~= (approximately equal) comparator, see math.isclose().

right_side: str | float#: Right side of the comparison.

DEFAULT_REL_TOL = 1e-06#

values within one part per million are equal.

Type:: Default relative tolerance of ~=

class IfElseParams(*, left_side: str | float, right_side: str | float, comparator: ValidOperators, rel_tol: float = 1e-06, abs_tol: float = 0.0, coerce: Coercion = 'auto', if_true: JobConfig, if_false: JobConfig)[source]#

Bases: BaseModel

Parameters for the if/else job.

Parameters:

left_side (str | float)
right_side (str | float)
comparator (ValidOperators)
rel_tol (float)
abs_tol (float)
coerce (Coercion)
if_true (JobConfig)
if_false (JobConfig)

_abc_impl = <_abc._abc_data object>#

abs_tol: float#: Absolute tolerance of the ~= (approximately equal) comparator, see math.isclose().

coerce: Coercion#: How the sides are read before comparing with =, <, >, …: auto as numbers if both can be, otherwise as they are; str as text; float as numbers (the job fails if one isn’t a number). in, not in and =~ always compare text.

comparator: ValidOperators#: How to compare the sides, see ValidOperators.

if_false: JobConfig#

submitted when the comparison is false.

Type:: The “else” part of the job

if_true: JobConfig#

submitted when the comparison is true.

Type:: The “if” part of the job

left_side: str | float#: Left side of the comparison.

rel_tol: float#: Relative tolerance of the ~= (approximately equal) comparator, see math.isclose().

right_side: str | float#: Right side of the comparison.

type ValidOperators = Literal['=', '==', '<', '<=', '>', '>=', '!=', '~=', 'in', 'not in', '=~']#

Comparators of left_side <comparator> right_side.

=/==, !=, <, <=, > and >= compare as usual; ~= is approximately equal (math.isclose()); in and not in check if the left side is a substring of the right side; =~ checks if the regular expression on the right side matches anywhere in the left side.

_compare(params: IfElseParams | ConditionalExecution)[source]#

Evaluate left <comparator> right.

~= means approximately equal: both sides are read as numbers and compared with math.isclose() using the rel_tol and abs_tol parameters.

Parameters:

params (IfElseParams | ConditionalExecution) – the parameters holding the sides, the comparator and how to read the sides.

Returns:

the result of the comparison; False for ~= when a side isn’t a number.

Return type:

bool

Raises:

ValueError – if coerce is float and a side isn’t a number.
re.error – if comparator is =~ and the right side isn’t a valid regular expression.

Example

from pyantz.jobs.branching.if_else import ConditionalExecution, _compare

job = {"function": "pyantz.jobs.testing.nop.do_nothing", "parameters": {}}
for coerce in ("auto", "str"):
    params = ConditionalExecution(
        left_side="10",
        comparator=">",
        right_side="9",
        coerce=coerce,
        if_true=job,
    )
    print(coerce, _compare(params))

auto True
str False

_compare_text(left: str, right: str, comparator: str)[source]#

Evaluate the comparators that read both sides as text.

Parameters:

left (str) – the left side.
right (str) – the right side; for =~, a regular expression.
comparator (str) – in, not in or =~.

Returns:

for in/not in, whether left is (not) a substring of right; for =~, whether right matches anywhere in left.

Return type:

bool

Raises:

re.error – if comparator is =~ and right isn’t a valid regular expression.

Example

from pyantz.jobs.branching.if_else import _compare_text

print(_compare_text("ant", "pyantz", "in"))
print(_compare_text("run_42", r"\d+$", "=~"))

True
True

_to_number(value: str | float)[source]#

Read a value as a number, or None if it isn’t one.

Parameters:: value (str | float) – the value to read.
Returns:: the value as a finite number, or None if it isn’t one ("nan" and "inf" count as text).
Return type:: float | None
Raises:: None. –

Example

from pyantz.jobs.branching.if_else import _to_number

print(_to_number("1.5"), _to_number(2), _to_number("nan"), _to_number("x"))

1.5 2.0 None None

conditionally_execute(params: ConditionalExecution, submit_fn: SubmissionFnType)[source]#

Execute a job iff a condition is true.

Like if_else() without an “else”: nothing is submitted when the comparison is false, and the job still succeeds.

Parameters:

params (ConditionalExecution) – the job’s parameters, validated as ConditionalExecution.
submit_fn (SubmissionFnType) – function to submit if_true with.

Job parameters:

left_side (str | float) – left side of the comparison.
right_side (str | float) – right side of the comparison.
comparator (str) – one of ValidOperators.
rel_tol (float) – relative tolerance of ~=. Defaults to DEFAULT_REL_TOL.
abs_tol (float) – absolute tolerance of ~=. Defaults to 0.
coerce (str) – auto, str or float; see IfElseParams.coerce. Defaults to auto.
if_true (JobConfig) – job to submit if the comparison is true.

Returns:

True, whether or not the job was submitted.

Return type:

bool

Raises:

ValueError – if coerce is float and a side isn’t a number.
re.error – if comparator is =~ and the right side isn’t a valid regular expression.

Example

from pyantz.jobs.branching.if_else import conditionally_execute

for log_line in ("all good", "ERROR: disk full"):
    submitted = []
    conditionally_execute(
        {
            "left_side": log_line,
            "comparator": "=~",
            "right_side": "^ERROR",
            "if_true": {
                "function": "pyantz.jobs.testing.nop.do_nothing",
                "parameters": {},
            },
        },
        submitted.append,
    )
    print(log_line, len(submitted))

all good 0
ERROR: disk full 1

if_else(params: IfElseParams, submit_fn: SubmissionFnType)[source]#

Submit a job based on a conditional value.

Either if_true or if_false is submitted, depending on whether left_side <comparator> right_side holds. The sides may use template variables, so the branch can depend on the results of earlier jobs.

Parameters:

params (IfElseParams) – the job’s parameters, validated as IfElseParams.
submit_fn (SubmissionFnType) – function to submit the chosen job with.

Job parameters:

left_side (str | float) – left side of the comparison.
right_side (str | float) – right side of the comparison.
comparator (str) – one of ValidOperators.
rel_tol (float) – relative tolerance of ~=. Defaults to DEFAULT_REL_TOL.
abs_tol (float) – absolute tolerance of ~=. Defaults to 0.
coerce (str) – auto, str or float; see IfElseParams.coerce. Defaults to auto.
if_true (JobConfig) – job to submit if the comparison is true.
if_false (JobConfig) – job to submit if the comparison is false.

Returns:

True once a job is submitted.

Return type:

bool

Raises:

ValueError – if coerce is float and a side isn’t a number.
re.error – if comparator is =~ and the right side isn’t a valid regular expression.

Example

from pyantz.jobs.branching.if_else import if_else

submitted = []
if_else(
    {
        "left_side": "10",
        "comparator": ">",
        "right_side": 9,
        "if_true": {
            "name": "bigger",
            "function": "pyantz.jobs.testing.nop.do_nothing",
            "parameters": {},
        },
        "if_false": {
            "name": "smaller",
            "function": "pyantz.jobs.testing.nop.do_nothing",
            "parameters": {},
        },
    },
    submitted.append,
)
print([job.name for job in submitted])

['bigger']

op_mapping: dict[ValidOperators, Callable[..., bool]] = {'!=': <built-in function ne>, '<': <built-in function lt>, '<=': <built-in function le>, '=': <built-in function eq>, '==': <built-in function eq>, '>': <built-in function gt>, '>=': <built-in function ge>}#: Function of each comparator which doesn’t need special handling.

Module contents#

Branching jobs create new jobs as they are run.

pyantz.jobs.branching package

Contents

pyantz.jobs.branching package#

Submodules#

pyantz.jobs.branching.case_matrix module#

pyantz.jobs.branching.if_else module#

Module contents#