AlgorithmMachine Class Reference

Inheritance diagram for AlgorithmMachine:

Public Member Functions
	__init__ (self, algorithm=None, initial_state="init")
	setup_from_dict (self, params)
	is_valid (self)
	initial_state (self)
	initial_state (self, state)
	state (self)
	state (self, state)
	add_state (self, state, enter=None, exit=None)
	add_transition (self, trigger, source, dest, conditions=None, before=None, after=None)
	__getattr__ (self, name, **kwargs)
	get_transitions (self, source)
	get_transition_dict (self, state, transition)
	save_graph (self, filename, graphname)

Static Public Member Functions
	default_condition (**kwargs)

Public Attributes
list	default_states
	Default states for the AlgorithmMachine.
	algorithm = algorithm
	Algorithm() object whose state we are modelling.
list	input_files = []
	Collector output files, will contain all files returned by the output patterns.
list	dependent_databases = []
	CAF created local databases from previous calibrations that this calibration/algorithm depends on.
list	database_chain = []
	Assigned database chain to the overall Calibration object, or to the 'default' Collection.
str	output_dir = ""
	The algorithm output directory which is mostly used to store the stdout file.
str	output_database_dir = ""
	The output database directory for the localdb that the algorithm will commit to.
	result = None
	IoV_Result object for a single execution, will be reset upon a new execution.
dict	states = {}
	Valid states for this machine.
	initial_state = initial_state
	Pointless docstring since it's a property.
	transitions = defaultdict(list)
	Allowed transitions between states.
	state = dest
	Current State of machine.

Static Public Attributes
list	required_attrs
	Required attributes that must exist before the machine can run properly.
list	required_true_attrs
	Attributes that must have a value that returns True when tested.

Protected Member Functions
	_create_output_dir (self, **kwargs)
	_setup_database_chain (self, **kwargs)
	_setup_logging (self, **kwargs)
	_change_working_dir (self, **kwargs)
	_pre_algorithm (self, **kwargs)
	_execute_over_iov (self, **kwargs)
	_set_input_data (self, **kwargs)
	_trigger (self, transition_name, transition_dict, **kwargs)

Static Protected Member Functions
	_callback (func, **kwargs)

Protected Attributes
dict	_initial_state = State(initial_state)
	Actual attribute holding initial state for this machine.
dict	_state = self.initial_state
	Actual attribute holding the Current state.

Detailed Description

A state machine to handle the logic of running the algorithm on the overall runs contained in the data.

Definition at line 935 of file state_machines.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
			algorithm = None,
			initial_state = "init" )

Takes an Algorithm object from the caf framework and defines the transitions.

Definition at line 957 of file state_machines.py.

    def __init__(self, algorithm=None, initial_state="init"):
        """
        Takes an Algorithm object from the caf framework and defines the transitions.
        """
        
        self.default_states = [State("init"),
                               State("ready"),
                               State("running_algorithm"),
                               State("completed"),
                               State("failed")]
 
        super().__init__(self.default_states, initial_state)
 
        
        self.algorithm = algorithm
        
        self.input_files = []
        
        self.dependent_databases = []
        
        self.database_chain = []
        
        self.output_dir = ""
        
        self.output_database_dir = ""
        
        self.result = None
 
        self.add_transition("setup_algorithm", "init", "ready",
                            before=[self._setup_logging,
                                    self._change_working_dir,
                                    self._setup_database_chain,
                                    self._set_input_data,
                                    self._pre_algorithm])
        self.add_transition("execute_runs", "ready", "running_algorithm",
                            after=self._execute_over_iov)
        self.add_transition("complete", "running_algorithm", "completed")
        self.add_transition("fail", "running_algorithm", "failed")
        self.add_transition("fail", "ready", "failed")
        self.add_transition("setup_algorithm", "completed", "ready")
        self.add_transition("setup_algorithm", "failed", "ready")
 

Member Function Documentation

__getattr__()

__getattr__ ( self, name, ** kwargs )

inherited

Allows us to create a new method for each trigger on the fly.
If there is no trigger name in the machine to match, then the normal
AttributeError is called.

Definition at line 306 of file state_machines.py.

    def __getattr__(self, name, **kwargs):
        """
        Allows us to create a new method for each trigger on the fly.
        If there is no trigger name in the machine to match, then the normal
        AttributeError is called.
        """
        possible_transitions = self.get_transitions(self.state)
        if name not in possible_transitions:
            raise AttributeError(f"{name} does not exist in transitions for state {self.state}.")
        transition_dict = self.get_transition_dict(self.state, name)
        # \cond silence doxygen warning about _trigger
        return partial(self._trigger, name, transition_dict, **kwargs)
        # \endcond
 

_callback()

_callback ( func, ** kwargs )

staticprotectedinherited

Calls a condition/before/after.. function using arguments passed (or not).

Definition at line 344 of file state_machines.py.

    def _callback(func, **kwargs):
        """
        Calls a condition/before/after.. function using arguments passed (or not).
        """
        return func(**kwargs)
 

_change_working_dir()

_change_working_dir ( self, ** kwargs )

protected

Definition at line 1082 of file state_machines.py.

    def _change_working_dir(self, **kwargs):
        """
        """
        B2INFO(f"Changing current working directory to {self.output_dir}.")
        os.chdir(self.output_dir)
 

_create_output_dir()

_create_output_dir ( self, ** kwargs )

protected

Create working/output directory of algorithm. Any old directory is overwritten.

Definition at line 1026 of file state_machines.py.

    def _create_output_dir(self, **kwargs):
        """
        Create working/output directory of algorithm. Any old directory is overwritten.
        """
        create_directories(Path(self.output_dir), overwrite=True)
 

_execute_over_iov()

_execute_over_iov ( self, ** kwargs )

protected

Does the actual execute of the algorithm on an IoV and records the result.

Definition at line 1098 of file state_machines.py.

    def _execute_over_iov(self, **kwargs):
        """
        Does the actual execute of the algorithm on an IoV and records the result.
        """
        B2INFO(f"Running {self.algorithm.name} in working directory {os.getcwd()}.")
 
        runs_to_execute = kwargs["runs"]
        iov = kwargs["apply_iov"]
        iteration = kwargs["iteration"]
        if not iov:
            iov = iov_from_runs(runs_to_execute)
        B2INFO(f"Execution will use {iov} for labelling payloads by default.")
        alg_result = self.algorithm.algorithm.execute(runs_to_execute, iteration, iov._cpp_iov)
        self.result = IoV_Result(iov, alg_result)
 

_pre_algorithm()

_pre_algorithm ( self, ** kwargs )

protected

Call the user defined algorithm setup function.

Definition at line 1088 of file state_machines.py.

    def _pre_algorithm(self, **kwargs):
        """
        Call the user defined algorithm setup function.
        """
        B2INFO("Running Pre-Algorithm function (if exists)")
        if self.algorithm.pre_algorithm:
            # We have to re-pass in the algorithm here because an outside user has created this method.
            # So the method isn't bound to the instance properly.
            self.algorithm.pre_algorithm(self.algorithm.algorithm, kwargs["iteration"])
 

_set_input_data()

_set_input_data ( self, ** kwargs )

protected

set input data

Definition at line 1113 of file state_machines.py.

    def _set_input_data(self, **kwargs):
        """set input data"""
        self.algorithm.data_input(self.input_files)
 
 

_setup_database_chain()

_setup_database_chain ( self, ** kwargs )

protected

Apply all databases in the correct order.

Definition at line 1032 of file state_machines.py.

    def _setup_database_chain(self, **kwargs):
        """
        Apply all databases in the correct order.
        """
        # We deliberately override the normal database ordering because we don't want input files GTs to affect
        # the processing. Only explicit GTs and intermediate local DBs made by the CAF should be added here.
        b2conditions.reset()
        b2conditions.override_globaltags()
 
        # Apply all the databases in order, starting with the user-set chain for this Calibration
        # Local variable avoids doxygen "no uniquely matching class member" warning
        database_chain = self.database_chain
        for database in database_chain:
            if database.db_type == 'local':
                B2INFO(f"Adding Local Database {database.filepath.as_posix()} to head of chain of local databases, "
                       f"for {self.algorithm.name}.")
                b2conditions.prepend_testing_payloads(database.filepath.as_posix())
            elif database.db_type == 'central':
                B2INFO(f"Adding Central database tag {database.global_tag} to head of GT chain, "
                       f"for {self.algorithm.name}.")
                b2conditions.prepend_globaltag(database.global_tag)
            else:
                raise ValueError(f"Unknown database type {database.db_type}.")
        # Here we add the finished databases of previous calibrations that we depend on.
        # We can assume that the databases exist as we can't be here until they have returned
        # with OK status.
        for filename, directory in self.dependent_databases:
            B2INFO(f"Adding Local Database {filename} to head of chain of local databases created by"
                   f" a dependent calibration, for {self.algorithm.name}.")
            b2conditions.prepend_testing_payloads(filename)
 
        # Create a directory to store the payloads of this algorithm
        create_directories(Path(self.output_database_dir), overwrite=False)
 
        # add local database to save payloads
        B2INFO(f"Output local database for {self.algorithm.name} stored at {self.output_database_dir}.")
        # Things have changed. We now need to do the expert settings to create a database directly.
        # LocalDB is readonly without this but we don't need 'use_local_database' during writing.
        b2conditions.expert_settings(save_payloads=str(self.output_database_dir.joinpath("database.txt")))
 

_setup_logging()

_setup_logging ( self, ** kwargs )

protected

Definition at line 1072 of file state_machines.py.

    def _setup_logging(self, **kwargs):
        """
        """
        # add logfile for output
        log_file = os.path.join(self.output_dir, self.algorithm.name + '_stdout')
        B2INFO(f"Output log file at {log_file}.")
        basf2.reset_log()
        basf2.set_log_level(basf2.LogLevel.INFO)
        basf2.log_to_file(log_file)
 

_trigger()

_trigger ( self, transition_name, transition_dict, ** kwargs )

protectedinherited

Runs the transition logic. Callbacks are evaluated in the order:
conditions -> before -> <new state set here> -> after.

Definition at line 320 of file state_machines.py.

    def _trigger(self, transition_name, transition_dict, **kwargs):
        """
        Runs the transition logic. Callbacks are evaluated in the order:
        conditions -> before -> <new state set here> -> after.
        """
        dest, conditions, before_callbacks, after_callbacks = (
            transition_dict["dest"],
            transition_dict["conditions"],
            transition_dict["before"],
            transition_dict["after"]
        )
        # Returns True only if every condition returns True when called
        if all(map(lambda condition: self._callback(condition, **kwargs), conditions)):
            for before_func in before_callbacks:
                self._callback(before_func, **kwargs)
            
            self.state = dest
            for after_func in after_callbacks:
                self._callback(after_func, **kwargs)
        else:
            raise ConditionError(f"Transition '{transition_name}' called for but one or more conditions "
                                 "evaluated False")
 

add_state()

add_state ( self, state, enter = None, exit = None )

inherited

Adds a single state to the list of possible ones.
Should be a unique string or a State object with a unique name.

Definition at line 193 of file state_machines.py.

    def add_state(self, state, enter=None, exit=None):
        """
        Adds a single state to the list of possible ones.
        Should be a unique string or a State object with a unique name.
        """
        if isinstance(state, str):
            self.add_state(State(state, enter, exit))
        elif isinstance(state, State):
            if state.name not in self.states.keys():
                self.states[state.name] = state
            else:
                B2WARNING(f"You asked to add a state {state} but it was already in the machine states.")
        else:
            B2WARNING(f"You asked to add a state {state} but it wasn't a State or str object")
 

add_transition()

add_transition ( self, trigger, source, dest, conditions = None, before = None, after = None )

inherited

Adds a single transition to the dictionary of possible ones.
Trigger is the method name that begins the transition between the
source state and the destination state.

The condition is an optional function that returns True or False
depending on the current state/input.

Definition at line 264 of file state_machines.py.

    def add_transition(self, trigger, source, dest, conditions=None, before=None, after=None):
        """
        Adds a single transition to the dictionary of possible ones.
        Trigger is the method name that begins the transition between the
        source state and the destination state.
 
        The condition is an optional function that returns True or False
        depending on the current state/input.
        """
        transition_dict = {}
        try:
            source = self.states[source]
            dest = self.states[dest]
            transition_dict["source"] = source
            transition_dict["dest"] = dest
        except KeyError as err:
            B2WARNING("Tried to add a transition where the source or dest isn't in the list of states")
            raise err
        if conditions:
            if isinstance(conditions, (list, tuple, set)):
                transition_dict["conditions"] = list(conditions)
            else:
                transition_dict["conditions"] = [conditions]
        else:
            transition_dict["conditions"] = [Machine.default_condition]
 
        if not before:
            before = []
        if isinstance(before, (list, tuple, set)):
            transition_dict["before"] = list(before)
        else:
            transition_dict["before"] = [before]
 
        if not after:
            after = []
        if isinstance(after, (list, tuple, set)):
            transition_dict["after"] = list(after)
        else:
            transition_dict["after"] = [after]
 
        self.transitions[trigger].append(transition_dict)
 

default_condition()

default_condition ( ** kwargs )

staticinherited

Method to always return True.

Definition at line 258 of file state_machines.py.

    def default_condition(**kwargs):
        """
        Method to always return True.
        """
        return True
 

get_transition_dict()

get_transition_dict ( self, state, transition )

inherited

Returns the transition dictionary for a state and transition out of it.

Definition at line 361 of file state_machines.py.

    def get_transition_dict(self, state, transition):
        """
        Returns the transition dictionary for a state and transition out of it.
        """
        transition_dicts = self.transitions[transition]
        for transition_dict in transition_dicts:
            if transition_dict["source"] == state:
                return transition_dict
        else:
            raise KeyError(f"No transition from state {state} with the name {transition}.")
 

get_transitions()

get_transitions ( self, source )

inherited

Returns allowed transitions from a given state.

Definition at line 350 of file state_machines.py.

    def get_transitions(self, source):
        """
        Returns allowed transitions from a given state.
        """
        possible_transitions = []
        for transition, transition_dicts in self.transitions.items():
            for transition_dict in transition_dicts:
                if transition_dict["source"] == source:
                    possible_transitions.append(transition)
        return possible_transitions
 

initial_state() [1/2]

initial_state ( self )

inherited

The initial state of the machine. Needs a special property to prevent trying to run on_enter callbacks when set.

Definition at line 209 of file state_machines.py.

    def initial_state(self):
        """
        The initial state of the machine. Needs a special property to prevent trying to run on_enter callbacks when set.
        """
        return self._initial_state
 

initial_state() [2/2]

initial_state ( self, state )

inherited

Definition at line 216 of file state_machines.py.

    def initial_state(self, state):
        """
        """
        if state in self.states.keys():
            self._initial_state = self.states[state]
            
            self._state = self.states[state]
        else:
            raise KeyError(f"Attempted to set state to '{state}' which is not in the 'states' attribute!")
 

is_valid()

is_valid

(

self

)

Returns:
    bool: Whether or not this machine has been set up correctly with all its necessary attributes.

Definition at line 1008 of file state_machines.py.

    def is_valid(self):
        """
        Returns:
            bool: Whether or not this machine has been set up correctly with all its necessary attributes.
        """
        B2INFO(f"Checking validity of current setup of AlgorithmMachine for {self.algorithm.name}.")
        # Check if we're somehow missing a required attribute (should be impossible since they get initialised in init)
        for attribute_name in self.required_attrs:
            if not hasattr(self, attribute_name):
                B2ERROR(f"AlgorithmMachine attribute {attribute_name} doesn't exist.")
                return False
        # Check if any attributes that need actual values haven't been set or were empty
        for attribute_name in self.required_true_attrs:
            if not getattr(self, attribute_name):
                B2ERROR(f"AlgorithmMachine attribute {attribute_name} returned False.")
                return False
        return True
 

save_graph()

save_graph ( self, filename, graphname )

inherited

Does a simple dot file creation to visualise states and transiitons.

Definition at line 372 of file state_machines.py.

    def save_graph(self, filename, graphname):
        """
        Does a simple dot file creation to visualise states and transiitons.
        """
        with open(filename, "w") as dotfile:
            dotfile.write("digraph " + graphname + " {\n")
            for state in self.states.keys():
                dotfile.write('"' + state + '" [shape=ellipse, color=black]\n')
            for trigger, transition_dicts in self.transitions.items():
                for transition in transition_dicts:
                    dotfile.write('"' + transition["source"].name + '" -> "' +
                                  transition["dest"].name + '" [label="' + trigger + '"]\n')
            dotfile.write("}\n")
 
 

setup_from_dict()

setup_from_dict	(		self,
			params )

Parameters:
    params (dict): Dictionary containing values to be assigned to the machine's attributes of the same name.

Definition at line 1000 of file state_machines.py.

    def setup_from_dict(self, params):
        """
        Parameters:
            params (dict): Dictionary containing values to be assigned to the machine's attributes of the same name.
        """
        for attribute_name, value in params.items():
            setattr(self, attribute_name, value)
 

state() [1/2]

state ( self )

inherited

        The current state of the machine. Actually a `property` decorator. It will call the exit method of the
        current state and enter method of the new one. To get around the behaviour e.g. for setting initial states,
        either use the `initial_state` property or directly set the _state attribute itself (at your own risk!).

Definition at line 227 of file state_machines.py.

    def state(self):
        """
                The current state of the machine. Actually a `property` decorator. It will call the exit method of the
                current state and enter method of the new one. To get around the behaviour e.g. for setting initial states,
                either use the `initial_state` property or directly set the _state attribute itself (at your own risk!).
        """
        return self._state
 

state() [2/2]

state ( self, state )

inherited

Definition at line 236 of file state_machines.py.

    def state(self, state):
        """
        """
        if isinstance(state, str):
            state_name = state
        else:
            state_name = state.name
 
        try:
            state = self.states[state_name]
            # Run exit callbacks of current state
            for callback in self.state.on_exit:
                callback(prior_state=self.state, new_state=state)
            # Run enter callbacks of new state
            for callback in state.on_enter:
                callback(prior_state=self.state, new_state=state)
            # Set the state
            self._state = state
        except KeyError:
            raise MachineError(f"Attempted to set state to '{state}' which not in the 'states' attribute!")
 

Member Data Documentation

_initial_state

dict _initial_state = State(initial_state)

protectedinherited

Actual attribute holding initial state for this machine.

Definition at line 186 of file state_machines.py.

_state

dict _state = self.initial_state

protectedinherited

Actual attribute holding the Current state.

Definition at line 189 of file state_machines.py.

algorithm

algorithm = algorithm

Algorithm() object whose state we are modelling.

Definition at line 971 of file state_machines.py.

database_chain

list database_chain = []

Assigned database chain to the overall Calibration object, or to the 'default' Collection.

Database chains for manually created Collections have no effect here.

Definition at line 978 of file state_machines.py.

default_states

default_states

Initial value:

=  [State("init"),
                               State("ready"),
                               State("running_algorithm"),
                               State("completed"),
                               State("failed")]

Default states for the AlgorithmMachine.

Definition at line 962 of file state_machines.py.

dependent_databases

list dependent_databases = []

CAF created local databases from previous calibrations that this calibration/algorithm depends on.

Definition at line 975 of file state_machines.py.

initial_state

initial_state = initial_state

inherited

Pointless docstring since it's a property.

Definition at line 182 of file state_machines.py.

input_files

input_files = []

Collector output files, will contain all files returned by the output patterns.

Definition at line 973 of file state_machines.py.

output_database_dir

output_database_dir = ""

The output database directory for the localdb that the algorithm will commit to.

Definition at line 982 of file state_machines.py.

output_dir

output_dir = ""

The algorithm output directory which is mostly used to store the stdout file.

Definition at line 980 of file state_machines.py.

required_attrs

list required_attrs

static

Initial value:

=  ["algorithm",
                      "dependent_databases",
                      "database_chain",
                      "output_dir",
                      "output_database_dir",
                      "input_files"
                      ]

Required attributes that must exist before the machine can run properly.

Some are allowed to be values that return False when tested e.g. "" or []

Definition at line 942 of file state_machines.py.

required_true_attrs

list required_true_attrs

static

Initial value:

=  ["algorithm",
                           "output_dir",
                           "output_database_dir",
                           "input_files"
                           ]

Attributes that must have a value that returns True when tested.

Definition at line 951 of file state_machines.py.

result

result = None

IoV_Result object for a single execution, will be reset upon a new execution.

Definition at line 984 of file state_machines.py.

state

state = dest

inherited

Current State of machine.

Definition at line 336 of file state_machines.py.

states

dict states = {}

inherited

Valid states for this machine.

Definition at line 176 of file state_machines.py.

transitions

transitions = defaultdict(list)

inherited

Allowed transitions between states.

Definition at line 191 of file state_machines.py.

---

The documentation for this class was generated from the following file:

• calibration/scripts/caf/state_machines.py