flowno.utilities.helpers

Generation tracking utilities for Flowno’s dataflow execution system.

This module provides helper functions for working with generation tuples, which are used throughout Flowno to track execution order, dependency resolution, and manage streaming data in dataflow graphs.

Generation tuples are ordered sequences of integers (e.g., (0,), (1, 0), (2, 1, 3)) that version the data produced by nodes. Each position represents a different run level, where: - The first element (main_gen) tracks the primary execution count - Later elements track nested levels for streaming or partial results

See the flowno.core.flow.flow module for more information on how generations are used in Flowno’s execution model.

flowno.utilities.helpers.clip_generation(gen: tuple[int, ...] | None, run_level: int) → tuple[int, ...] | None[source]

Clip a generation tuple to be compatible with a specific run level.

This function returns the “highest” generation (according to cmp_generation) that is less than or equal to gen and has a length of at most run_level + 1. It’s used to determine if a node with streaming capabilities should wait for more data or can proceed with what’s available.

Parameters:

  • gen – The generation tuple to clip

  • run_level – The run level to clip to

Returns:

A clipped generation tuple, or None if no suitable generation exists

Examples

>>> clip_generation((0, 0), 3)
(0, 0)      # Already compatible with run_level 3
>>> clip_generation((0, 0), 2)
(0, 0)      # Already compatible with run_level 2
>>> clip_generation((0, 0), 0)
None        # No compatible generation exists
>>> clip_generation((1, 0), 1)
(1, 0)      # Already compatible with run_level 1
>>> clip_generation((1, 0), 0)
(0,)        # Clipped to run_level 0
flowno.utilities.helpers.cmp_generation(gen_a: tuple[int, ...] | None, gen_b: tuple[int, ...] | None) → Literal[-1, 0, 1][source]

Compare two generation tuples according to Flowno’s ordering rules.

The ordering follows these principles:

  1. None values are considered “newest” (not yet run)

  2. For non-None tuples, lexicographical comparison is used first

  3. If lexicographical comparison gives equality but lengths differ, shorter tuples are considered “greater” (more final) than longer ones

Parameters:

  • gen_a – First generation tuple to compare

  • gen_b – Second generation tuple to compare

Returns:

-1 if gen_a < gen_b (gen_a comes before gen_b) 0 if gen_a == gen_b 1 if gen_a > gen_b (gen_a comes after gen_b)

Examples

>>> cmp_generation(None, None)
0
>>> cmp_generation(None, (0,))
-1  # None comes before any tuple
>>> cmp_generation((1,), (0,))
1   # (1,) > (0,)
>>> cmp_generation((0,), (0, 0))
1   # Shorter tuple is considered greater
>>> cmp_generation((0, 1), (0, 2))
-1  # Lexicographical comparison
flowno.utilities.helpers.inc_generation(gen: tuple[int, ...] | None, run_level: int = 0) → tuple[int, ...][source]

Increment the generation at the specified run level.

Computes the minimal generation greater than gen according to cmp_generation. This is used to calculate the next generation when a node runs.

Parameters:

  • gen – The current generation to increment, or None

  • run_level – The index within the generation tuple to increment

Returns:

A new generation tuple that is minimally greater than the input

Raises:

ValueError – If no generation greater than the input can be found

Examples

>>> inc_generation(None, 0)
(0,)        # First generation at run level 0
>>> inc_generation((0,), 0)
(1,)        # Increment run level 0
>>> inc_generation((1,), 1)
(1, 0)      # First generation at run level 1
>>> inc_generation((1, 0), 1)
(1, 1)      # Increment run level 1
>>> inc_generation((0, 0), 2)
(0, 0, 0)   # First generation at run level 2
flowno.utilities.helpers.parent_generation(gen: tuple[int, ...] | None) → tuple[int, ...] | None[source]

Return the parent generation of the given generation tuple.

The parent generation is created by removing the last element of the generation tuple, representing moving up one run level in the hierarchy.

Parameters:

gen – The generation tuple to find the parent of

Returns:

The parent generation, or None if gen is None or empty

Examples

>>> parent_generation(None)
None
>>> parent_generation((1, 2, 3))
(1, 2)
>>> parent_generation((1,))
()
>>> parent_generation(())
None
flowno.utilities.helpers.stitched_generation(gen: tuple[int, ...] | None, stitch_0: int) → tuple[int, ...] | None[source]

Apply a “stitch” adjustment to a generation tuple.

This function is used for cycle breaking in dataflow graphs. It adds the stitch value to the first element of the generation tuple, which affects how nodes in a cycle will be scheduled.

Parameters:

  • gen – The generation tuple to stitch

  • stitch_0 – Value to add to the first element

Returns:

The modified generation tuple, or a special value for None input

Examples

>>> stitched_generation(None, 0)
None
>>> stitched_generation(None, 1)
(0,)
>>> stitched_generation((1, 2), 3)
(4, 2)
>>> stitched_generation((), 5)
()