CalibrationMachine Class Reference

Inheritance diagram for CalibrationMachine:

Public Member Functions
	__init__ (self, calibration, iov_to_calibrate=None, initial_state="init", iteration=0)
	files_containing_iov (self, file_paths, files_to_iovs, iov)
	dependencies_completed (self)
	automatic_transition (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
	States that are defaults to the CalibrationMachine (could override later)
	calibration = calibration
	Calibration object whose state we are modelling.
	iteration = iteration
	Which iteration step are we in.
	collector_backend = None
	Backend used for this calibration machine collector.
	iov_to_calibrate = iov_to_calibrate
	IoV to be executed, currently will loop over all runs in IoV.
	root_dir = Path(os.getcwd(), calibration.name)
	root directory for this Calibration
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
str	collector_input_dir = 'collector_input'
	input directory of collector
str	collector_output_dir = 'collector_output'
	output directory of collector
str	algorithm_output_dir = 'algorithm_output'
	output directory of algorithm

Protected Member Functions
	_update_cal_state (self, **kwargs)
	_dump_job_config (self)
	_recover_collector_jobs (self)
	_iov_requested (self)
	_resolve_file_paths (self)
	_build_iov_dicts (self)
	_below_max_iterations (self)
	_increment_iteration (self)
	_collection_completed (self)
	_collection_failed (self)
	_runner_not_failed (self)
	_runner_failed (self)
	_collector_jobs_ready (self)
	_submit_collections (self)
	_no_require_iteration (self)
	_require_iteration (self)
	_log_new_state (self, **kwargs)
	_make_output_dir (self)
	_make_collector_path (self, name, collection)
	_make_pre_collector_path (self, name, collection)
	_create_collector_jobs (self)
	_check_valid_collector_output (self)
	_run_algorithms (self)
	_prepare_final_db (self)
	_trigger (self, transition_name, transition_dict, **kwargs)

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

Protected Attributes
dict	_algorithm_results = {}
	Results of each iteration for all algorithms of this calibration.
	_runner_final_state = None
	Final state of the algorithm runner for the current iteration.
dict	_collector_timing = {}
	Times of various useful updates to the collector job e.g.
dict	_collector_jobs = {}
	The collector jobs used for submission.
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 `Calibration` objects and the flow of
processing for them.

Definition at line 383 of file state_machines.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
			calibration,
			iov_to_calibrate = None,
			initial_state = "init",
			iteration = 0 )

Takes a Calibration object from the caf framework and lets you
set the initial state.

Definition at line 396 of file state_machines.py.

    def __init__(self, calibration, iov_to_calibrate=None, initial_state="init", iteration=0):
        """
        Takes a Calibration object from the caf framework and lets you
        set the initial state.
        """
        
        self.default_states = [State("init", enter=[self._update_cal_state,
                                                    self._log_new_state]),
                               State("running_collector", enter=[self._update_cal_state,
                                                                 self._log_new_state]),
                               State("collector_failed", enter=[self._update_cal_state,
                                                                self._log_new_state]),
                               State("collector_completed", enter=[self._update_cal_state,
                                                                   self._log_new_state]),
                               State("running_algorithms", enter=[self._update_cal_state,
                                                                  self._log_new_state]),
                               State("algorithms_failed", enter=[self._update_cal_state,
                                                                 self._log_new_state]),
                               State("algorithms_completed", enter=[self._update_cal_state,
                                                                    self._log_new_state]),
                               State("completed", enter=[self._update_cal_state,
                                                         self._log_new_state]),
                               State("failed", enter=[self._update_cal_state,
                                                      self._log_new_state])
                               ]
 
        super().__init__(self.default_states, initial_state)
 
        
        self.calibration = calibration
        # Monkey Patching for the win!
        
        self.calibration.machine = self
        
        self.iteration = iteration
        
        self.collector_backend = None
        
        self._algorithm_results = {}
        
        self._runner_final_state = None
        
        self.iov_to_calibrate = iov_to_calibrate
        
        self.root_dir = Path(os.getcwd(), calibration.name)
 
        
        self._collector_timing = {}
 
        
        self._collector_jobs = {}
 
        self.add_transition("submit_collector", "init", "running_collector",
                            conditions=self.dependencies_completed,
                            before=[self._make_output_dir,
                                    self._resolve_file_paths,
                                    self._build_iov_dicts,
                                    self._create_collector_jobs,
                                    self._submit_collections,
                                    self._dump_job_config])
        self.add_transition("fail", "running_collector", "collector_failed",
                            conditions=self._collection_failed)
        self.add_transition("complete", "running_collector", "collector_completed",
                            conditions=self._collection_completed)
        self.add_transition("run_algorithms", "collector_completed", "running_algorithms",
                            before=self._check_valid_collector_output,
                            after=[self._run_algorithms,
                                   self.automatic_transition])
        self.add_transition("complete", "running_algorithms", "algorithms_completed",
                            after=self.automatic_transition,
                            conditions=self._runner_not_failed)
        self.add_transition("fail", "running_algorithms", "algorithms_failed",
                            conditions=self._runner_failed)
        self.add_transition("iterate", "algorithms_completed", "init",
                            conditions=[self._require_iteration,
                                        self._below_max_iterations],
                            after=self._increment_iteration)
        self.add_transition("finish", "algorithms_completed", "completed",
                            conditions=self._no_require_iteration,
                            before=self._prepare_final_db)
        self.add_transition("fail_fully", "algorithms_failed", "failed")
        self.add_transition("fail_fully", "collector_failed", "failed")
 

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 302 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
 

_below_max_iterations()

_below_max_iterations ( self )

protected

Definition at line 564 of file state_machines.py.

    def _below_max_iterations(self):
        """
        """
        return self.iteration < self.calibration.max_iterations
 

_build_iov_dicts()

_build_iov_dicts ( self )

protected

Build IoV file dictionary for each collection if required.

Definition at line 541 of file state_machines.py.

    def _build_iov_dicts(self):
        """
        Build IoV file dictionary for each collection if required.
        """
        iov_requested = self._iov_requested()
        if iov_requested or self.calibration.ignored_runs:
            for coll_name, collection in self.calibration.collections.items():
                if not collection.files_to_iovs:
                    B2INFO("Creating IoV dictionaries to map files to (Exp,Run) ranges for"
                           f" Calibration '{self.calibration.name} and Collection '{coll_name}'."
                           " Filling dictionary from input file metadata."
                           " If this is slow, set the 'files_to_iovs' attribute of each Collection before running.")
 
                    files_to_iovs = {}
                    for file_path in collection.input_files:
                        files_to_iovs[file_path] = get_iov_from_file(file_path)
                    collection.files_to_iovs = files_to_iovs
                else:
                    B2INFO("Using File to IoV mapping from 'files_to_iovs' attribute for "
                           f"Calibration '{self.calibration.name}' and Collection '{coll_name}'.")
        else:
            B2INFO("No File to IoV mapping required.")
 

_callback()

_callback ( func, ** kwargs )

staticprotectedinherited

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

Definition at line 340 of file state_machines.py.

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

_check_valid_collector_output()

_check_valid_collector_output ( self )

protected

check that collector output is valid

Definition at line 828 of file state_machines.py.

    def _check_valid_collector_output(self):
        """check that collector output is valid"""
        B2INFO("Checking that Collector output exists for all collector jobs "
               f"using {self.calibration.name}.output_patterns.")
        if not self._collector_jobs:
            B2INFO("We're restarting so we'll recreate the collector Job object.")
            self._recover_collector_jobs()
 
        for job in self._collector_jobs.values():
            if not job.subjobs:
                output_files = []
                for pattern in job.output_patterns:
                    output_files.extend(glob.glob(os.path.join(job.output_dir, pattern)))
                if not output_files:
                    raise MachineError("No output files from Collector Job")
            else:
                for subjob in job.subjobs.values():
                    output_files = []
                    for pattern in subjob.output_patterns:
                        output_files.extend(glob.glob(os.path.join(subjob.output_dir, pattern)))
                    if not output_files:
                        raise MachineError(f"No output files from Collector {subjob}")
 

_collection_completed()

_collection_completed ( self )

protected

Did all the collections succeed?

Definition at line 575 of file state_machines.py.

    def _collection_completed(self):
        """
        Did all the collections succeed?
        """
        B2DEBUG(29, "Checking for failed collector job.")
        if self._collector_jobs_ready():
            return all([job.status == "completed" for job in self._collector_jobs.values()])
 

_collection_failed()

_collection_failed ( self )

protected

Did any of the collections fail?

Definition at line 583 of file state_machines.py.

    def _collection_failed(self):
        """
        Did any of the collections fail?
        """
        B2DEBUG(29, "Checking for failed collector job.")
        if self._collector_jobs_ready():
            return any([job.status == "failed" for job in self._collector_jobs.values()])
 

_collector_jobs_ready()

_collector_jobs_ready ( self )

protected

Definition at line 608 of file state_machines.py.

    def _collector_jobs_ready(self):
        """
        """
        since_last_update = time.time() - self._collector_timing["last_update"]
        if since_last_update > self.calibration.collector_full_update_interval:
            B2INFO("Updating full collector job statuses.")
            for job in self._collector_jobs.values():
                job.update_status()
                self._collector_timing["last_update"] = time.time()
                if job.subjobs:
                    num_completed = sum((subjob.status in subjob.exit_statuses) for subjob in job.subjobs.values())
                    total_subjobs = len(job.subjobs)
                    B2INFO(f"{num_completed}/{total_subjobs} Collector SubJobs finished in"
                           f" Calibration {self.calibration.name} Job {job.name}.")
        return all([job.ready() for job in self._collector_jobs.values()])
 

_create_collector_jobs()

_create_collector_jobs ( self )

protected

Creates a Job object for the collections of this iteration, ready for submission
to backend.

Definition at line 730 of file state_machines.py.

    def _create_collector_jobs(self):
        """
        Creates a Job object for the collections of this iteration, ready for submission
        to backend.
        """
        for collection_name, collection in self.calibration.collections.items():
            iteration_dir = self.root_dir.joinpath(str(self.iteration))
            job = Job('_'.join([self.calibration.name, collection_name, 'Iteration', str(self.iteration)]))
            job.output_dir = iteration_dir.joinpath(self.collector_output_dir, collection_name)
            job.working_dir = iteration_dir.joinpath(self.collector_output_dir, collection_name)
            # Remove previous failed attempt to avoid problems
            if job.output_dir.exists():
                B2INFO(f"Previous output directory for {self.calibration.name} collector {collection_name} exists."
                       f"Deleting {job.output_dir} before re-submitting.")
                shutil.rmtree(job.output_dir)
            job.cmd = collection.job_cmd
            job.append_current_basf2_setup_cmds()
            job.input_sandbox_files.append(collection.job_script)
            collector_path_file = Path(self._make_collector_path(collection_name, collection))
            job.input_sandbox_files.append(collector_path_file)
            if collection.pre_collector_path:
                pre_collector_path_file = Path(self._make_pre_collector_path(collection_name, collection))
                job.input_sandbox_files.append(pre_collector_path_file)
 
            # Want to figure out which local databases are required for this job and their paths
            list_dependent_databases = []
 
            # 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
            for dependency in self.calibration.dependencies:
                database_dir = os.path.join(os.getcwd(), dependency.name, 'outputdb')
                B2INFO(f"Adding local database from {dependency.name} for use by {self.calibration.name}.")
                list_dependent_databases.append((os.path.join(database_dir, 'database.txt'), database_dir))
 
            # Add previous iteration databases from this calibration
            if self.iteration > 0:
                previous_iteration_dir = self.root_dir.joinpath(str(self.iteration - 1))
                database_dir = os.path.join(previous_iteration_dir, self.calibration.alg_output_dir, 'outputdb')
                list_dependent_databases.append((os.path.join(database_dir, 'database.txt'), database_dir))
                B2INFO(f"Adding local database from previous iteration of {self.calibration.name}.")
 
            # Let's use a directory to store some files later for input to the collector jobs. Should already exist from
            # collector path
            input_data_directory = self.root_dir.joinpath(str(self.iteration), self.collector_input_dir, collection_name)
 
            # Need to pass setup info to collector which would be tricky as arguments
            # We make a dictionary and pass it in as json
            job_config = {}
            # Apply the user-set Calibration database chain to the base of the overall chain.
            json_db_chain = []
            for database in collection.database_chain:
                if database.db_type == 'local':
                    json_db_chain.append(('local', (database.filepath.as_posix(), database.payload_dir.as_posix())))
                elif database.db_type == 'central':
                    json_db_chain.append(('central', database.global_tag))
                else:
                    raise ValueError(f"Unknown database type {database.db_type}.")
            # CAF created ones for dependent calibrations and previous iterations of this calibration
            for database in list_dependent_databases:
                json_db_chain.append(('local', database))
            job_config['database_chain'] = json_db_chain
 
            job_config_file_path = input_data_directory.joinpath('collector_config.json').absolute()
            with open(job_config_file_path, 'w') as job_config_file:
                json.dump(job_config, job_config_file, indent=2)
            job.input_sandbox_files.append(job_config_file_path)
 
            # Define the input files
            input_data_files = set(collection.input_files)
            # Reduce the input data files to only those that overlap with the optional requested IoV
            if self.iov_to_calibrate:
                input_data_files = self.files_containing_iov(input_data_files,
                                                             collection.files_to_iovs,
                                                             self.iov_to_calibrate)
            # Remove any files that ONLY contain runs from our optional ignored_runs list
            files_to_ignore = set()
            for exprun in self.calibration.ignored_runs:
                for input_file in input_data_files:
                    file_iov = self.calibration.files_to_iovs[input_file]
                    if file_iov == exprun.make_iov():
                        B2INFO(f"You have asked for {exprun} to be ignored for Calibration '{self.calibration.name}'. "
                               f"Therefore the input file '{input_file}' from Collection '{collection_name}' "
                               "is being removed from input files list.")
                        files_to_ignore.add(input_file)
            input_data_files.difference_update(files_to_ignore)
 
            if not input_data_files:
                raise MachineError(f"No valid input files for Calibration '{self.calibration.name}' "
                                   f" and Collection '{collection_name}'.")
            job.input_files = list(input_data_files)
 
            job.splitter = collection.splitter
            job.backend_args = collection.backend_args
            # Output patterns to be returned from collector job
            job.output_patterns = collection.output_patterns
            B2DEBUG(20, f"Collector job for {self.calibration.name}:{collection_name}:\n{job}")
            self._collector_jobs[collection_name] = job
 

_dump_job_config()

_dump_job_config ( self )

protected

Dumps the `Job` object for the collections to JSON files so that it's configuration can be recovered
later in case of failure.

Definition at line 498 of file state_machines.py.

    def _dump_job_config(self):
        """
        Dumps the `Job` object for the collections to JSON files so that it's configuration can be recovered
        later in case of failure.
        """
        # Wait for jobs (+subjobs) to be submitted so that all information is filled. Since the parent CAF object asynchronously
        # submits the jobs this might need to wait a while.
        while any(map(lambda j: j.status == "init", self._collector_jobs.values())):
            B2DEBUG(29, "Some Collector Jobs still in 'init' state. Waiting...")
            time.sleep(5)
 
        for collection_name, job in self._collector_jobs.items():
            collector_job_output_file_name = self.calibration.collections[collection_name].job_config
            output_file = self.root_dir.joinpath(str(self.iteration), self.collector_input_dir,
                                                 collection_name, collector_job_output_file_name)
            job.dump_to_json(output_file)
 

_increment_iteration()

_increment_iteration ( self )

protected

Definition at line 569 of file state_machines.py.

    def _increment_iteration(self):
        """
        """
        self.iteration += 1
        self.calibration.iteration = self.iteration
 

_iov_requested()

_iov_requested ( self )

protected

Definition at line 526 of file state_machines.py.

    def _iov_requested(self):
        """
        """
        if self.iov_to_calibrate:
            B2DEBUG(20, f"Overall IoV {self.iov_to_calibrate} requested for calibration: {self.calibration.name}.")
            return True
        else:
            B2DEBUG(20, f"No overall IoV requested for calibration: {self.calibration.name}.")
            return False
 

_log_new_state()

_log_new_state ( self, ** kwargs )

protected

Definition at line 655 of file state_machines.py.

    def _log_new_state(self, **kwargs):
        """
        """
        B2INFO(f"Calibration Machine {self.calibration.name} moved to state {kwargs['new_state'].name}.")
 

_make_collector_path()

_make_collector_path ( self, name, collection )

protected

Creates a basf2 path for the correct collector and serializes it in the
self.output_dir/<calibration_name>/<iteration>/paths directory

Definition at line 696 of file state_machines.py.

    def _make_collector_path(self, name, collection):
        """
        Creates a basf2 path for the correct collector and serializes it in the
        self.output_dir/<calibration_name>/<iteration>/paths directory
        """
        path_output_dir = self.root_dir.joinpath(str(self.iteration), self.collector_input_dir, name)
        # Automatically overwrite any previous directory
        create_directories(path_output_dir)
        path_file_name = collection.collector.name() + '.path'
        path_file_name = path_output_dir / path_file_name
        # Create empty path and add collector to it
        coll_path = create_path()
        coll_path.add_module(collection.collector)
        # Dump the basf2 path to file
        with open(path_file_name, 'bw') as serialized_path_file:
            pickle.dump(serialize_path(coll_path), serialized_path_file)
        # Return the pickle file path for addition to the input sandbox
        return str(path_file_name.absolute())
 

_make_output_dir()

_make_output_dir ( self )

protected

Creates the overall root directory of the Calibration. Will not overwrite if it already exists.
Also creates s

Definition at line 689 of file state_machines.py.

    def _make_output_dir(self):
        """
        Creates the overall root directory of the Calibration. Will not overwrite if it already exists.
        Also creates s
        """
        create_directories(self.root_dir, overwrite=False)
 

_make_pre_collector_path()

_make_pre_collector_path ( self, name, collection )

protected

Creates a basf2 path for the collectors setup path (Collection.pre_collector_path) and serializes it in the
self.output_dir/<calibration_name>/<iteration>/<colector_output>/<name> directory.

Definition at line 715 of file state_machines.py.

    def _make_pre_collector_path(self, name, collection):
        """
        Creates a basf2 path for the collectors setup path (Collection.pre_collector_path) and serializes it in the
        self.output_dir/<calibration_name>/<iteration>/<colector_output>/<name> directory.
        """
        path_output_dir = self.root_dir.joinpath(str(self.iteration), self.collector_input_dir, name)
        coll_path = collection.pre_collector_path
        path_file_name = 'pre_collector.path'
        path_file_name = os.path.join(path_output_dir, path_file_name)
        # Dump the basf2 path to file
        with open(path_file_name, 'bw') as serialized_path_file:
            pickle.dump(serialize_path(coll_path), serialized_path_file)
        # Return the pickle file path for addition to the input sandbox
        return path_file_name
 

_no_require_iteration()

_no_require_iteration ( self )

protected

Definition at line 631 of file state_machines.py.

    def _no_require_iteration(self):
        """
        """
        if self._require_iteration() and self._below_max_iterations():
            return False
        elif self._require_iteration() and not self._below_max_iterations():
            B2INFO(f"Reached maximum number of iterations ({self.calibration.max_iterations}), will complete now.")
            return True
        elif not self._require_iteration():
            return True
 

_prepare_final_db()

_prepare_final_db ( self )

protected

Take the last iteration's outputdb and copy it to a more easily findable place.

Definition at line 915 of file state_machines.py.

    def _prepare_final_db(self):
        """
        Take the last iteration's outputdb and copy it to a more easily findable place.
        """
        database_location = self.root_dir.joinpath(str(self.iteration),
                                                   self.calibration.alg_output_dir,
                                                   'outputdb')
        final_database_location = self.root_dir.joinpath('outputdb')
        if final_database_location.exists():
            B2INFO(f"Removing previous final output database for {self.calibration.name} before copying new one.")
            shutil.rmtree(final_database_location)
        shutil.copytree(database_location, final_database_location)
 
 

_recover_collector_jobs()

_recover_collector_jobs ( self )

protected

Recovers the `Job` object for the collector from a JSON file in the event that we are starting from a reset.

Definition at line 515 of file state_machines.py.

    def _recover_collector_jobs(self):
        """
        Recovers the `Job` object for the collector from a JSON file in the event that we are starting from a reset.
        """
        for collection_name, collection in self.calibration.collections.items():
            output_file = self.root_dir.joinpath(str(self.iteration),
                                                 self.collector_input_dir,
                                                 collection_name,
                                                 collection.job_config)
            self._collector_jobs[collection_name] = Job.from_json(output_file)
 

_require_iteration()

_require_iteration ( self )

protected

Definition at line 642 of file state_machines.py.

    def _require_iteration(self):
        """
        """
        iteration_called = False
        for alg_name, results in self._algorithm_results[self.iteration].items():
            for result in results:
                if result.result == CalibrationAlgorithm.c_Iterate:
                    iteration_called = True
                    break
            if iteration_called:
                break
        return iteration_called
 

_resolve_file_paths()

_resolve_file_paths ( self )

protected

Definition at line 536 of file state_machines.py.

    def _resolve_file_paths(self):
        """
        """
        pass
 

_run_algorithms()

_run_algorithms ( self )

protected

Runs the Calibration Algorithms for this calibration machine.

Will run them sequentially locally (possible benefits to using a
processing pool for low memory algorithms later on.)

Definition at line 851 of file state_machines.py.

    def _run_algorithms(self):
        """
        Runs the Calibration Algorithms for this calibration machine.
 
        Will run them sequentially locally (possible benefits to using a
        processing pool for low memory algorithms later on.)
        """
        # Get an instance of the Runner for these algorithms and run it
        algs_runner = self.calibration.algorithms_runner(name=self.calibration.name)
        algs_runner.algorithms = self.calibration.algorithms
        algorithm_output_dir = self.root_dir.joinpath(str(self.iteration), self.calibration.alg_output_dir)
        output_database_dir = algorithm_output_dir.joinpath("outputdb")
        # Remove it, if we failed previously, to start clean
        if algorithm_output_dir.exists():
            B2INFO(f"Output directory for {self.calibration.name} already exists from a previous CAF attempt. "
                   f"Deleting and recreating {algorithm_output_dir}.")
        create_directories(algorithm_output_dir)
        B2INFO(f"Output local database for {self.calibration.name} will be stored at {output_database_dir}.")
        algs_runner.output_database_dir = output_database_dir
        algs_runner.output_dir = self.root_dir.joinpath(str(self.iteration), self.calibration.alg_output_dir)
        input_files = []
 
        for job in self._collector_jobs.values():
            if job.subjobs:
                for subjob in job.subjobs.values():
                    for pattern in subjob.output_patterns:
                        input_files.extend(glob.glob(os.path.join(subjob.output_dir, pattern)))
            else:
                for pattern in job.output_patterns:
                    input_files.extend(glob.glob(os.path.join(job.output_dir, pattern)))
 
        algs_runner.input_files = input_files
 
        # Add any user defined database chain for this calibration
        algs_runner.database_chain = self.calibration.database_chain
 
        # 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
        list_dependent_databases = []
        for dependency in self.calibration.dependencies:
            database_dir = os.path.join(os.getcwd(), dependency.name, 'outputdb')
            B2INFO(f"Adding local database from {dependency.name} for use by {self.calibration.name}.")
            list_dependent_databases.append((os.path.join(database_dir, 'database.txt'), database_dir))
 
        # Add previous iteration databases from this calibration
        if self.iteration > 0:
            previous_iteration_dir = self.root_dir.joinpath(str(self.iteration - 1))
            database_dir = os.path.join(previous_iteration_dir, self.calibration.alg_output_dir, 'outputdb')
            list_dependent_databases.append((os.path.join(database_dir, 'database.txt'), database_dir))
            B2INFO(f"Adding local database from previous iteration of {self.calibration.name}.")
        algs_runner.dependent_databases = list_dependent_databases
 
        algs_runner.ignored_runs = self.calibration.ignored_runs
 
        try:
            algs_runner.run(self.iov_to_calibrate, self.iteration)
        except Exception as err:
            print(err)
            # We directly set the state without triggering the transition because normally we fail based on checking the algorithm
            # results. But here we had an actual exception so we just force into failure instead.
            self._state = State("algorithms_failed")
        self._algorithm_results[self.iteration] = algs_runner.results
        self._runner_final_state = algs_runner.final_state
 

_runner_failed()

_runner_failed ( self )

protected

Returns:
    bool: If AlgorithmsRunner failed return True.

Definition at line 598 of file state_machines.py.

    def _runner_failed(self):
        """
        Returns:
            bool: If AlgorithmsRunner failed return True.
        """
        if self._runner_final_state == AlgorithmsRunner.FAILED:
            return True
        else:
            return False
 

_runner_not_failed()

_runner_not_failed ( self )

protected

Returns:
    bool: If AlgorithmsRunner succeeded return True.

Definition at line 591 of file state_machines.py.

    def _runner_not_failed(self):
        """
        Returns:
            bool: If AlgorithmsRunner succeeded return True.
        """
        return not self._runner_failed()
 

_submit_collections()

_submit_collections ( self )

protected

Definition at line 624 of file state_machines.py.

    def _submit_collections(self):
        """
        """
        self.calibration.jobs_to_submit.extend(list(self._collector_jobs.values()))
        self._collector_timing["start"] = time.time()
        self._collector_timing["last_update"] = time.time()
 

_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 316 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")
 

_update_cal_state()

_update_cal_state ( self, ** kwargs )

protected

update calibration state

Definition at line 481 of file state_machines.py.

    def _update_cal_state(self, **kwargs):
        """update calibration state"""
        self.calibration.state = str(kwargs["new_state"])
 

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 189 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 260 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)
 

automatic_transition()

automatic_transition

(

self

)

Automatically try all transitions out of this state once. Tries fail last.

Definition at line 671 of file state_machines.py.

    def automatic_transition(self):
        """
        Automatically try all transitions out of this state once. Tries fail last.
        """
        possible_transitions = self.get_transitions(self.state)
        for transition in possible_transitions:
            try:
                if transition != "fail":
                    getattr(self, transition)()
                    break
            except ConditionError:
                continue
        else:
            if "fail" in possible_transitions:
                getattr(self, "fail")()
            else:
                raise MachineError(f"Failed to automatically transition out of {self.state} state.")
 

default_condition()

default_condition ( ** kwargs )

staticinherited

Method to always return True.

Definition at line 254 of file state_machines.py.

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

dependencies_completed()

dependencies_completed

(

self

)

Condition function to check that the dependencies of our calibration are in the 'completed' state.
Technically only need to check explicit dependencies.

Definition at line 660 of file state_machines.py.

    def dependencies_completed(self):
        """
        Condition function to check that the dependencies of our calibration are in the 'completed' state.
        Technically only need to check explicit dependencies.
        """
        for calibration in self.calibration.dependencies:
            if not calibration.state == calibration.end_state:
                return False
        else:
            return True
 

files_containing_iov()

files_containing_iov	(		self,
			file_paths,
			files_to_iovs,
			iov )

Lookup function that returns all files from the file_paths that
overlap with this IoV.

Definition at line 485 of file state_machines.py.

    def files_containing_iov(self, file_paths, files_to_iovs, iov):
        """
        Lookup function that returns all files from the file_paths that
        overlap with this IoV.
        """
        # Files that contain an Exp,Run range that overlaps with given IoV
        overlapping_files = set()
 
        for file_path, file_iov in files_to_iovs.items():
            if file_iov.overlaps(iov) and (file_path in file_paths):
                overlapping_files.add(file_path)
        return overlapping_files
 

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 357 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 346 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 205 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 212 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!")
 

save_graph()

save_graph ( self, filename, graphname )

inherited

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

Definition at line 368 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")
 
 

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 223 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 232 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

_algorithm_results

dict _algorithm_results = {}

protected

Results of each iteration for all algorithms of this calibration.

Definition at line 434 of file state_machines.py.

_collector_jobs

dict _collector_jobs = {}

protected

The collector jobs used for submission.

Definition at line 448 of file state_machines.py.

_collector_timing

dict _collector_timing = {}

protected

Times of various useful updates to the collector job e.g.

start, elapsed, last update Used to periodically call update_status on the collector job and find out an overall number of jobs remaining + estimated remaining time

Definition at line 445 of file state_machines.py.

_initial_state

dict _initial_state = State(initial_state)

protectedinherited

Actual attribute holding initial state for this machine.

Definition at line 182 of file state_machines.py.

_runner_final_state

_runner_final_state = None

protected

Final state of the algorithm runner for the current iteration.

Definition at line 436 of file state_machines.py.

_state

dict _state = self.initial_state

protectedinherited

Actual attribute holding the Current state.

Definition at line 185 of file state_machines.py.

algorithm_output_dir

str algorithm_output_dir = 'algorithm_output'

static

output directory of algorithm

Definition at line 394 of file state_machines.py.

calibration

calibration = calibration

Calibration object whose state we are modelling.

Definition at line 425 of file state_machines.py.

collector_backend

collector_backend = None

Backend used for this calibration machine collector.

Definition at line 432 of file state_machines.py.

collector_input_dir

str collector_input_dir = 'collector_input'

static

input directory of collector

Definition at line 390 of file state_machines.py.

collector_output_dir

str collector_output_dir = 'collector_output'

static

output directory of collector

Definition at line 392 of file state_machines.py.

default_states

default_states

Initial value:

=  [State("init", enter=[self._update_cal_state,
                                                    self._log_new_state]),
                               State("running_collector", enter=[self._update_cal_state,
                                                                 self._log_new_state]),
                               State("collector_failed", enter=[self._update_cal_state,
                                                                self._log_new_state]),
                               State("collector_completed", enter=[self._update_cal_state,
                                                                   self._log_new_state]),
                               State("running_algorithms", enter=[self._update_cal_state,
                                                                  self._log_new_state]),
                               State("algorithms_failed", enter=[self._update_cal_state,
                                                                 self._log_new_state]),
                               State("algorithms_completed", enter=[self._update_cal_state,
                                                                    self._log_new_state]),
                               State("completed", enter=[self._update_cal_state,
                                                         self._log_new_state]),
                               State("failed", enter=[self._update_cal_state,
                                                      self._log_new_state])
                               ]

States that are defaults to the CalibrationMachine (could override later)

Definition at line 402 of file state_machines.py.

initial_state

initial_state = initial_state

inherited

Pointless docstring since it's a property.

Definition at line 178 of file state_machines.py.

iov_to_calibrate

iov_to_calibrate = iov_to_calibrate

IoV to be executed, currently will loop over all runs in IoV.

Definition at line 438 of file state_machines.py.

iteration

iteration = iteration

Which iteration step are we in.

Definition at line 430 of file state_machines.py.

root_dir

root_dir = Path(os.getcwd(), calibration.name)

root directory for this Calibration

Definition at line 440 of file state_machines.py.

state

state = dest

inherited

Current State of machine.

Definition at line 332 of file state_machines.py.

states

dict states = {}

inherited

Valid states for this machine.

Definition at line 172 of file state_machines.py.

transitions

transitions = defaultdict(list)

inherited

Allowed transitions between states.

Definition at line 187 of file state_machines.py.

---

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

• calibration/scripts/caf/state_machines.py