`structfs.runner`#

Run functions in a module in a defined, named, sequence.

Module Contents#

Classes#

`Step`	Step to run by the `Runner`.
`Runner`	Run a set of defined steps.
`Action`	The Action class implements a simple framework to run arbitrary functions before and/or after each step and/or run performed by the `Runner` class.

Data#

`__author__`
`_log`
`ActionType`

API#

structfs.runner.__author__: ‘Dan Gunter (LBNL)’

structfs.runner._log: ‘Logger(…)’

class structfs.runner.Step(name: str, func: Callable)

Step to run by the Runner.

Initialization

Constructor

Args: name: Name of the step func: Function to call to execute the step

SEP: ‘::’

add_substep(name: str, func: Callable)

Add a sub-step to this step. Substeps are run in the order given.

Args: name: The name of substep func: Function to call to execute this substep

structfs.runner.ActionType: ‘TypeVar(…)’

class structfs.runner.Runner(steps: Sequence[str])

Run a set of defined steps.

Initialization

Constructor.

Args: steps: List of step names

STEP_ANY: ‘-’

__getitem__(key): Look for key in context

add_step(name: str, func: Callable)

Add a step.

Steps are executed by calling func(context), where context is a dict (or dict-like) object that is used to pass state between steps.

Args: name: Add a step to be executed func: Function to execute for the step.

Raises: KeyError: description

add_substep(base_name, name, func)

Add a substep for a given step.

Substeps are all executed, in the order added, immediately after their base step is executed.

Args: base_name: Step name name: Substep name func: Function to execute

Raises: KeyError: Base step or substep is not found ValueError: Base step does not have any substeps

run_step(name): Syntactic sugar for calling run_steps for a single step.

run_steps(first: str = '', last: str = '', after: str = '', before: str = '')

Run steps from first/after to step last/before.

Specify only one of the first/after and last/before pairs.

Use the special value STEP_ANY to mean the first or last defined step.

Args: first: First step to run (include) after: Run first defined step after this one (exclude) last: Last step to run (include) before: Run last defined step before this one (exclude)

Raises: KeyError: Unknown or undefined step given ValueError: Steps out of order or both first/after or before/last given

_run_steps(first: str, last: str, endpoints: tuple[bool, bool])

reset(): Reset runner internal state, especially the context.

list_steps(all_steps=False) → list[str]: Get list of [runnable] steps.

add_action(name: str, action_class: type, *args, **kwargs) → object

Add a named action.

Args: name: Arbitrary name for the action, used to get/remove it action_class: Subclass of Action to use args: Positional arguments passed to action_class constructor kwargs: Keyword arguments passed to action_class constructor

get_action(name: str) → structfs.runner.ActionType

Get an action object.

Args: name: Name of action (as provided to add_action)

Returns: ActionType: Action object

Raises: KeyError: If action name does not match any known action

remove_action(name: str)

Remove an action object.

Args: name: Name of action (as provided to add_action)

Raises: KeyError: If action name does not match any known action

_find_step(reverse=False)

classmethod normalize_name(s: Optional[str]) → str

Normalize a step name. Args: s: Step name

Returns: normalized name

_step_begin(name: str)

_substep_begin(base: str, name: str)

_step_end(name: str)

_substep_end(base: str, name: str)

step(name: str)

Decorator function for creating a new step.

Args: name: Step name

Returns: Decorator function.

substep(base: str, name: str)

Decorator function for creating a new substep.

Substeps are not run directly, and must have an already existing base step as their parent.

Args: base: Base step name name: Substep name

Returns: Decorator function.

report() → dict[str, dict]

Compile reports of each action into a combined report

Returns: dict: Mapping with two key-value pairs: - actions: Keys are names given to actions during add_action(), values are the reports returned by that action, in Python dictionary form. - last_run: List of steps (names, as strings) in previous run

class structfs.runner.Action(runner: structfs.runner.Runner, log: Optional[logging.Logger] = None)

Bases: abc.ABC

The Action class implements a simple framework to run arbitrary functions before and/or after each step and/or run performed by the Runner class.

To create and use your own Action, inherit from this class and then define one or more of the methods:

before_step - Called before a given step is executed
after_step - Called after a given step is executed
before/after_substep - Called before/after a named substep is executed (these can have arbitrary names)
before_run - Called before the first step is executed
after_run - Called after the last step is executed

Then add the action to the Runner class (e.g., FlowsheetRunner) instance with add_action(). Note that you pass the action class, not instance. Additional settings can be passed to the created action instance with arguments to add_action. Also note that the name argument is used to retrieve the action instance later, as needed.

All actions must also implement the report() method, which returns the results of the action to the caller as either a Pydantic BaseModel subclass or a Python dict.

Example

Below is a simple example that prints a message before/after every step and prints the total number of steps run at the end of the run.

Runner HelloGoodbye class#

from idaes.core.util.structfs.runner import Action
class HelloGoodbye(Action):
    "Example action, for tutorial purposes."

    def __init__(self, runner, hello="hi", goodbye="bye", **kwargs):
        super().__init__(runner, **kwargs)
        self._hello, self._goodbye = hello, goodbye
        self.step_counter = -1

    def before_run(self):
        self.step_counter = 0

    def before_step(self, name):
        print(f">> {self._hello} from step {name}")

    def before_substep(self, name, subname):
        print(f"  >> {self._hello} from sub-step {subname}")

    def after_step(self, name):
        print(f"<< {self._goodbye} from step {name}")
        self.step_counter += 1

    def after_substep(self, name, subname):
        print(f"  << {self._goodbye} from sub-step {subname}")

    def after_run(self):
        print(f"Ran {self.step_counter} steps")

    def report(self):
        return {"steps": self.step_counter}

You could add the above example to a Runner subclass, here called my_runner, like this:

my_runner.add_action(
    "hg",
    HelloGoodbye,
    hello="Greetings and salutations",
    goodbye="Smell you later",
)

Then, after running steps, you could print the value of the step_counter attribute with:

print(my_runner.get_action("hg").step_counter)

See the pre-defined actions in the runner_actions module, and their usage in the FlowsheetRunner class, for more examples.

Initialization

Constructor

Args: runner: Reference to the runner that will trigger this action. log: Logger to use when logging informational or error messages

before_step(step_name: str)

Perform this action before the named step.

Args: step_name: Name of the step

before_substep(step_name: str, substep_name: str)

Perform this action before the named sub-step.

Args: step_name: Name of the step substep_name: Name of the sub-step

after_step(step_name: str)

Perform this action after the named step.

Args: step_name: Name of the step

after_substep(step_name: str, substep_name: str)

Perform this action after the named sub-step.

Args: step_name: Name of the step substep_name: Name of the sub-step

before_run(): Perform this action before a run starts.

after_run(): Perform this action after a run ends.

abstractmethod report() → pydantic.BaseModel | dict

Report the results of the action to the caller.

Returns: Results as a Pydantic model or Python dict

structfs.runner

Contents

structfs.runner#

Module Contents#

Classes#

Data#

API#

`structfs.runner`#