ConfigBlock.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
 
import textwrap
import inspect
from dataclasses import dataclass
from functools import wraps
import warnings
 
from AnaAlgorithm.Logging import logging
logCPAlgCfgBlock = logging.getLogger('CPAlgCfgBlock')
 
from AnalysisAlgorithmsConfig.ConfigAccumulator import DataType, ExpertModeWarning
import re
 
def filter_dsids (filterList, config) :
    """check whether the sample being run passes a
    possible DSID filter on the block"""
    if len(filterList) == 0:
        return True
    for dsid_filter in filterList:
        # Check if the pattern is enclosed in regex delimiters (e.g., starts with '^' or contains regex metacharacters)
        if any(char in str(dsid_filter) for char in "^$*+?.()|[]{}\\"):
            pattern = re.compile(dsid_filter)
            if pattern.match(str(config.dsid())):
                return True
        else:
            # Otherwise it's an exact DSID (but could be int or string)
            if str(dsid_filter) == str(config.dsid()):
                return True
    return False
 
def alphanumeric_block_name(func):
    """this wrapper ensures that the 'instanceName' of the various
    config blocks is cleaned up of any non-alphanumeric characters
    that may arise from using 'selectionName' in the naming."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Get the string returned by the 'instanceName()' method of a config block
        orig_name = func(*args, **kwargs)
 
        if orig_name is None:
            return ""
 
        # Allowed replacements - anything else is likely a mistake on the user-side
        result = orig_name.replace("||", "OR").replace("&&", "AND").replace("(","LB").replace(")","RB").replace(" ","")
 
        return result
    return wrapper
 
class BlockNameProcessorMeta(type):
    """this meta class enforces the application of 'alphanumeric_block_names()'
    to 'instanceName()' and will be used in the main ConfigBlock class in order
    to propagate this rule also to all derived classes (the individual config blocks."""
    def __new__(cls, name, bases, dct):
        # Automatically apply alphanumeric-only decorator to 'instanceName()' method
        if 'instanceName' in dct and callable(dct['instanceName']):
            dct['instanceName'] = alphanumeric_block_name(dct['instanceName'])
        return super().__new__(cls, name, bases, dct)
 
@dataclass(eq=False)
class ConfigBlockOption:
    """the information for a single option on a configuration block"""
 
    type: object = None
    info: str = ''
    noneAction: str = 'ignore'
    required: bool = False
    default: object = None
    meta: object = None # metadata used only for downstream applications (e.g. docs)
 
 
 
@dataclass(eq=False)
class ConfigBlockDependency:
    """Class encoding a blocks dependence on other blocks."""
 
    blockName: str
    required: bool = True
 
 
    def __eq__(self, name):
        return self.blockName == name
 
 
    def __str__(self):
        return self.blockName
 
 
    def __repr__(self):
        return f'ConfigBlockDependency(blockName="{self.blockName}", required={self.required})'
 
 
class ConfigBlock(metaclass=BlockNameProcessorMeta):
    """the base class for classes implementing individual blocks of
    configuration
 
    A configuration block is a sequence of one or more algorithms that
    should always be scheduled together, e.g. the muon four momentum
    corrections could be a single block, muon selection could then be
    another block.  The blocks themselves generally have their own
    configuration options/properties specific to the block, and will
    perform a dynamic configuration based on those options as well as
    the overall job.
 
    The actual configuration of the algorithms in the block will
    depend on what other blocks are scheduled before and afterwards,
    most importantly some algorithms will introduce shallow copies
    that subsequent algorithms will need to run on, and some
    algorithms will add selection decorations that subsequent algorithms
    should use as preselections.
 
    The algorithms get created in a multi-step process (that may be
    extended in the future): As a first step each block retrieves
    references to the containers it uses (essentially marking its spot
    in the processing chain) and also registering any shallow copies
    that will be made.  In the second/last step each block then
    creates the fully configured algorithms.
 
    One goal is that when the algorithms get created they will have
    their final configuration and there needs to be no
    meta-configuration data attached to the algorithms, essentially an
    inversion of the approach in AnaAlgSequence in which the
    algorithms got created first with associated meta-configuration
    and then get modified in subsequent configuration steps.
 
    For now this is mostly an empty base class, but another goal of
    this approach is to make it easier to build another configuration
    layer on top of this one, and this class will likely be extended
    and get data members at that point.
 
    The child class needs to implement the method `makeAlgs` which is
    given a single `ConfigAccumulator` type argument. This is meant to
    create the sequence of algorithms that this block configures. This
    is currently (28 Jul 2025) called twice and should do the same thing
    during both calls, but the plan is to change that to a single call.
 
    The child class should also implement the method `getInstanceName`
    which should return a string that is used to distinguish between
    multiple instances of the same block. This is used to append the
    instance name to the names of all algorithms created by this block,
    and may in the future also be used to distinguish between multiple
    instances of the block.
    """
 
    # Class-level dictionary to keep track of instance counts for each derived class
    instance_counts = {}
 
    def __init__ (self) :
        self._blockName = ''
        self._factoryName = None
        self._dependencies = []
        self._options = {} # used with block configuration to set arbitrary option
        self._expertModeSettings = {} # dictionary to track expert mode requirements for each option
        self.addOption('groupName', '', type=str,
            info=('Used to specify this block when setting an'
                ' option at an arbitrary location.'))
        self.addOption('skipOnData', False, type=bool,
            info=('User option to prevent the block from running'
                  ' on data. This only affects blocks that are'
                  ' intended to run on data.'))
        self.addOption('skipOnMC', False, type=bool,
            info=('User option to prevent the block from running'
                  ' on MC. This only affects blocks that are'
                  ' intended to run on MC.'))
        self.addOption('skipWithSystematics', False, type=bool,
            info=('User option to prevent the block from running with systematics.'))
        self.addOption('onlyForDSIDs', [], type=list,
            info=('Used to specify which MC DSIDs to allow this'
                  ' block to run on. Each element of the list'
                  ' can be a full DSID (e.g. 410470), or a regex'
                  ' (e.g. 410.* to select all 410xxx DSIDs, or'
                  ' ^(?!410) to veto them). An empty list means no'
                  ' DSID restriction.'))
        self.addOption('propertyOverrides', {}, type=dict,
            info=('EXPERT USE ONLY: A dictionary of properties to'
                  ' override at the end of configuration. This should'
                  ' take the form'
                  ' {"algName.toolName.propertyName": value, ...},'
                  ' without any automatically applied postfixes for'
                  ' the algorithm name. THIS IS MEANT TO BE EXPERT'
                  ' USAGE ONLY. Properties that need to be set by'
                  ' the user should be declared as options on the'
                  ' block itself. EXPERT USE ONLY!'),
                  expertMode=True)
        # Increment the instance count for the current class
        cls = type(self)  # Get the actual class of the instance (also derived!)
        if cls not in ConfigBlock.instance_counts:
            ConfigBlock.instance_counts[cls] = 0
        # Note: we do need to check in the call stack that we are
        # in a real makeConfig situation, and not e.g. printAlgs
        frame = inspect.currentframe()
        while frame is not None:
            # Get the class name (if any) from the frame
            parent_cls = frame.f_locals.get('self', None)
            if parent_cls is None or not isinstance(parent_cls, ConfigBlock):
                # If the frame does not belong to an instance of ConfigBlock, it's an external caller
                if frame.f_code.co_name == "makeConfig":
                    ConfigBlock.instance_counts[cls] += 1
                    break
            frame = frame.f_back
        del frame
 
 
    def setBlockName(self, name):
        """Set blockName"""
        self._blockName = name
 
    def getBlockName(self):
        """Get blockName"""
        return self._blockName
 
    def factoryName(self):
        """get the factory name for this block
        
        This is mostly to give a reliable means of identifying the type
        of block we have in error messages. This is meant to be
        automatically set by the factory based on the requested block
        name, but there are a number of fallbacks. It is best not to
        assume a specific format, this is mostly meant to be used as an
        identifier in output messages.
        """
        if self._factoryName is not None and self._factoryName != '':
            return self._factoryName
        # If no factory name is set and the block has a name, use that
        if self._blockName is not None and self._blockName != '':
            return self._blockName
        # Use the class name as a fallback
        return self.__class__.__name__
 
    def setFactoryName(self, name):
        """set the factory name for this block
        
        This is meant to be called automatically by the factory based on
        the requested block name. If you are creating a block without a factory,
        you can call this method to set the factory name manually.
        """
        self._factoryName = name
 
    def instanceName(self):
        """Get the name of the instance
 
        The name of the instance is used to distinguish between multiple
        instances of the same block. Most importantly, this will be
        appended to the names of all algorithms created by this block.
        This defaults to an empty string, but block implementations
        should override it with an appropriate name based on identifying
        options set on this instance. A typical example would be the
        name of the (main) container, plus potentially the selection or
        working point.
 
        Ideally all blocks should override this method, but for backward
        compatibility (28 Jul 25) it defaults to an empty string.
        """
        return ''
 
    def isUsedForConfig(self, config):
        """
        whether this block should be used for the given configuration
 
        This is used by `ConfigSequence` to determine whether this block
        should be included in the configuration.
        """
        if self.skipWithSystematics and not config.noSystematics():
            return False
        if self.skipOnData and config.dataType() is DataType.Data:
            return False
        if self.skipOnMC and config.dataType() is not DataType.Data:
            return False
        if not filter_dsids(self.onlyForDSIDs, config):
            return False
        return True
 
    def applyConfigOverrides(self, config):
        """
        Apply any configuration overrides specified in the block's
        `propertyOverrides` option. This is meant to be called at the
        end of the configuration process, after all algorithms have been
        created and configured.
        """
        for key, value in self.propertyOverrides.items():
            # Split the key into algorithm name, tool name, and property name
            parts = key.split('.')
            if len(parts) < 2:
                raise Exception(f"Invalid override key format: {key}")
            alg = config.getAlgorithm(parts[0])
            if alg is None:
                raise Exception(f"Algorithm {parts[0]} not found in config for override: {key}")
            for name in parts[1:-1]:
                # Navigate through tools if necessary
                if hasattr(alg, name):
                    alg = getattr(alg, name)
                else:
                    raise Exception(f"Tool {name} not found for override: {key}")
            # Set the property on the algorithm/tool.
            setattr(alg, parts[-1], value)
 
    def addDependency(self, dependencyName, required=True):
        """
        Add a dependency for the block. Dependency is corresponds to the
        blockName of another block. If required is True, will throw an
        error if dependency is not present; otherwise will move this
        block after the required block. If required is False, will do
        nothing if required block is not present; otherwise, it will
        move block after required block.
        """
        if not self.hasDependencies():
            # add option to block ignore dependencies
            self.addOption('ignoreDependencies', [], type=list,
                           info='List of dependencies defined in the ConfigBlock to ignore.')
        self._dependencies.append(ConfigBlockDependency(dependencyName, required))
 
    def hasDependencies(self):
        """Return True if there is a dependency."""
        return bool(self._dependencies)
 
    def getDependencies(self):
        """Return the list of dependencies. """
        return self._dependencies
 
    def addOption (self, name, defaultValue, *,
                   type, info='', noneAction='ignore', required=False,
                   expertMode=None, meta=None) :
        """declare the given option on the configuration block
 
        This should only be called in the constructor of the
        configuration block.
 
        NOTE: The backend to option handling is slated to be replaced
        at some point.  This particular function should essentially
        stay the same, but some behavior may change.
        """
        if name in self._options :
            raise KeyError (f'duplicate option: {name}')
        if type not in [str, bool, int, float, list, dict, None] :
            raise TypeError (f'unknown option type: {type}')
        noneActions = ['error', 'set', 'ignore']
        if noneAction not in noneActions :
            raise ValueError (f'invalid noneAction: {noneAction} [allowed values: {noneActions}]')
 
        # Store expert mode settings if provided
        if expertMode is not None:
            if expertMode is True:
                # in this case we will just check against the default value
                self._expertModeSettings[name] = True
            elif not isinstance(expertMode, list):
                raise TypeError (f'expertMode must be a list, got {expertMode.__class__}')
            else:
                # here we will check against a list of custom values
                self._expertModeSettings[name] = expertMode
 
        if meta is not None:
            if not isinstance(meta, dict):
                raise TypeError(f'meta must be a dictionary, got {meta.__class__}')
 
            unknown = set(meta) - {'choices', 'role'}
            if unknown:
                raise ValueError(f'meta received unknown keys: {unknown}')
 
            if 'choices' in meta:
                choices = meta['choices']
                if (not isinstance(choices, tuple)
                    or len(choices) != 2
                    or not isinstance(choices[0], list)
                    or not all(isinstance(choice, str) for choice in choices[0])
                    or (choices[1] is not None and not isinstance(choices[1], int))
                ):
                    raise TypeError("meta['choices'] must be a (list[str], int | None) tuple")
 
            if 'role' in meta:
                role = meta['role']
                if role not in {'container', 'containerRef', 'selection', 'region'}:
                    # container: defines a new container name
                    # containerRef: expects a container and possibly a selection, i.e. 'container' or 'container.selection'
                    # selection: defines a new selection name
                    # region: specifically for event selections
                    raise ValueError(f"meta['role'] must be one of 'container', 'containerRef', 'selection', 'region', got '{role}'")
 
        setattr (self, name, defaultValue)
        self._options[name] = ConfigBlockOption(type=type, info=info,
                                                noneAction=noneAction, required=required,
                                                default=defaultValue, meta=meta)
 
 
    def setOptionValue (self, name, value) :
        """set the given option on the configuration block
 
        NOTE: The backend to option handling is slated to be replaced
        at some point.  This particular function should essentially
        stay the same, but some behavior may change.
        """
 
        if name not in self._options :
            raise KeyError (f'unknown option "{name}" in block "{self.__class__.__name__}"')
        noneAction = self._options[name].noneAction
        if value is not None or noneAction == 'set' :
            # check type if specified
            optType = self._options[name].type
            # convert int to float to prevent crash
            if optType is float and type(value) is int:
                value = float(value)
            if optType is not None and optType != type(value):
                raise ValueError(f'{name} for block {self.__class__.__name__} should '
                    f'be of type {optType} not {type(value)}')
            setattr (self, name, value)
        elif noneAction == 'ignore' :
            pass
        elif noneAction == 'error' :
            raise ValueError (f'passed None for setting option {name} with noneAction=error')
 
 
    def getOptionValue(self, name):
        """Returns config option value, if present; otherwise return None"""
        if name in self._options:
            return getattr(self, name)
 
 
    def getOptions(self):
        """Return a copy of the options associated with the block"""
        return self._options.copy()
 
 
    def printOptions(self, verbose=False, width=60, indent="    "):
        """
        Prints options and their values
        """
        def printWrap(text, width=60, indent="    "):
            wrapper = textwrap.TextWrapper(width=width, initial_indent=indent,
                subsequent_indent=indent)
            for line in wrapper.wrap(text=text):
                logCPAlgCfgBlock.info(line)
 
        for opt, vals in self.getOptions().items():
            if verbose:
                logCPAlgCfgBlock.info(indent + f"\033[4m{opt}\033[0m: {self.getOptionValue(opt)}")
                logCPAlgCfgBlock.info(indent*2 + f"\033[4mtype\033[0m: {vals.type}")
                logCPAlgCfgBlock.info(indent*2 + f"\033[4mdefault\033[0m: {vals.default}")
                logCPAlgCfgBlock.info(indent*2 + f"\033[4mrequired\033[0m: {vals.required}")
                logCPAlgCfgBlock.info(indent*2 + f"\033[4mnoneAction\033[0m: {vals.noneAction}")
                printWrap(f"\033[4minfo\033[0m: {vals.info}", indent=indent*2)
            else:
                logCPAlgCfgBlock.info(indent + f"{ opt}: {self.getOptionValue(opt)}")
 
 
    def hasOption (self, name) :
        """whether the configuration block has the given option
 
        WARNING: The backend to option handling is slated to be
        replaced at some point.  This particular function may change
        behavior, interface or be removed/replaced entirely.
        """
        return name in self._options
 
 
    def __eq__(self, blockName):
        """
        Implementation of == operator. Used for searching configSeq.
        E.g. if blockName in configSeq:
        """
        return self._blockName == blockName
 
 
    def __str__(self):
        return self._blockName if self._blockName else self.factoryName()
 
 
    @classmethod
    def get_instance_count(cls):
        # Access the current count for this class
        return ConfigBlock.instance_counts.get(cls, 0)
 
    def _is_expert_value(self, rule, value):
        """
        Check whether value matches an expert mode rule.
        Rule can be:
        - A literal (compared with ==)
        - A callable predicate (called with value)
        - A special marker string (common callable)
        """
        if callable(rule):
            return rule(value)
 
        if isinstance(rule, str):
            if rule == "nonemptystring":
                return isinstance(value, str) and value != ""
            if rule == "nonemptylist":
                return isinstance(value, list) and value != []
            if rule == "positiveint":
                return isinstance(value, int) and value > 0
 
        # Fallback: direct value comparison
        return value == rule
 
    def checkExpertSettings(self, config):
        """
        Check if any settings require expert mode and validate accordingly.
        If any setting is set to a value that requires expert mode but we're
        not in expert mode, raise an error.
        """
        for option_name, expert_rule in self._expertModeSettings.items():
            current_value = self.getOptionValue(option_name)
            default_value = self._options[option_name].default
 
            if expert_rule is True:
                # Any deviation from the default requires expert mode
                if current_value != default_value:
                    warnings.warn(
                        f"Block '{self.factoryName()}' option '{option_name}' "
                        f"set to '{current_value}' (default '{default_value}'), "
                        f"requires expert mode.",
                        ExpertModeWarning, stacklevel=2
                    )
 
            else:  # it's a list of expert values/markers/predicates
                for ev in expert_rule:
                    if self._is_expert_value(ev, current_value):
                        warnings.warn(
                            f"Block '{self.factoryName()}' option '{option_name}' "
                            f"set to expert-only value '{current_value}'. "
                            f"Requires expert mode.",
                            ExpertModeWarning, stacklevel=2
                        )
        # All checks passed
        return