hein_control.action package¶

Contains classes for managing and applying context to “actions” (functions).

Action is a function register which enables default arg/kwarg specification and retrieval of action by function name.

ConfiguredAction is an action that is configured to run (allows for multiple configurations of Action classes.

TrackedAction is an action that may be run with storage of context (duration, time started, function return, args/kwargs that were used for execution, and error details if any).

hein_control.action.basic module¶

Module containing base action classes used as the foundation of the package.

class hein_control.action.basic.Action(action: Callable, *default_args, parent_identifier: str = None, **default_kwargs)¶

Bases: hein_control.mixin.reg.InstanceRegistry

Base class for storing and tracking actions (functions) to be executed. Effectively this is a wrapper class which provides easy access to attributes relevant to automation (name, signature, kwarg defaults, etc.).

Parameters:

action – action (function) to execute when called
default_args – default arguments for the action. Effectively turns arguments into keyword arguments.
default_kwargs – default kwargs for the action. Overrides kwarg defaults of the provided action.
parent_identifier – identifier string of a parent of the action (used for action-instance associations where an action is specific to a parent instance)

UUID_PREFIX = 'action-'¶

classmethod action_is_registered(action: Union[str, Callable, Action], parent_identifier: str = None) → bool¶

Checks whether the provided action is registered as a Action instance.

Parameters:	action – action to check parent_identifier – parent identifier (parent instance association; if this is defined, the other instance must match both name and parent identifier)
Returns:	whether action is registered

compare_equality(other: Union[str, Action, Callable], parent_identifier: str = None) → bool¶

Performs a comparison between the other and this instance. Matches instances, action names, and callable names.

Parameters:	other – other instance to compare parent_identifier – parent identifier (parent instance association; if this is defined, the other instance must match both name and parent identifier)
Returns:	whether other matches self

classmethod get_registered_action_names() → List[str]¶: returns a list of the names of the registered Scheduler Actions

classmethod get_registered_actions() → List[hein_control.action.basic.Action]¶: returns a list of registered Action instances

name¶: action (function) name

classmethod register_action(action: Union[str, Callable, Action], parent_identifier: str = None) → hein_control.action.basic.Action¶

Registers a new instance or returns an existing instance if the action is already registered.

Parameters:	action – action method parent_identifier – identifier string of a parent of the action (used for action-instance associations where an action is specific to a parent instance)
Returns:	Action instance

classmethod register_actions(*actions) → List[hein_control.action.basic.Action]¶

Registers multiple actions at once

Parameters:	actions – action methods to be registered
Returns:	list of actions which were registered

registered_action_names = ['execute_sequence_actions', 'decision', 'run_automation']¶

registered_actions = [Action(execute_sequence_actions), Action(decision), Action(run_automation)]¶

return_type¶: the return type of the action

returns_bool¶: whether the action return type bool

class hein_control.action.basic.ActionList(*actions)¶

Bases: collections.abc.MutableSequence, typing.Generic

A manager for a list of Actions.

Parameters:	actions – actions to sequence

insert(index: int, action: hein_control.action.basic.Action) → None¶

Inserts the provided action into the Action list

Parameters:	index – index to insert at action – action object to insert
Returns:

class hein_control.action.basic.EvalAction(action: Callable, *default_args, **default_kwargs)¶

Bases: hein_control.action.basic.Action

Subclass of an Action where the action must return a bool. Note that this class does not enforce this, but warns the user if a non-bool type hint was detected.

Parameters:	action – action method to execute when called default_args – default arguments for the action default_kwargs – default kwargs for the action

hein_control.action.configured module¶

Module for describing and controlling actions which are configured for execution.

class hein_control.action.configured.ConfigurableInstance(instance_specific_actions: bool = False)¶

Bases: object

A base class which will register a class instance’s methods as ConfiguredActions. The method names in the methods_to_register list will be registered as configured actions upon instantiation.

Subclasses should overwrite the list of methods to register and call init.

Parameters:	instance_specific_actions – whether the configured instance should use instance-specific actions (e.g. the action depends on the state of the instance)

execute_action(action: str, *args, **kwargs) → hein_control.action.configured.TrackedAction¶

Executes an action associated with the instance with the provided args and kwargs.

Parameters:	action – action name to execute (must be a registered method of the instance) args – arguments to provide for the execution kwargs – keyword arguments to provide for the execution
Returns:	the generated TrackedAction instance

execute_action_threaded(action: str, *args, **kwargs) → hein_control.action.configured.ThreadedTrackedAction¶

Executes an action in a thread associated with the instance with the provided args and kwargs. This method is non-blocking.

Parameters:	action – action name to execute (must be a registered method of the instance) args – arguments to provide for the execution kwargs – keyword arguments to provide for the execution
Returns:	the generated ThreadedTrackedAction instance

methods_to_register = []¶

class hein_control.action.configured.ConfiguredAction(action: Union[Callable, hein_control.action.basic.Action, str], *args, configuration_name: str = None, next_action: Union[str, ConfiguredAction] = None, condition: Union[Tuple[object, str], Callable] = None, aliases: List[str] = None, use_unique_action: bool = False, parent_identifier_override: str = None, **kwargs)¶

Bases: hein_control.mixin.reg.InstanceRegistry, hein_control.mixin.config.Configuration, hein_control.mixin.status.Status

An action that is configured for execution. Arguments and Keyword arguments may be specified and will be handed to the action on execution unless overridden by the user. (Using kwargs over args is strongly recommended.)

Parameters:

action – action to perform (callable)
args – arguments for the action
configuration_name – optional name for the configuration (for key retrieval). This can be a useful way of differentiating specific configurations of an action if there are multiple configurations for that action.
next_action – optional next action for sequence building
condition – a condition evaulation which must be True for the execution to proceed. This may either be a callable which can be evaluated, or a pointer to some attribute of an object which will be checked at runtime for a bool.
kwargs – keyword arguments for the action
aliases – list of aliases which refer to the configured action instance
use_unique_action – whether to register action as a unique action (usually in cases where actions have stateful instance dependencies and there may be more than one action of the same name registered).
parent_identifier_override – Optional override for the parent identifier used for unique configured action generation.

UUID_PREFIX = 'ca-'¶

action¶: the action which is configured for execution

add_alias(alias: str)¶: adds a configuration name alias to the instance

as_dict() → dict¶: dictionary of relevant information

classmethod class_instance_by_id(identifier: str, parent_id: str = None) → hein_control.action.configured.ConfiguredAction¶

Retrieves the instance of the current class by UUID

Parameters:	identifier – unique identifier parent_id – optional specification of parent ID to retrieve instances specifically registered to a parent identifier.
Raises:	NameError – if no matching instance could be found
Returns:	registered class instance

duplicate_configuration(new_name: str = None) → hein_control.action.configured.ConfiguredAction¶

Duplicates a configuration with the provided name.

Parameters:	new_name – new name for the generated ConfiguredAction instance
Returns:	duplicated configuration

classmethod ensure_configuration(configuration: Union[str, Callable, hein_control.action.basic.Action, ConfiguredAction]) → hein_control.action.configured.ConfiguredAction¶

Ensures that the provided configuration is a class instance, instantiating or converting as necessary.

Parameters:	configuration – configuration instance, name, or key
Raises:	ValueError – if a string is provided and an action is not registered
Returns:	associated ConfiguredAction

evaluate_condition() → bool¶: Evaluates the condition (if specified) for the configuration

classmethod get_configuration_by_name(configuration_name: str) → hein_control.action.configured.ConfiguredAction¶

Retrieves a ConfiguredAction by name

Parameters:	configuration_name – name identifier of configuration
Returns:	ConfiguredAction

classmethod get_configuration_names() → List[str]¶

Retrieves the names of all configurations.

Warning: return will be sorted, but order does not correspond to order in registry (indexing based on this return is not recommended).

classmethod get_configurations_of_action(action: Union[str, hein_control.action.basic.Action]) → List[hein_control.action.configured.ConfiguredAction]¶

Retrieves all of the Configurations associated with the specified action

Parameters:	action – action to reference
Returns:	list of ConfiguredAction instances

get_tracked_from_config(**kwargs) → hein_control.action.configured.TrackedAction¶

Generates a tracked instance from the configured action

Parameters:	kwargs – additional kwargs that differ from the instance configuration
Returns:	TrackedAction (or subclass) instance.

match_name_and_parent(name: Union[Callable, str, ConfiguredAction], parent_identifier: str = None) → bool¶

Compares the provided name and parent identifier against the Configured Action instance (in effect an equality check)

Parameters:	name – name to use parent_identifier – parent identifier (optional)
Returns:	whether the name and parent identifier match

name¶: name for the configuration (useful for identifying a specific configuration of an action)

next_action¶: The next action in the sequence (for contextual action linking)

previous_actions¶: returns a list of configured actions which have this instance as a next step

remove_alias(alias: str)¶: removes an alias from the instance

classmethod remove_configuration_from_registry(configuration: Union[str, ConfiguredAction], parent_identifier: str = None)¶

Removes the specified configuration from the configuration registry

Parameters:	configuration – name or instance of the configuration parent_identifier – optional parent identifier

class hein_control.action.configured.ConfiguredActionList(*actions, use_unique_action: bool = False, parent_identifier_override: str = None)¶

Bases: hein_control.action.basic.ActionList, hein_control.mixin.serialization.DumpableList

A manager for a list of Configured Actions.

Parameters:	actions – actions to sequence use_unique_action – whether to use unique actions for any action added to this list parent_identifier_override – optional override for parent identifier used in ConfiguredAction unique identifier association.

action_registered_in_list(action: Callable) → bool¶

Checks whether the provided action is the associated action of any actions registered in the Configured List

Parameters:	action – action to check against

as_list() → list¶: returns a serializable list of relevant information about the configured actions

get_ca_by_action(action: Callable) → Optional[hein_control.action.configured.ConfiguredAction]¶

Gets the corresponding ConfiguredAction instance

Parameters:	action – gets a configured action from
Returns:

class hein_control.action.configured.ThreadedTrackedAction(configuration: Union[str, Callable, hein_control.action.basic.Action, hein_control.action.configured.ConfiguredAction], *args, **kwargs)¶

Bases: hein_control.action.configured.TrackedAction

A threaded version of a tracked action. This action may be executed asynchronously.

Parameters:	configuration – action to perform (callable) ConfiguredAction, next_step (Action,) – for creating a sequence of actions, the next_action is the action that should be executed after this action has been executed args – arguments for the action kwargs – keyword arguments for the action

UUID_PREFIX = 'tta-'¶

complete¶: whether the actions have completed

in_progress¶: whether the action is in progress (the action is in the process of being executed)

trigger(*args, **kwargs)¶

Triggers the action execution. Call order:

_trigger
_executor
TrackedAction superclass __call__

Parameters:	args – arguments for the call kwargs – kwargs for the call

wait_for_completion()¶: Waits for the completion of the time point

class hein_control.action.configured.TrackedAction(configuration: Union[str, Callable, hein_control.action.basic.Action, hein_control.action.configured.ConfiguredAction], *args, **kwargs)¶

Bases: hein_control.mixin.reg.InstanceRegistry, hein_control.mixin.config.Configuration, hein_control.mixin.status.Status

Creates a wrapped action which will track the time of start, completion, and return of the action.

Parameters:	configuration – configuration to reference or action to perform (callable) args – arguments for the instance’s action execution kwargs – keyword arguments for the instance’s action execution

UUID_PREFIX = 'ta-'¶

action¶: configured action associated with the TrackedAction instances

action_duration¶: task total duration (seconds)

action_return¶: the return of the method once complete

as_dict() → dict¶: dictionary of relevant information

configuration¶: the configuration associated with the tracked action

copy_configuration() → hein_control.action.configured.TrackedAction¶

Copies the configuration of a TrackedAction instance.

Returns:	TrackedAction instance with identical configuration

classmethod create_from_configured(source: Union[hein_control.action.basic.Action, hein_control.action.configured.ConfiguredAction]) → hein_control.action.configured.TrackedAction¶

Creates a TrackedAction instance from a provided Action or ConfiguredAction

Parameters:	source – action to generate from
Returns:	instantiated TrackedAction

method_return¶: DEPRECATED retrieve action return

next_action¶: next action to execute based on the configuration

started¶: whether the timepoint has been started

started_timestamp¶: timestamp for when the action was started

status_code¶: returns a status code representative of the state of the Action

time_completed¶: time stamp when the action was completed

time_started¶: time stamp when the action was started

trigger(*args, **kwargs)¶: trigger execution of the tracked action

triggered¶: whether the time point has been triggered

class hein_control.action.configured.TrackedActionList(*actions)¶

Bases: hein_control.action.basic.ActionList, hein_control.mixin.serialization.DumpableList

A manager for a list of Tracked Actions.

Parameters:	actions – actions to sequence

as_list() → list¶: if the dumpable is iterable, represents the list as a list

hein_control.action.configured.ensure_triggered(fn: Callable)¶: a decorator which ensures that the instance has been triggered

hein_control.action.logic module¶

class hein_control.action.logic.ConfiguredDecision(*logic_checks, configuration_name: str = None, **kwargs)¶

Bases: hein_control.action.configured.ConfiguredAction

Class for configuring a logic tree. The logic checks will iterate until an action return is True. When that condition is met, the next action of that logic check will be returned from the executor.

Parameters:	logic_checks – logic check actions. These must return a bool configuration_name – name for the configuration

UUID_PREFIX = 'cda-'¶

as_dict() → dict¶: dictionary of relevant information

get_tracked_from_config(**kwargs) → hein_control.action.logic.TrackedDecision¶

Generates a tracked instance from the configured action

Parameters:	kwargs – additional kwargs that differ from the instance configuration
Returns:	TrackedDecision instance.

logic_sequence¶: list of logic check actions to be performed

class hein_control.action.logic.TrackedDecision(configuration: Union[str, hein_control.action.logic.ConfiguredDecision], **kwargs)¶

Bases: hein_control.action.configured.TrackedAction

When called, given multiple ConfiguredActions, each of which their action must return a bool, sequentially run each ConfiguredAction, and for the first ConfiguredAction that returns True, return the next_action of the ConfiguredAction

Parameters:	args – ConfiguredActions that have actions that return a bool configuration_name – optional name for the configuration (for key retrieval)

UUID_PREFIX = 'da-'¶

as_dict() → dict¶: dictionary of relevant information

configuration¶: the configured sequence associated with the instance

copy_configuration() → hein_control.action.logic.TrackedDecision¶

Copies the configuration of a ExecutionSequence instance.

Returns:	ExecutionSequence instance with identical configuration

next_action¶: next action as determined by the logic tree

hein_control.action.operators module¶

functions for execution of specific actions