src.services.routine_service package

Submodules

src.services.routine_service.executor module

src.services.routine_service.executor.execute_routine(routines, routine_slug, routine_params=None)

src.services.routine_service.executor.execute_with_middleware(step: Dict, middleware_chain, execute_step_fn: Callable) → Any

Build and execute a nested chain of middleware around the actual step function.

src.services.routine_service.executor.resolve_parameters(value: Any, routine_params: Dict, globals_dict: Dict) → Any

Recursively search for strings containing parameter references like “{param_name}” and replace them with the corresponding value from: 1. First from routine_params (highest priority) 2. Then from globals_dict if not found in routine_params

src.services.routine_service.executor.resolve_references(value: Any, routine_name: str, step_results: Dict) → Any

Recursively search for strings containing “prev(…)” in the value (which may be a string, list, or dict) and replace them with the corresponding data from step_results.

src.services.routine_service.executor.wrap_middleware(mw_entry: Dict) → Callable

Given a middleware configuration entry (either a string or a dict with “name” and “args”), return a callable that takes (step, next_callable) and applies the middleware with the provided arguments.

src.services.routine_service.helper_methods module

src.services.routine_service.helper_methods.evaluate_expression(expression: str | bool) → Any

Evaluate a mathematical expression.

Parameters:

expression (str) – Expression to evaluate.

Returns:

Result of the evaluated expression.

Return type:

Any

src.services.routine_service.helper_methods.execute_python_script(script_path: str, kwargs) → Any

Execute a Python script with optional keyword arguments.

Parameters:

• script_path (str) – Path to the Python script to execute.

• **kwargs – Keyword arguments to pass to the script.

Returns:

Result of the script execution.

Return type:

Any

src.services.routine_service.helper_methods.execute_routine(routine_name: str, routine_params: Dict[str, Any] | None = None) → Any

Execute a routine by name with optional parameters.

Parameters:

• routine_name (str) – Name of the routine to execute.

• routine_params (dict, optional) – Parameters to pass to the routine. Defaults to None. These parameters will override any default values defined in the routine. Example: {“frequency”: “2MHz”}

Returns:

Result of the routine execution.

Return type:

Any

src.services.routine_service.helper_methods.output_values(values: list) → tuple[bool, list]

src.services.routine_service.helper_methods.pause(seconds: float) → None

Pause execution for a specified number of seconds.

Parameters:

seconds (float) – Number of seconds to pause for.

src.services.routine_service.helper_methods.raise_exception(e)

src.services.routine_service.loader module

src.services.routine_service.loader.get_global_params()

Get all global parameters from loaded configuration files

src.services.routine_service.loader.load_routines_from_yaml(yaml_path)

src.services.routine_service.middleware module

src.services.routine_service.middleware.metrics_middleware(step, next_callable)

Measures the execution time of a step.

src.services.routine_service.middleware.pause_after_execution(step, next_callable, seconds=0.1)

Pause execution after a step.

src.services.routine_service.middleware.pause_before_exection(step, next_callable, seconds=0.1)

Pause before execution of a step.

src.services.routine_service.middleware.repeat_middleware(step, next_callable, repeat_count=1)

Repeats the step execution a specified number of times.

src.services.routine_service.middleware.retry_middleware(step, next_callable, retries=3)

Retries the step execution if an exception occurs.

Module contents