SequentialBoundaries Class Reference

Inheritance diagram for SequentialBoundaries:

Public Member Functions
	__init__ (self, algorithm)
	run (self, iov, iteration, queue)
	execute_over_boundaries (self, boundary_iovs_to_run_lists, lowest_exprun, highest_exprun, iteration)
	execute_runs (self, runs, iteration, iov)
	alg_success (self)
	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 first calculate run boundaries where execution should be attempted.
Runs the algorithm over the input data contained within the requested IoV of the boundaries,
starting with the first boundary data only.
If the algorithm returns 'not enough data' on the current boundary IoV, it won't commit the payloads,
but instead adds the next boundarie's data and tries again. Basically the same logic as `SequentialRunByRun`
but using run boundaries instead of runs directly.
Notice that boundaries cannot span multiple experiments.

By default the algorithm will get the payload boundaries directly from the algorithm that need to
have implemented the function ``isBoundaryRequired``. If the desired boundaries are already known it
is possible to pass them directly setting the algorithm parameter ``payload_boundaries`` and avoid
the need to define the  ``isBoundaryRequired`` function.

``payload_boundaries`` is a list ``[(exp1, run1), (exp2, run2), ...]``. A boundary at the beginning of each
experiment will be added if not already present. An empty list will thus produce a single payload for each
experiment. A ``payload_boundaries`` set to ``None`` is equivalent to not passing it and restores the default
behaviour where the boundaries are computed in the ``isBoundaryRequired`` function of the algorithm.

Definition at line 658 of file strategies.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
			algorithm )

Definition at line 688 of file strategies.py.

    def __init__(self, algorithm):
        """
        """
        super().__init__(algorithm)
        
        self.machine = AlgorithmMachine(self.algorithm)
        
        self.first_execution = True
 

Member Function Documentation

alg_success()

alg_success

(

self

)

Check whether algorithm was successful

Definition at line 1057 of file strategies.py.

    def alg_success(self):
        """Check whether algorithm was successful"""
        return ((self.machine.result.result == AlgResult.ok.value) or (self.machine.result.result == AlgResult.iterate.value))
 
 

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
 

execute_over_boundaries()

execute_over_boundaries	(		self,
			boundary_iovs_to_run_lists,
			lowest_exprun,
			highest_exprun,
			iteration )

Take the previously found boundaries and the run lists they correspond to and actually perform the
Algorithm execution. This is assumed to be for a single experiment.

Definition at line 868 of file strategies.py.

    def execute_over_boundaries(self, boundary_iovs_to_run_lists, lowest_exprun, highest_exprun, iteration):
        """
        Take the previously found boundaries and the run lists they correspond to and actually perform the
        Algorithm execution. This is assumed to be for a single experiment.
        """
        # Copy of boundary IoVs
        remaining_boundary_iovs = sorted(list(boundary_iovs_to_run_lists.keys())[:])
 
        # The current runs we are executing
        current_runs = []
        # The IoV of the current boundary(s)
        current_boundary_iov = None
        # The current execution's applied IoV, may be different to the boundary IoV
        current_iov = None
 
        # The last successful payload list and result. We hold on to them so that we can commit or discard later.
        last_successful_payloads = None
        last_successful_result = None
        # The previous execution's runs
        last_successful_runs = []
        # The previous execution's applied IoV
        last_successful_iov = None
 
        while True:
            # Do we have previous successes?
            if not last_successful_result:
                if not current_runs:
                    # Did we actually have any boundaries?
                    if not remaining_boundary_iovs:
                        # Fail because we have no boundaries to use
                        B2ERROR("No boundaries found for the current experiment's run list. Failing the strategy.")
                        return False
 
                    B2INFO("This appears to be the first attempted execution of the experiment.")
                    # Attempt to execute on the first boundary
                    current_boundary_iov = remaining_boundary_iovs.pop(0)
                    current_runs = boundary_iovs_to_run_lists[current_boundary_iov]
                    # What if there is only one boundary? Need to apply the highest exprun
                    if not remaining_boundary_iovs:
                        current_iov = IoV(*lowest_exprun, *highest_exprun)
                    else:
                        current_iov = IoV(*lowest_exprun, current_boundary_iov.exp_high, current_boundary_iov.run_high)
                # Returned not enough data from first execution
                else:
                    # Any remaining boundaries?
                    if not remaining_boundary_iovs:
                        # Fail because we have no boundaries to use
                        B2ERROR("Not enough data found for the current experiment's run list. Failing the strategy.")
                        return False
 
                    B2INFO("There wasn't enough data previously. Merging with the runs from the next boundary.")
                    # Extend the previous run lists/iovs
                    next_boundary_iov = remaining_boundary_iovs.pop(0)
                    current_boundary_iov = IoV(current_boundary_iov.exp_low, current_boundary_iov.run_low,
                                               next_boundary_iov.exp_high, next_boundary_iov.run_high)
                    current_runs.extend(boundary_iovs_to_run_lists[next_boundary_iov])
                    # At the last boundary? Need to apply the highest exprun
                    if not remaining_boundary_iovs:
                        current_iov = IoV(current_iov.exp_low, current_iov.run_low, *highest_exprun)
                    else:
                        current_iov = IoV(current_iov.exp_low, current_iov.run_low,
                                          current_boundary_iov.exp_high, current_boundary_iov.run_high)
 
                self.execute_runs(current_runs, iteration, current_iov)
 
                # Does this count as a successful execution?
                if self.alg_success():
                    # Commit previous values we were holding onto
                    B2INFO("Found a success. Will save the payloads for later.")
                    # Save success
                    last_successful_payloads = self.machine.algorithm.algorithm.getPayloadValues()
                    last_successful_result = self.machine.result
                    last_successful_runs = current_runs[:]
                    last_successful_iov = current_iov
                    # Reset values for next loop
                    current_runs = []
                    current_boundary_iov = None
                    current_iov = None
                    self.machine.complete()
                    continue
                elif self.machine.result.result == AlgResult.not_enough_data.value:
                    B2INFO("Not Enough Data result.")
                    # Just complete but leave the current runs alone for next loop
                    self.machine.complete()
                    continue
                else:
                    B2ERROR("Hit a failure or some kind of result we can't continue from. Failing out...")
                    self.machine.fail()
                    return False
            # Previous result exists
            else:
                # Previous loop was a success
                if not current_runs:
                    # Remaining boundaries?
                    if not remaining_boundary_iovs:
                        # Out of data, can now commit
                        B2INFO("Finished this experiment's boundaries. "
                               f"Committing remaining payloads from {last_successful_result.iov}")
                        self.machine.algorithm.algorithm.commit(last_successful_payloads)
                        self.results.append(last_successful_result)
                        self.send_result(last_successful_result)
                        return True
 
                    # Remaining boundaries exist so we try to execute
                    current_boundary_iov = remaining_boundary_iovs.pop(0)
                    current_runs = boundary_iovs_to_run_lists[current_boundary_iov]
                    # What if there is only one boundary? Need to apply the highest exprun
                    if not remaining_boundary_iovs:
                        current_iov = IoV(current_boundary_iov.exp_low, current_boundary_iov.run_low, *highest_exprun)
                    else:
                        current_iov = current_boundary_iov
 
                # Returned not enough data from last execution
                else:
                    # Any remaining boundaries?
                    if not remaining_boundary_iovs:
                        B2INFO("We have no remaining runs to increase the amount of data. "
                               "Instead we will merge with the previous successful runs.")
                        # Merge with previous success IoV
                        new_current_runs = last_successful_runs[:]
                        new_current_runs.extend(current_runs)
                        current_runs = new_current_runs[:]
                        current_iov = IoV(last_successful_iov.exp_low, last_successful_iov.run_low,
                                          current_iov.exp_high, current_iov.run_high)
                        # We reset the last successful stuff because we are dropping it
                        last_successful_payloads = []
                        last_successful_result = None
                        last_successful_runs = []
                        last_successful_iov = None
 
                    else:
                        B2INFO("Since there wasn't enough data previously, we will merge with the runs from the next boundary.")
                        # Extend the previous run lists/iovs
                        next_boundary_iov = remaining_boundary_iovs.pop(0)
                        current_boundary_iov = IoV(current_boundary_iov.exp_low, current_boundary_iov.run_low,
                                                   next_boundary_iov.exp_high, next_boundary_iov.run_high)
                        # Extend previous execution's runs with the next set
                        current_runs.extend(boundary_iovs_to_run_lists[next_boundary_iov])
                        # At the last boundary? Need to apply the highest exprun
                        if not remaining_boundary_iovs:
                            current_iov = IoV(current_iov.exp_low, current_iov.run_low, *highest_exprun)
                        else:
                            current_iov = IoV(current_iov.exp_low, current_iov.run_low,
                                              current_boundary_iov.exp_high, current_boundary_iov.run_high)
 
                self.execute_runs(current_runs, iteration, current_iov)
 
                # Does this count as a successful execution?
                if self.alg_success():
                    # Commit previous values we were holding onto
                    B2INFO("Found a success.")
                    if last_successful_result:
                        B2INFO("Can now commit the previous success.")
                        self.machine.algorithm.algorithm.commit(last_successful_payloads)
                        self.results.append(last_successful_result)
                        self.send_result(last_successful_result)
                    # Replace last success
                    last_successful_payloads = self.machine.algorithm.algorithm.getPayloadValues()
                    last_successful_result = self.machine.result
                    last_successful_runs = current_runs[:]
                    last_successful_iov = current_iov
                    # Reset values for next loop
                    current_runs = []
                    current_boundary_iov = None
                    current_iov = None
                    self.machine.complete()
                    continue
                elif self.machine.result.result == AlgResult.not_enough_data.value:
                    B2INFO("Not Enough Data result.")
                    # Just complete but leave the current runs alone for next loop
                    self.machine.complete()
                    continue
                else:
                    B2ERROR("Hit a failure or some other result we can't continue from. Failing out...")
                    self.machine.fail()
                    return False
 

execute_runs()

execute_runs	(		self,
			runs,
			iteration,
			iov )

Execute runs

Definition at line 1045 of file strategies.py.

    def execute_runs(self, runs, iteration, iov):
        """Execute runs"""
        # 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
 
        B2INFO(f"Executing and applying {iov} to the payloads.")
        self.machine.execute_runs(runs=runs, iteration=iteration, apply_iov=iov)
        B2INFO(f"Finished execution with result code {self.machine.result.result}.")
 

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 698 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. 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"]
 
        payload_boundaries = None
        if "payload_boundaries" in self.algorithm.params:
            B2INFO(f"Detected that you have set payload_boundaries to {self.algorithm.params['payload_boundaries']}.")
            payload_boundaries = self.algorithm.params["payload_boundaries"]
 
        number_of_experiments = len(runs_to_execute)
        B2INFO(f"We are iterating over {number_of_experiments} experiments.")
 
        # Iterate over experiment run lists
        for i_exp, run_list in enumerate(runs_to_execute, start=1):
            B2DEBUG(26, f"Run List for this experiment={run_list}")
            current_experiment = run_list[0].exp
            B2INFO(f"Executing over data from experiment {current_experiment}")
            # 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 i_exp == 1:
                if iov_coverage:
                    lowest_exprun = ExpRun(iov_coverage.exp_low, iov_coverage.run_low)
                else:
                    lowest_exprun = run_list[0]
            # We are calibrating across multiple experiments so we shouldn't start from the middle but from the 0th run
            else:
                lowest_exprun = ExpRun(current_experiment, 0)
 
            # Override the normal value for the highest ExpRun (from data) if iov_coverage was set
            if iov_coverage and i_exp == number_of_experiments:
                highest_exprun = ExpRun(iov_coverage.exp_high, iov_coverage.run_high)
            # If we have more experiments to execute then we will be setting the final payload IoV in this experiment
            # to be unbounded
            elif i_exp < number_of_experiments:
                highest_exprun = ExpRun(current_experiment, -1)
            # Otherwise just get the values from data
            else:
                highest_exprun = run_list[-1]
 
            # Find the boundaries for this experiment's runs
            vec_run_list = vector_from_runs(run_list)
            if payload_boundaries is None:
                # Find the boundaries using the findPayloadBoundaries implemented in the algorithm
                B2INFO("Attempting to find payload boundaries.")
                vec_boundaries = self.algorithm.algorithm.findPayloadBoundaries(vec_run_list)
                # If this vector is empty then that's bad. Maybe the isBoundaryRequired function
                # wasn't implemented? Either way we should stop.
                if vec_boundaries.empty():
                    B2ERROR("No boundaries found but we are in a strategy that requires them! Failing...")
                    # Tell the Runner that we have failed
                    self.send_final_state(self.FAILED)
                    break
                vec_boundaries = runs_from_vector(vec_boundaries)
            else:
                # Using boundaries set by user
                B2INFO(f"Using as payload boundaries {payload_boundaries}.")
                vec_boundaries = [ExpRun(exp, run) for exp, run in payload_boundaries]
                # No need to check that vec_boundaries is not empty. In case it is we will anyway add
                # a boundary at the first run of each experiment.
            # Remove any boundaries not from the current experiment (only likely if they were set manually)
            # We sort just to make everything easier later and just in case something mad happened.
            run_boundaries = sorted([er for er in vec_boundaries if er.exp == current_experiment])
            # In this strategy we consider separately each experiment. We then now check that the
            # boundary (exp, 0) is present and if not we add it. It is indeed possible to miss it
            # if the boundaries were given manually
            first_exprun = ExpRun(current_experiment, 0)
            if first_exprun not in run_boundaries:
                B2WARNING(f"No boundary found at ({current_experiment}, 0), adding it.")
                run_boundaries[0:0] = [first_exprun]
            B2INFO(f"Found {len(run_boundaries)} boundaries for this experiment. "
                   "Checking if we have some data for all boundary IoVs...")
            # First figure out the run lists to use for each execution (potentially different from the applied IoVs)
            # We use the boundaries and the run_list
            boundary_iovs_to_run_lists = find_run_lists_from_boundaries(run_boundaries, run_list)
            B2DEBUG(26, f"Boundary IoVs before checking data = {boundary_iovs_to_run_lists}")
            # If there were any boundary IoVs with no run data, just remove them. Otherwise they will execute over all data.
            boundary_iovs_to_run_lists = {key: value for key, value in boundary_iovs_to_run_lists.items() if value}
            B2DEBUG(26, f"Boundary IoVs after checking data = {boundary_iovs_to_run_lists}")
            # If any were removed then we might have gaps between the boundary IoVs. Fix those now by merging IoVs.
            new_boundary_iovs_to_run_lists = {}
            previous_boundary_iov = None
            previous_boundary_run_list = None
            for boundary_iov, run_list in boundary_iovs_to_run_lists.items():
                if not previous_boundary_iov:
                    previous_boundary_iov = boundary_iov
                    previous_boundary_run_list = run_list
                    continue
                # We are definitely dealiing with IoVs from one experiment so we can make assumptions here
                if previous_boundary_iov.run_high != (boundary_iov.run_low-1):
                    B2WARNING("Gap in boundary IoVs found before execution! "
                              "Will correct it by extending the previous boundary up to the next one.")
                    B2INFO(f"Original boundary IoV={previous_boundary_iov}")
                    previous_boundary_iov = IoV(previous_boundary_iov.exp_low, previous_boundary_iov.run_low,
                                                previous_boundary_iov.exp_high, boundary_iov.run_low-1)
                    B2INFO(f"New boundary IoV={previous_boundary_iov}")
                new_boundary_iovs_to_run_lists[previous_boundary_iov] = previous_boundary_run_list
                previous_boundary_iov = boundary_iov
                previous_boundary_run_list = run_list
            else:
                new_boundary_iovs_to_run_lists[previous_boundary_iov] = previous_boundary_run_list
            boundary_iovs_to_run_lists = new_boundary_iovs_to_run_lists
            B2DEBUG(26, f"Boundary IoVs after fixing gaps = {boundary_iovs_to_run_lists}")
            # Actually execute now that we have an IoV list to apply
            success = self.execute_over_boundaries(boundary_iovs_to_run_lists, lowest_exprun, highest_exprun, iteration)
            if not success:
                # Tell the Runner that we have failed
                self.send_final_state(self.FAILED)
                break
        # Only executes if we didn't fail any experiment execution
        else:
            # Print any knowable gaps between result IoVs, if any are found there is a problem, but not necessarily too bad.
            gaps = self.find_iov_gaps()
            if gaps:
                B2WARNING("There were gaps between the output IoV payloads! See the JSON file in the algorithm output directory.")
            # 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 696 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 694 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:

=  {
        "iov_coverage": IoV,
        "payload_boundaries": []  # [(exp1, run1), (exp2, run2), ...]
    }

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

Just here for documentation reasons.

Definition at line 680 of file strategies.py.

---

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

• calibration/scripts/caf/strategies.py