SequentialRunByRun Class Reference

Inheritance diagram for SequentialRunByRun:

Public Member Functions
	__init__ (self, algorithm)
	apply_experiment_settings (self, algorithm, experiment)
	run (self, iov, iteration, queue)
	execute_over_run_list (self, iteration, run_list, lowest_exprun, highest_exprun)
	setup_from_dict (self, params)
	is_valid (self)
	find_iov_gaps (self)
	any_failed_iov (self)
	send_result (self, result)
	send_final_state (self, state)

Public Attributes
	machine = AlgorithmMachine(self.algorithm)
	:py:class:caf.state_machines.AlgorithmMachine used to help set up and execute CalibrationAlgorithm It gets setup properly in :py:func:run
bool	first_execution = True
	boolean storing whether this is the first time the algorithm is executed
	algorithm = algorithm
	Algorithm() class that we're running.
list	input_files = []
	Collector output files, will contain all files returned by the output patterns.
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.
list	database_chain = []
	User defined database chain i.e.
list	dependent_databases = []
	CAF created local databases from previous calibrations that this calibration/algorithm depends on.
list	ignored_runs = []
	Runs that will not be included in ANY execution of the algorithm.
list	results = []
	The list of results objects which will be sent out before the end.
	queue = None
	The multiprocessing Queue we use to pass back results one at a time.

Static Public Attributes
dict	usable_params
	The params that you could set on the Algorithm object which this Strategy would use.
list	required_attrs
	Required attributes that must exist before the strategy can run properly.
list	required_true_attrs
	Attributes that must have a value that returns True when tested by :py:meth:is_valid.
list	allowed_granularities = ["run", "all"]
	Granularity of collector that can be run by this algorithm properly.
str	FINISHED_RESULTS = "DONE"
	Signal value that is put into the Queue when there are no more results left.
str	COMPLETED = "COMPLETED"
	Completed state.
str	FAILED = "FAILED"
	Failed state.

Detailed Description

Algorithm strategy to do run-by-run calibration of collected data.
Runs the algorithm over the input data contained within the requested IoV, starting with the first run's data only.
If the algorithm returns 'not enough data' on the current run set, it won't commit the payloads, but instead adds
the next run's data and tries again.

Once an execution on a set of runs return 'iterate' or 'ok' we move onto the next runs (if any are left)
and start the same procedure again. Committing of payloads to the outputdb only happens once we're sure that there
is enough data in the remaining runs to get a full execution. If there isn't enough data remaining, the last runs
are merged with the previous successful execution's runs and a final execution is performed on all remaining runs.

Additionally this strategy will automatically make sure that IoV gaps in your input data are covered by a payload.
This means that there shouldn't be any IoVs that don't get a new payload by the  end of running an iteration.

This uses a `caf.state_machines.AlgorithmMachine` to actually execute the various steps rather than operating on
a CalibrationAlgorithm C++ class directly.

Definition at line 260 of file strategies.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
			algorithm )

Definition at line 289 of file strategies.py.

    def __init__(self, algorithm):
        """
        """
        super().__init__(algorithm)
        
        self.machine = AlgorithmMachine(self.algorithm)
        if "step_size" not in self.algorithm.params:
            self.algorithm.params["step_size"] = 1
        
        self.first_execution = True
 

Member Function Documentation

any_failed_iov()

any_failed_iov ( self )

inherited

Returns:
    bool: If any result in the current results list has a failed algorithm code we return True

Definition at line 152 of file strategies.py.

    def any_failed_iov(self):
        """
        Returns:
            bool: If any result in the current results list has a failed algorithm code we return True
        """
        failed_results = []
        for result in self.results:
            if result.result == AlgResult.failure.value or result.result == AlgResult.not_enough_data.value:
                failed_results.append(result)
        if failed_results:
            B2WARNING("Failed results found.")
            for result in failed_results:
                if result.result == AlgResult.failure.value:
                    B2ERROR(f"c_Failure returned for {result.iov}.")
                elif result.result == AlgResult.not_enough_data.value:
                    B2WARNING(f"c_NotEnoughData returned for {result.iov}.")
            return True
        else:
            return False
 

apply_experiment_settings()

apply_experiment_settings	(		self,
			algorithm,
			experiment )

Apply experiment-dependent settings.
This is the default version, which does not do anything.
If necessary, it should be reimplemented by derived classes.

Definition at line 301 of file strategies.py.

    def apply_experiment_settings(self, algorithm, experiment):
        """
        Apply experiment-dependent settings.
        This is the default version, which does not do anything.
        If necessary, it should be reimplemented by derived classes.
        """
        return
 

execute_over_run_list()

execute_over_run_list	(		self,
			iteration,
			run_list,
			lowest_exprun,
			highest_exprun )

Execute runs given in list

Definition at line 393 of file strategies.py.

    def execute_over_run_list(self, iteration, run_list, lowest_exprun, highest_exprun):
        """Execute runs given in list"""
        # The runs (data) we have left to execute from this run list
        remaining_runs = run_list[:]
        # The previous execution's runs
        previous_runs = []
        # The current runs we are executing
        current_runs = []
        # The last successful payload and result
        last_successful_payloads = None
        last_successful_result = None
 
        # Iterate over ExpRuns within an experiment in chunks of 'step_size'
        for expruns in grouper(self.algorithm.params["step_size"], run_list):
            # Already set up earlier the first time, so we shouldn't do it again
            if not self.first_execution:
                self.machine.setup_algorithm()
            else:
                self.first_execution = False
 
            # Add on the next step of runs
            current_runs.extend(expruns)
            # Remove them from our remaining runs
            remaining_runs = [run for run in remaining_runs if run not in current_runs]
 
            # Is this the first payload of the experiment
            if not last_successful_result:
                B2INFO("Detected that this will be the first payload of this experiment.")
                # If this is the first payload but we have other data, we need the IoV to cover from the
                # lowest IoV extent requested up to the ExpRun right before the next run in the remaining runs list.
                if remaining_runs:
                    apply_iov = IoV(*lowest_exprun, remaining_runs[0].exp, remaining_runs[0].run - 1)
                # If this is the first payload but there isn't more data, we set the IoV to cover the full range
                else:
                    B2INFO("Detected that this will be the only payload of the experiment.")
                    apply_iov = IoV(*lowest_exprun, *highest_exprun)
            # If there were previous successes
            else:
                if not remaining_runs:
                    B2INFO("Detected that there are no more runs to execute in this experiment after this next execution.")
                    apply_iov = IoV(*current_runs[0], *highest_exprun)
                # Otherwise, it's just a normal IoV in the middle.
                else:
                    B2INFO("Detected that there are more runs to execute in this experiment after this next execution.")
                    apply_iov = IoV(*current_runs[0], remaining_runs[0].exp, remaining_runs[0].run - 1)
 
            B2INFO(f"Executing and applying {apply_iov} to the payloads.")
            self.machine.execute_runs(runs=current_runs, iteration=iteration, apply_iov=apply_iov)
            B2INFO(f"Finished execution with result code {self.machine.result.result}.")
 
            # Does this count as a successful execution?
            if (self.machine.result.result == AlgResult.ok.value) or (self.machine.result.result == AlgResult.iterate.value):
                self.machine.complete()
                # If we've succeeded but we have a previous success we can commit the previous payloads
                # since we have new ones ready
                if last_successful_payloads and last_successful_result:
                    B2INFO("Saving this execution's payloads to be committed later.")
                    # Save the payloads and result
                    new_successful_payloads = self.machine.algorithm.algorithm.getPayloadValues()
                    new_successful_result = self.machine.result
                    B2INFO("We just succeeded in execution of the Algorithm."
                           f" Will now commit payloads from the previous success for {last_successful_result.iov}.")
                    self.machine.algorithm.algorithm.commit(last_successful_payloads)
                    self.results.append(last_successful_result)
                    self.send_result(last_successful_result)
                    # If there are remaining runs we need to have the current payloads ready to commit after the next execution
                    if remaining_runs:
                        last_successful_payloads = new_successful_payloads
                        last_successful_result = new_successful_result
                    # If there's not more runs to process we should also commit the new ones
                    else:
                        B2INFO("We have no more runs to process. "
                               f"Will now commit the most recent payloads for {new_successful_result.iov}.")
                        self.machine.algorithm.algorithm.commit(new_successful_payloads)
                        self.results.append(new_successful_result)
                        self.send_result(new_successful_result)
                        break
                # if there's no previous success this must be the first run executed
                else:
                    # Need to save payloads for later if we have a success but runs remain
                    if remaining_runs:
                        B2INFO(f"Saving the most recent payloads for {self.machine.result.iov} to be committed later.")
                        # Save the payloads and result
                        last_successful_payloads = self.machine.algorithm.algorithm.getPayloadValues()
                        last_successful_result = self.machine.result
                    # Need to commit and exit if we have a success and no remaining data
                    else:
                        B2INFO("We just succeeded in execution of the Algorithm."
                               " No runs left to be processed, so we are committing results of this execution.")
                        self.machine.algorithm.algorithm.commit()
                        self.results.append(self.machine.result)
                        self.send_result(self.machine.result)
                        break
 
                previous_runs = current_runs[:]
                current_runs = []
            # If it wasn't successful, was it due to lack of data in the runs?
            elif (self.machine.result.result == AlgResult.not_enough_data.value):
                B2INFO(f"There wasn't enough data in {self.machine.result.iov}.")
                if remaining_runs:
                    B2INFO("Some runs remain to be processed. "
                           f"Will try to add at most {self.algorithm.params['step_size']} more runs of data and execute again.")
                elif not remaining_runs and not last_successful_result:
                    B2ERROR("There aren't any more runs remaining to merge with, and we never had a previous success."
                            " There wasn't enough data in the full input data requested.")
                    self.results.append(self.machine.result)
                    self.send_result(self.machine.result)
                    self.machine.fail()
                    break
                elif not remaining_runs and last_successful_result:
                    B2INFO("There aren't any more runs remaining to merge with. But we had a previous success"
                           ", so we'll merge with the previous IoV.")
                    final_runs = current_runs[:]
                    current_runs = previous_runs
                    current_runs.extend(final_runs)
                self.machine.fail()
            elif self.machine.result.result == AlgResult.failure.value:
                B2ERROR(f"{self.algorithm.name} returned failure exit code.")
                self.results.append(self.machine.result)
                self.send_result(self.machine.result)
                self.machine.fail()
                break
        else:
            # Check if we need to run a final execution on the previous execution + dangling set of runs
            if current_runs:
                self.machine.setup_algorithm()
                apply_iov = IoV(last_successful_result.iov.exp_low,
                                last_successful_result.iov.run_low,
                                *highest_exprun)
                B2INFO(f"Executing on {apply_iov}.")
                self.machine.execute_runs(runs=current_runs, iteration=iteration, apply_iov=apply_iov)
                B2INFO(f"Finished execution with result code {self.machine.result.result}.")
                if (self.machine.result.result == AlgResult.ok.value) or (
                        self.machine.result.result == AlgResult.iterate.value):
                    self.machine.complete()
                    # Commit all the payloads and send out the results
                    self.machine.algorithm.algorithm.commit()
                    # Save the result
                    self.results.append(self.machine.result)
                    self.send_result(self.machine.result)
                else:
                    # Save the result
                    self.results.append(self.machine.result)
                    self.send_result(self.machine.result)
                    # But failed
                    self.machine.fail()
 
 

find_iov_gaps()

find_iov_gaps ( self )

inherited

Finds and prints the current gaps between the IoVs of the strategy results. Basically these are the IoVs
not covered by any payload. It CANNOT find gaps if they exist across an experiment boundary. Only gaps
within the same experiment are found.

Returns:
    iov_gaps(list[IoV])

Definition at line 132 of file strategies.py.

    def find_iov_gaps(self):
        """
        Finds and prints the current gaps between the IoVs of the strategy results. Basically these are the IoVs
        not covered by any payload. It CANNOT find gaps if they exist across an experiment boundary. Only gaps
        within the same experiment are found.
 
        Returns:
            iov_gaps(list[IoV])
        """
        iov_gaps = find_gaps_in_iov_list(sorted([result.iov for result in self.results]))
        if iov_gaps:
            gap_msg = ["Found gaps between IoVs of algorithm results (regardless of result)."]
            gap_msg.append("You may have requested these gaps deliberately by not passing in data containing these runs.")
            gap_msg.append("This may not be a problem, but you will not have payoads defined for these IoVs")
            gap_msg.append("unless you edit the final database.txt yourself.")
            B2INFO_MULTILINE(gap_msg)
            for iov in iov_gaps:
                B2INFO(f"{iov} not covered by any execution of the algorithm.")
        return iov_gaps
 

is_valid()

is_valid ( self )

inherited

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

Definition at line 114 of file strategies.py.

    def is_valid(self):
        """
        Returns:
            bool: Whether or not this strategy has been set up correctly with all its necessary attributes.
        """
        B2INFO("Checking validity of current AlgorithmStrategy setup.")
        # 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"AlgorithmStrategy 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"AlgorithmStrategy attribute {attribute_name} returned False.")
                return False
        return True
 

run()

run	(		self,
			iov,
			iteration,
			queue )

Runs the algorithm machine over the collected data and fills the results.

Reimplemented from AlgorithmStrategy.

Definition at line 309 of file strategies.py.

    def run(self, iov, iteration, queue):
        """
        Runs the algorithm machine over the collected data and fills the results.
        """
        if not self.is_valid():
            raise StrategyError("This AlgorithmStrategy was not set up correctly!")
        self.queue = queue
        B2INFO(f"Setting up {self.__class__.__name__} strategy for {self.algorithm.name}.")
        # Now add all the necessary parameters for a strategy to run
        machine_params = {}
        machine_params["database_chain"] = self.database_chain
        machine_params["dependent_databases"] = self.dependent_databases
        machine_params["output_dir"] = self.output_dir
        machine_params["output_database_dir"] = self.output_database_dir
        machine_params["input_files"] = self.input_files
        machine_params["ignored_runs"] = self.ignored_runs
        self.machine.setup_from_dict(machine_params)
        # Start moving through machine states
        self.machine.setup_algorithm(iteration=iteration)
        # After this point, the logging is in the stdout of the algorithm
        B2INFO(f"Beginning execution of {self.algorithm.name} using strategy {self.__class__.__name__}.")
        runs_to_execute = []
        all_runs_collected = runs_from_vector(self.algorithm.algorithm.getRunListFromAllData())
        # If we were given a specific IoV to calibrate we just execute over runs in that IoV
        if iov:
            runs_to_execute = runs_overlapping_iov(iov, all_runs_collected)
        else:
            runs_to_execute = all_runs_collected[:]
 
        # Remove the ignored runs from our run list to execute
        if self.ignored_runs:
            B2INFO(f"Removing the ignored_runs from the runs to execute for {self.algorithm.name}.")
            runs_to_execute.difference_update(set(self.ignored_runs))
        # Sets aren't ordered so lets go back to lists and sort
        runs_to_execute = sorted(runs_to_execute)
 
        # We don't want to cross the boundary of Experiments accidentally. So we will split our run list
        # into separate lists, one for each experiment number contained. That way we can evaluate each experiment
        # separately and prevent IoVs from crossing the boundary.
        runs_to_execute = split_runs_by_exp(runs_to_execute)
 
        # Now iterate through the experiments, executing runs in blocks of 'step_size'. We DO NOT allow a payload IoV to
        # extend over multiple experiments, only multiple runs
        iov_coverage = None
        if "iov_coverage" in self.algorithm.params:
            B2INFO(f"Detected that you have set iov_coverage to {self.algorithm.params['iov_coverage']}.")
            iov_coverage = self.algorithm.params["iov_coverage"]
 
        number_of_experiments = len(runs_to_execute)
        # Iterate over experiment run lists
        for i_exp, run_list in enumerate(runs_to_execute, start=1):
 
            # Apply experiment-dependent settings.
            if "has_experiment_settings" in self.algorithm.params:
                if self.algorithm.params["has_experiment_settings"]:
                    self.apply_experiment_settings(self.machine.algorithm.algorithm, run_list[0].exp)
 
            # If 'iov_coverage' was set in the algorithm.params and it is larger (at both ends) than the
            # input data runs IoV, then we also have to set the first payload IoV to encompass the missing beginning
            # of the iov_coverage, and the last payload IoV must cover up to the end of iov_coverage.
            # This is only true for the lowest and highest experiments in our input data.
            # If we have multiple experiments the iov must be adjusted to avoid gaps at the iov boundaries
            lowest_exprun = ExpRun(run_list[0].exp, 0)
            highest_exprun = ExpRun(run_list[-1].exp, -1)
 
            if i_exp == 1:
                lowest_exprun = ExpRun(iov_coverage.exp_low, iov_coverage.run_low) if iov_coverage else run_list[0]
            if i_exp == number_of_experiments:
                highest_exprun = ExpRun(iov_coverage.exp_high, iov_coverage.run_high) if iov_coverage else run_list[-1]
 
            self.execute_over_run_list(iteration, run_list, lowest_exprun, highest_exprun)
 
        # Print any knowable gaps between result IoVs, if any are foun there is a problem.
        gaps = self.find_iov_gaps()
        # Dump them to a file for logging
        with open(f"{self.algorithm.name}_iov_gaps.json", "w") as f:
            json.dump(gaps, f)
 
        # If any results weren't successes we fail
        if self.any_failed_iov():
            self.send_final_state(self.FAILED)
        else:
            self.send_final_state(self.COMPLETED)
 

send_final_state()

send_final_state ( self, state )

inherited

send final state

Definition at line 176 of file strategies.py.

    def send_final_state(self, state):
        """send final state"""
        self.queue.put({"type": "final_state", "value": state})
 
 

send_result()

send_result ( self, result )

inherited

send result

Definition at line 172 of file strategies.py.

    def send_result(self, result):
        """send result"""
        self.queue.put({"type": "result", "value": result})
 

setup_from_dict()

setup_from_dict ( self, params )

inherited

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

Definition at line 106 of file strategies.py.

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

Member Data Documentation

algorithm

algorithm = algorithm

inherited

Algorithm() class that we're running.

Definition at line 80 of file strategies.py.

allowed_granularities

list allowed_granularities = ["run", "all"]

staticinherited

Granularity of collector that can be run by this algorithm properly.

Definition at line 65 of file strategies.py.

COMPLETED

str COMPLETED = "COMPLETED"

staticinherited

Completed state.

Definition at line 71 of file strategies.py.

database_chain

list database_chain = []

inherited

User defined database chain i.e.

the default global tag, or if you have localdb's/tags for custom alignment etc

Definition at line 88 of file strategies.py.

dependent_databases

list dependent_databases = []

inherited

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

Definition at line 90 of file strategies.py.

FAILED

str FAILED = "FAILED"

staticinherited

Failed state.

Definition at line 74 of file strategies.py.

FINISHED_RESULTS

str FINISHED_RESULTS = "DONE"

staticinherited

Signal value that is put into the Queue when there are no more results left.

Definition at line 68 of file strategies.py.

first_execution

bool first_execution = True

boolean storing whether this is the first time the algorithm is executed

Definition at line 299 of file strategies.py.

ignored_runs

list ignored_runs = []

inherited

Runs that will not be included in ANY execution of the algorithm.

Usually set by Calibration.ignored_runs. The different strategies may handle the resulting run gaps differently.

Definition at line 93 of file strategies.py.

input_files

list input_files = []

inherited

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

Definition at line 82 of file strategies.py.

machine

machine = AlgorithmMachine(self.algorithm)

:py:class:caf.state_machines.AlgorithmMachine used to help set up and execute CalibrationAlgorithm It gets setup properly in :py:func:run

Definition at line 295 of file strategies.py.

output_database_dir

str output_database_dir = ""

inherited

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

Definition at line 86 of file strategies.py.

output_dir

str output_dir = ""

inherited

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

Definition at line 84 of file strategies.py.

queue

queue = None

inherited

The multiprocessing Queue we use to pass back results one at a time.

Definition at line 97 of file strategies.py.

required_attrs

list required_attrs

staticinherited

Initial value:

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

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

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

Definition at line 48 of file strategies.py.

required_true_attrs

list required_true_attrs

staticinherited

Initial value:

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

Attributes that must have a value that returns True when tested by :py:meth:is_valid.

Definition at line 58 of file strategies.py.

results

list results = []

inherited

The list of results objects which will be sent out before the end.

Definition at line 95 of file strategies.py.

usable_params

dict usable_params

static

Initial value:

=  {
        "has_experiment_settings": bool,
        "iov_coverage": IoV,
        "step_size": int
    }

The params that you could set on the Algorithm object which this Strategy would use.

Just here for documentation reasons.

Definition at line 280 of file strategies.py.

---

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

• calibration/scripts/caf/strategies.py