forest_helper module

class forest_helper.PersistentWorkerPool(workers: int, blob_bytes: bytes | None = None, serialized_funcs: dict[str, Any] | None = None)

Bases: object

Context manager for a persistent ProcessPoolExecutor that initializes workers with ForestModel and coeff_funcs.

Constructor

Parameters:

• workers (int) – Number of workers

• blob_bytes (bytes, optional) – Serialized ForestModel objects, defaults to None

• serialized_funcs (dict[str, function], optional) – dict of serialzed coeff_funcs functions, defaults to None

forest_helper.auto_batch(tasks: list[Any], workers: int, max_batch_factor: int | None = None, size_fn: Callable[[Any], float] | None = None) → list[list[Any]]

Split tasks into batches for parallel processing. Optionally sorts tasks by size (descending) and greedily fills batches.

Parameters:

• tasks (list) – List of tasks to batch

• workers (int) – Number of workers (cores) that will be used later to process batches

• max_batch_factor (int, optional) – Scaling parameter (larger value yields more smaller batches), defaults to None

• size_fn (function, optional) – Task size estimation function returning float for greedy task sort, defaults to lambda x: 1.

Returns:

List of task batches

Return type:

list[list]

forest_helper.choose_max_batch_factor(workers: int) → int

Adaptive max_batch_factor for auto_batch based on number of workers.

Parameters:

workers (int) – Number of worker processes. Integer value representing the number of worker processes available.

Returns:

Optimized max_batch_factor value. An integer value that represents the optimized max_batch_factor value based on the number of workers.

Return type:

int

Usage notes:

• This function is designed to work with auto_batch, which controls batch sizes for parallel processing.

• The function’s output is an integer representing the optimal max_batch_factor value for a given number of worker processes.

Examples:

>>> choose_max_batch_factor(1)
2

>>> choose_max_batch_factor(8)
4

Edge case warnings:

• If workers <= 0, this function will raise a ValueError.

forest_helper.init_worker_gen_vars(blob_bytes_local: bytes, serialized_funcs_local: dict[str, bytes], workers: int = 1) → None

Initializer for _gen_vars_m1 workers: load model and coefficient functions once. Also stores desired worker count for _bld_tree_m1.

Parameters:

• blob_bytes_local (bytes) – Serialized ForestModel object

• serialized_funcs_local (dict[str, bytes]) – dict of serialized functions keyed on coeff_funcs keys

• workers (int, optional) – Number of workers, defaults to 1

forest_helper.sanitize_func(f: Any) → Any

Make a version of f that is safe to serialize via dill in spawn mode

Parameters:

f (function) – Function to sanitize

Returns:

Sanitized function

Return type:

function

forest_helper.worker_cmp_cflw_batch(args: list[Any]) → list[tuple[int, str, tuple[Any, ...], tuple[Any, ...], float]]

Worker function to process batches of tasks for _cmp_cflw_m1

Parameters:

args (list[list, dict, list]) – (batch, cflw_keys, periods)

Returns:

list of (t, o, i, j, value) tuples

Return type:

list[int, str, tuple, tuple, float]

forest_helper.worker_cmp_cflw_phase3(args: tuple[int, str, dict[Any, float], dict[Any, float], Any, Any, list[str]]) → list[tuple[str, dict[str, float], str, float]]

Worker function to compute (name, coeffs, sense, rhs) tuples for Phase 3 of _cmp_cflw_m1.

Parameters:

args (tuple(int, str, dict, dict, float|None, float|None, list[str])) – (t, o, mu_t_o, mu_ref_o, alpha, beta, xnames), where alpha is the maximum fractional period-over-period decrease (lower-bound row; None omits it) and beta is the maximum fractional increase (upper-bound row; None omits it).

Returns:

list of (constraint_name, mu_lb, sense, 0.) tuples

Return type:

list[(str, float, str, float)]

forest_helper.worker_cmp_cflw_phase3_batch(batch: list[tuple[int, str, dict[Any, float], dict[Any, float], Any, Any, list[str]]]) → list[tuple[str, dict[str, float], str, float]]

Worker function to process batches of phase 3 tasks for _cmp_cflw_m1

Parameters:

batch (list[tuple]) – list of tasks (tuples)

Returns:

list of results

Return type:

list[tuple]

forest_helper.worker_cmp_cgen_batch(args: list[Any]) → list[tuple[int, str, tuple[Any, ...], tuple[Any, ...], float]]

Worker function to process batches of tasks for _cmp_cgen_m1

Parameters:

args (list[list, dict, list]) – (batch, cgen_keys, periods)

Returns:

list of (t, o, i, j, value) tuples

Return type:

list[int, str, tuple, tuple, float]

forest_helper.worker_cmp_cgen_phase3(args: tuple[int, str, dict[Any, float], dict[Any, float] | None, dict[Any, float] | None]) → list[tuple[str, dict[str, float], str, float]]

Args: (t, o, mu_t_o, lb, ub) Returns: [(name, coeffs, sense, rhs), …]

forest_helper.worker_cmp_cgen_phase3_batch(batch: list[tuple[int, str, dict[Any, float], dict[Any, float] | None, dict[Any, float] | None]]) → list[tuple[str, dict[str, float], str, float]]

Process a batch of Phase 3 CGEN tasks.

forest_helper.worker_gen_vars(tasks: list[tuple[str, int]], acodes: list[str]) → list[tuple[str, int, Any]]

Worker for building trees in _gen_vars_m1.

Parameters:

• tasks (list[(str, ...), int)]) – list of (dtk, age) tuples to process

• acodes (list[str]) – list of action codes to use when building trees

Returns:

list of (dtk, age, tree) tuples

Return type:

list[(str, …), int, Tree]

forest_helper.worker_summarize_tree_batch(args: list[Any]) → list[tuple[str, dict[str, float], dict[str, float]]]

Summarize a batch of trees into coverage constraints and leaf outputs.

Parameters:

args (list) – [batch, z_coeff_key]

Returns:

[(cname, coeffs, z_coeffs), …]

Return type:

list[list]