CAF Class Reference

Public Member Functions
	__init__ (self, calibration_defaults=None)
	add_calibration (self, calibration)
	run (self, iov=None)
	backend (self)
	backend (self, backend)

Public Attributes
dict	calibrations = {}
	Dictionary of calibrations for this CAF instance.
dict	future_dependencies = {}
	Dictionary of future dependencies of Calibration objects, where the value is all calibrations that will depend on the key, filled during self.run()
dict	dependencies = {}
	Dictionary of dependencies of Calibration objects, where value is the list of Calibration objects that the key depends on.
str	output_dir = "calibration_results"
	Output path to store results of calibration and bookkeeping information.
	order = None
	The ordering and explicit future dependencies of calibrations.
int	heartbeat = 5
	The heartbeat (seconds) between polling for Calibrations that are finished.
dict	calibration_defaults = {**self.default_calibration_config, **calibration_defaults}
	Default options applied to each calibration known to the CAF, if the Calibration has these defined by the user then the defaults aren't applied.
	backend = caf.backends.Local()
	backend property

Static Public Attributes
dict	default_calibration_config
	The defaults for Calibrations.

Protected Member Functions
	_remove_missing_dependencies (self)
	_order_calibrations (self)
	_check_backend (self)
	_prune_invalid_collections (self)
	_make_output_dir (self)
	_make_database (self)

Protected Attributes
	_backend = None
	Private backend attribute.
	_db_path = None
	The path of the SQLite DB.

Static Protected Attributes
str	_db_name = "caf_state.db"
	The name of the SQLite DB that gets created.

Detailed Description

Parameters:
  calibration_defaults (dict): A dictionary of default options for calibrations run by this `CAF` instance e.g.

                               >>> calibration_defaults={"max_iterations":2}

This class holds `Calibration` objects and processes them. It defines the initial configuration/setup
for the calibrations. But most of the real processing is done through the `caf.state_machines.CalibrationMachine`.

The `CAF` class essentially does some initial setup, holds the `CalibrationBase` instances and calls the
`CalibrationBase.start` when the dependencies are met.

Much of the checking for consistency is done in this class so that no processing is done with an invalid
setup. Choosing which files to use as input should be done from outside during the setup of the `CAF` and
`CalibrationBase` instances.

Definition at line 1183 of file framework.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
			calibration_defaults = None )

Definition at line 1209 of file framework.py.

    def __init__(self, calibration_defaults=None):
        """
        """
        
        self.calibrations = {}
        
        self.future_dependencies = {}
        
        self.dependencies = {}
        
        self.output_dir = "calibration_results"
        
        self.order = None
        
        self._backend = None
        
        self.heartbeat = 5
 
        if not calibration_defaults:
            calibration_defaults = {}
        
        self.calibration_defaults = {**self.default_calibration_config, **calibration_defaults}
        
        self._db_path = None
 

Member Function Documentation

_check_backend()

_check_backend ( self )

protected

Makes sure that the CAF has a valid backend setup. If one isn't set by the user (or if the
one that is stored isn't a valid Backend object) we should create a default Local backend.

Definition at line 1324 of file framework.py.

    def _check_backend(self):
        """
        Makes sure that the CAF has a valid backend setup. If one isn't set by the user (or if the
        one that is stored isn't a valid Backend object) we should create a default Local backend.
        """
        if not isinstance(self._backend, caf.backends.Backend):
            
            self.backend = caf.backends.Local()
 

_make_database()

_make_database ( self )

protected

Creates the CAF status database. If it already exists we don't overwrite it.

Definition at line 1489 of file framework.py.

    def _make_database(self):
        """
        Creates the CAF status database. If it already exists we don't overwrite it.
        """
        self._db_path = Path(self.output_dir, self._db_name).absolute()
        if self._db_path.exists():
            B2INFO(f"Previous CAF database found {self._db_path}")
        # Will create a new database + tables, or do nothing but checks we can connect to existing one
        with CAFDB(self._db_path):
            pass

_make_output_dir()

_make_output_dir ( self )

protected

Creates the output directory. If it already exists we are now going to try and restart the program from the last state.

Returns:
    str: The absolute path of the new output_dir

Definition at line 1470 of file framework.py.

    def _make_output_dir(self):
        """
        Creates the output directory. If it already exists we are now going to try and restart the program from the last state.
 
        Returns:
            str: The absolute path of the new output_dir
        """
        p = Path(self.output_dir).resolve()
        if p.is_dir():
            B2INFO(f"{p.as_posix()} output directory already exists. "
                   "We will try to restart from the previous finishing state.")
            return p.as_posix()
        else:
            p.mkdir(parents=True)
            if p.is_dir():
                return p.as_posix()
            else:
                raise FileNotFoundError(f"Attempted to create output_dir {p.as_posix()}, but it didn't work.")
 

_order_calibrations()

_order_calibrations ( self )

protected

- Uses dependency attributes of calibrations to create a dependency dictionary and passes it
to a sorting algorithm.
- Returns valid OrderedDict if sort was successful, empty one if it failed (most likely a cyclic dependency)

Definition at line 1280 of file framework.py.

    def _order_calibrations(self):
        """
        - Uses dependency attributes of calibrations to create a dependency dictionary and passes it
        to a sorting algorithm.
        - Returns valid OrderedDict if sort was successful, empty one if it failed (most likely a cyclic dependency)
        """
        # First remove any dependencies on calibrations not added to the CAF
        self._remove_missing_dependencies()
        # Filling dependencies dictionaries of CAF for sorting, only explicit dependencies for now
        # Note that they currently use the names not the calibration objects.
        for calibration in self.calibrations.values():
            future_dependencies_names = [dependency.name for dependency in calibration.future_dependencies]
            past_dependencies_names = [dependency.name for dependency in calibration.dependencies]
 
            self.future_dependencies[calibration.name] = future_dependencies_names
            self.dependencies[calibration.name] = past_dependencies_names
        # Gives us a list of A (not THE) valid ordering and checks for cyclic dependencies
        order = topological_sort(self.future_dependencies)
        if not order:
            return False
 
        # Get an ordered dictionary of the sort order but including all implicit dependencies.
        ordered_full_dependencies = all_dependencies(self.future_dependencies, order)
 
        # Return all the implicit+explicit past dependencies
        full_past_dependencies = past_from_future_dependencies(ordered_full_dependencies)
        # Correct each calibration's dependency list to reflect the implicit dependencies
        for calibration in self.calibrations.values():
            full_deps = full_past_dependencies[calibration.name]
            explicit_deps = [cal.name for cal in calibration.dependencies]
            for dep in full_deps:
                if dep not in explicit_deps:
                    calibration.dependencies.append(self.calibrations[dep])
            # At this point the calibrations have their full dependencies but they aren't in topological
            # sort order. Correct that here
            ordered_dependency_list = []
            for ordered_calibration_name in order:
                if ordered_calibration_name in [dep.name for dep in calibration.dependencies]:
                    ordered_dependency_list.append(self.calibrations[ordered_calibration_name])
            calibration.dependencies = ordered_dependency_list
        order = ordered_full_dependencies
        # We should also patch in all of the implicit dependencies for the calibrations
        return order
 

_prune_invalid_collections()

_prune_invalid_collections ( self )

protected

Checks all current calibrations and removes any invalid Collections from their collections list.

Definition at line 1333 of file framework.py.

    def _prune_invalid_collections(self):
        """
        Checks all current calibrations and removes any invalid Collections from their collections list.
        """
        B2INFO("Checking for any invalid Collections in Calibrations.")
        for calibration in self.calibrations.values():
            valid_collections = {}
            for name, collection in calibration.collections.items():
                if collection.is_valid():
                    valid_collections[name] = collection
                else:
                    B2WARNING(f"Removing invalid Collection '{name}' from Calibration '{calibration.name}'.")
            calibration.collections = valid_collections
 

_remove_missing_dependencies()

_remove_missing_dependencies ( self )

protected

This checks the future and past dependencies of each `Calibration` in the `CAF`.
If any dependencies are not known to the `CAF` then they are removed from the `Calibration`
object directly.

Definition at line 1253 of file framework.py.

    def _remove_missing_dependencies(self):
        """
        This checks the future and past dependencies of each `Calibration` in the `CAF`.
        If any dependencies are not known to the `CAF` then they are removed from the `Calibration`
        object directly.
        """
        calibration_names = [calibration.name for calibration in self.calibrations.values()]
 
        def is_dependency_in_caf(dependency):
            """
            Quick function to use with filter() and check dependencies against calibrations known to `CAF`
            """
            dependency_in_caf = dependency.name in calibration_names
            if not dependency_in_caf:
                B2WARNING(f"The calibration {dependency.name} is a required dependency but is not in the CAF."
                          " It has been removed as a dependency.")
            return dependency_in_caf
 
        # Check that there aren't dependencies on calibrations not added to the framework
        # Remove them from the calibration objects if there are.
        for calibration in self.calibrations.values():
            filtered_future_dependencies = list(filter(is_dependency_in_caf, calibration.future_dependencies))
            calibration.future_dependencies = filtered_future_dependencies
 
            filtered_dependencies = list(filter(is_dependency_in_caf, calibration.dependencies))
            calibration.dependencies = filtered_dependencies
 

add_calibration()

add_calibration	(		self,
			calibration )

Adds a `Calibration` that is to be used in this program to the list.
Also adds an empty dependency list to the overall dictionary.
You should not directly alter a `Calibration` object after it has been
added here.

Definition at line 1237 of file framework.py.

    def add_calibration(self, calibration):
        """
        Adds a `Calibration` that is to be used in this program to the list.
        Also adds an empty dependency list to the overall dictionary.
        You should not directly alter a `Calibration` object after it has been
        added here.
        """
        if calibration.is_valid():
            if calibration.name not in self.calibrations:
                self.calibrations[calibration.name] = calibration
            else:
                B2WARNING(f"Tried to add a calibration with the name {calibration.name} twice.")
        else:
            B2WARNING(f"Tried to add incomplete/invalid calibration ({calibration.name}) to the framework."
                      "It was not added and will not be part of the final process.")
 

backend() [1/2]

backend

(

self

)

The `backend <backends.Backend>` that runs the collector job.
When set, this is checked that a `backends.Backend` class instance was passed in.

Definition at line 1454 of file framework.py.

    def backend(self):
        """
        The `backend <backends.Backend>` that runs the collector job.
        When set, this is checked that a `backends.Backend` class instance was passed in.
        """
        return self._backend
 

backend() [2/2]

backend	(		self,
			backend )

Definition at line 1462 of file framework.py.

    def backend(self, backend):
        """
        """
        if isinstance(backend, caf.backends.Backend):
            self._backend = backend
        else:
            B2ERROR('Backend property must inherit from Backend class.')
 

run()

run	(		self,
			iov = None )

Keyword Arguments:
    iov(`caf.utils.IoV`): IoV to calibrate for this processing run. Only the input files necessary to calibrate
                          this IoV will be used in the collection step.

This function runs the overall calibration job, saves the outputs to the output_dir directory,
and creates database payloads.

Upload of final databases is not done here. This simply creates the local databases in
the output directory. You should check the validity of your new local database before uploading
to the conditions DB via the basf2 tools/interface to the DB.

Definition at line 1347 of file framework.py.

    def run(self, iov=None):
        """
        Keyword Arguments:
            iov(`caf.utils.IoV`): IoV to calibrate for this processing run. Only the input files necessary to calibrate
                                  this IoV will be used in the collection step.
 
        This function runs the overall calibration job, saves the outputs to the output_dir directory,
        and creates database payloads.
 
        Upload of final databases is not done here. This simply creates the local databases in
        the output directory. You should check the validity of your new local database before uploading
        to the conditions DB via the basf2 tools/interface to the DB.
        """
        if not self.calibrations:
            B2FATAL("There were no Calibration objects to run. Maybe you tried to add invalid ones?")
        # Checks whether the dependencies we've added will give a valid order
        order = self._order_calibrations()
        if not order:
            B2FATAL("Couldn't order the calibrations properly. Could be a cyclic dependency.")
 
        # Check that a backend has been set and use default Local() one if not
        self._check_backend()
 
        self._prune_invalid_collections()
 
        # Creates the overall output directory and reset the attribute to use an absolute path to it.
        self.output_dir = self._make_output_dir()
 
        #  Creates a SQLite DB to save the status of the various calibrations
        self._make_database()
 
        # Enter the overall output dir during processing and opena  connection to the DB
        with temporary_workdir(self.output_dir):
            db = CAFDB(self._db_path)
            db.open()
            db_initial_calibrations = db.query("select * from calibrations").fetchall()
            for calibration in self.calibrations.values():
                # Apply defaults given to the `CAF` to the calibrations if they aren't set
                calibration._apply_calibration_defaults(self.calibration_defaults)
                calibration._db_path = self._db_path
                calibration.output_database_dir = Path(self.output_dir, calibration.name, "outputdb").as_posix()
                calibration.iov = iov
                if not calibration.backend:
                    calibration.backend = self.backend
                # Do some checking of the db to see if we need to add an entry for this calibration
                if calibration.name not in [db_cal[0] for db_cal in db_initial_calibrations]:
                    db.insert_calibration(calibration.name)
                    db.commit()
                else:
                    for cal_info in db_initial_calibrations:
                        if cal_info[0] == calibration.name:
                            cal_initial_state = cal_info[2]
                            cal_initial_iteration = cal_info[3]
                    B2INFO(f"Previous entry in database found for {calibration.name}.")
                    B2INFO(f"Setting {calibration.name} state to checkpoint state '{cal_initial_state}'.")
                    calibration.state = cal_initial_state
                    B2INFO(f"Setting {calibration.name} iteration to '{cal_initial_iteration}'.")
                    calibration.iteration = cal_initial_iteration
                # Daemonize so that it exits if the main program exits
                calibration.daemon = True
 
            db.close()
 
            # Is it possible to keep going?
            keep_running = True
            while keep_running:
                keep_running = False
                # Do we have calibrations that may yet complete?
                remaining_calibrations = []
 
                for calibration in self.calibrations.values():
                    # Find the currently ended calibrations (may not be joined yet)
                    if (calibration.state == CalibrationBase.end_state or calibration.state == CalibrationBase.fail_state):
                        # Search for any alive Calibrations and join them
                        if calibration.is_alive():
                            B2DEBUG(29, f"Joining {calibration.name}.")
                            calibration.join()
                    else:
                        if calibration.dependencies_met():
                            if not calibration.is_alive():
                                B2DEBUG(29, f"Starting {calibration.name}.")
                                try:
                                    calibration.start()
                                except RuntimeError:
                                    # Catch the case when the calibration just finished so it ended up here
                                    # in the "else" and not above where it should have been joined.
                                    B2DEBUG(29, f"{calibration.name} probably just finished, join it later.")
                            remaining_calibrations.append(calibration)
                        else:
                            if not calibration.failed_dependencies():
                                remaining_calibrations.append(calibration)
                if remaining_calibrations:
                    keep_running = True
                    # Loop over jobs that the calibrations want submitted and submit them.
                    # We do this here because some backends don't like us submitting in parallel from multiple CalibrationThreads
                    # So this is like a mini job queue without getting too clever with it
                    for calibration in remaining_calibrations:
                        for job in calibration.jobs_to_submit[:]:
                            calibration.backend.submit(job)
                            calibration.jobs_to_submit.remove(job)
                sleep(self.heartbeat)
 
            B2INFO("Printing summary of final CAF status.")
            with CAFDB(self._db_path, read_only=True) as db:
                print(db.output_calibration_table())
 

Member Data Documentation

_backend

_backend = None

protected

Private backend attribute.

Definition at line 1225 of file framework.py.

_db_name

str _db_name = "caf_state.db"

staticprotected

The name of the SQLite DB that gets created.

Definition at line 1202 of file framework.py.

_db_path

_db_path = None

protected

The path of the SQLite DB.

Definition at line 1235 of file framework.py.

backend

backend = caf.backends.Local()

backend property

Definition at line 1331 of file framework.py.

calibration_defaults

calibration_defaults = {**self.default_calibration_config, **calibration_defaults}

Default options applied to each calibration known to the CAF, if the Calibration has these defined by the user then the defaults aren't applied.

A simple way to define the same configuration to all calibrations in the CAF.

Definition at line 1233 of file framework.py.

calibrations

dict calibrations = {}

Dictionary of calibrations for this CAF instance.

You should use add_calibration to add to this.

Definition at line 1213 of file framework.py.

default_calibration_config

dict default_calibration_config

static

Initial value:

=  {
                                  "max_iterations": 5,
                                  "ignored_runs": []
                                 }

The defaults for Calibrations.

Definition at line 1204 of file framework.py.

dependencies

dict dependencies = {}

Dictionary of dependencies of Calibration objects, where value is the list of Calibration objects that the key depends on.

This attribute is filled during self.run()

Definition at line 1219 of file framework.py.

future_dependencies

dict future_dependencies = {}

Dictionary of future dependencies of Calibration objects, where the value is all calibrations that will depend on the key, filled during self.run()

Definition at line 1216 of file framework.py.

heartbeat

heartbeat = 5

The heartbeat (seconds) between polling for Calibrations that are finished.

Definition at line 1227 of file framework.py.

order

order = None

The ordering and explicit future dependencies of calibrations.

Will be filled during CAF.run() for you.

Definition at line 1223 of file framework.py.

output_dir

output_dir = "calibration_results"

Output path to store results of calibration and bookkeeping information.

Definition at line 1221 of file framework.py.

---

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

• calibration/scripts/caf/framework.py