python.ConfigText Namespace Reference

Classes
class	TextConfig
class	TextConfigWarning

Functions
def	readYaml (yamlPath)
def	printYaml (d, sort=False, jsonFormat=False)
def	makeSequence (configPath, *flags=None, algSeq=None, noSystematics=None, dataType=None, geometry=None, autoconfigFromFlags=None, isPhyslite=None, noPhysliteBroken=False)
def	combineConfigFiles (local, config_path, fragment_key="include")
def	_load_fragment (pathlib.Path fragment_path)
def	_find_fragment (fragment_path, config_path)
def	_merge_dicts (local, fragment)

Variables
	logCPAlgTextCfg = logging.getLogger('CPAlgTextCfg')

Function Documentation

_find_fragment()

def python.ConfigText._find_fragment (   fragment_path,   config_path  )

private

Definition at line 459 of file ConfigText.py.

def _find_fragment(fragment_path, config_path):
    paths_to_check = [
        fragment_path,
        config_path / fragment_path,
        *[x / fragment_path for x in os.environ["DATAPATH"].split(":")]
    ]
    for path in paths_to_check:
        if path.exists():
            return path
 
    raise FileNotFoundError(fragment_path)
 
 

_load_fragment()

def python.ConfigText._load_fragment ( pathlib.Path  fragment_path )

private

Load a YAML or JSON fragment

This function is superfluous as of the yaml 1.2 spec (which
has not been implemented in ATLAS Yaml dependencies).
Once https://github.com/yaml/pyyaml/issues/173 is resolved
pyyaml will support yaml 1.2, which is compatable with json. 
Until then yaml and json behave differently in some scientific
notation edge cases.

Definition at line 442 of file ConfigText.py.

def _load_fragment(fragment_path: pathlib.Path):
    """Load a YAML or JSON fragment
 
    This function is superfluous as of the yaml 1.2 spec (which
    has not been implemented in ATLAS Yaml dependencies).
    Once https://github.com/yaml/pyyaml/issues/173 is resolved
    pyyaml will support yaml 1.2, which is compatable with json. 
    Until then yaml and json behave differently in some scientific
    notation edge cases.
    """
    
    with open(fragment_path, 'r') as fragment_file:
        if fragment_path.suffix.lower() == '.json':
            return json.load(fragment_file)
        else:
            return yaml.safe_load(fragment_file)
 

_merge_dicts()

def python.ConfigText._merge_dicts (   local,   fragment  )

private

Definition at line 472 of file ConfigText.py.

def _merge_dicts(local, fragment):
    # In the list case append the fragment to the local list
    if isinstance(local, list):
        local += fragment
        return
    # In the dict case, append only missing values to local: the local
    # values take precedence over the fragment ones.
    if isinstance(local, dict):
        for key, value in fragment.items():
            if key in local:
                _merge_dicts(local[key], value)
            else:
                local[key] = value
        return

combineConfigFiles()

def python.ConfigText.combineConfigFiles	(		local,
			config_path,
			fragment_key = "include"
	)

Recursively combine configuration fragments into `local`.

- Looks for `fragment_key` at any dict node.
- If value is a string/path: merge that fragment.
- If value is a list: merge all fragments in order.
  For conflicts between fragments, the **earlier** file in the list wins. 
  Local keys still override the merged fragments.

Returns True if any merging happened below this node.

Definition at line 373 of file ConfigText.py.

def combineConfigFiles(local, config_path, fragment_key="include"):
    """
    Recursively combine configuration fragments into `local`.
 
    - Looks for `fragment_key` at any dict node.
    - If value is a string/path: merge that fragment.
    - If value is a list: merge all fragments in order.
      For conflicts between fragments, the **earlier** file in the list wins. 
      Local keys still override the merged fragments.
 
    Returns True if any merging happened below this node.
    """
    combined = False
 
    # If this isn't an iterable there's nothing to combine
    if isinstance(local, dict):
        to_combine = local.values()
    elif isinstance(local, list):
        to_combine = local
    else:
        return combined
 
    # Recurse first so that nested nodes are resolved
    for sub in to_combine:
        combined = combineConfigFiles(sub, config_path, fragment_key=fragment_key) or combined
 
    # if there are no fragments to include we're done
    if fragment_key not in local:
        return combined
 
    # Only dict nodes can have include keys
    if not isinstance(local, dict):
        return combined
 
    # Normalize to a list of paths
    value = local[fragment_key]
    if isinstance(value, (str, pathlib.Path)):
        warnings.warn(
            f"{fragment_key} should be followed with a list of files", 
            TextConfigWarning,
            stacklevel=2,
        )
        paths = [value]
    elif isinstance(value, list):
        paths = value
    else:
        raise TypeError(f"'{fragment_key}' must be a string path or a list of paths, got {type(value).__name__}")
 
    # Build an accumulator of all fragments, earlier paths win on conflicts
    fragments_acc = {}
    for entry in paths:
        fragment_path = _find_fragment(pathlib.Path(entry), config_path)
        fragment = _load_fragment(fragment_path)
 
        # Allow recursion inside each fragment, using the fragment's directory as base
        combineConfigFiles(fragment, fragment_path.parent, fragment_key=fragment_key)
 
        # Merge this fragment into the accumulator; earlier entries win
        _merge_dicts(fragments_acc, fragment)
 
    # Remove the key before merging to avoid re-processing it
    del local[fragment_key]
 
    # Merge fragments into local; local values take precedence
    _merge_dicts(local, fragments_acc)
 
    return True
 
 

makeSequence()

def python.ConfigText.makeSequence	(		configPath,
		*	flags = None,
			algSeq = None,
			noSystematics = None,
			dataType = None,
			geometry = None,
			autoconfigFromFlags = None,
			isPhyslite = None,
			noPhysliteBroken = False
	)

Definition at line 321 of file ConfigText.py.

def makeSequence(configPath, *, flags=None, algSeq=None, noSystematics=None, dataType=None, geometry=None, autoconfigFromFlags=None, isPhyslite=None, noPhysliteBroken=False):
    """
    """
 
    # Historically we have used the identifier
    # `autoconfigFromFlags`, but in the rest of the code base
    # `flags` is used. So for now we allow either, and can hopefully
    # at some point remove the former (21 Aug 25).
    if autoconfigFromFlags is not None:
        if flags is not None:
            raise ValueError("Cannot pass both flags and autoconfigFromFlags arguments")
        flags = autoconfigFromFlags
        warnings.warn ('Using autoconfigFromFlags parameter is deprecated, use flags instead', category=deprecationWarningCategory, stacklevel=2)
    elif flags is None:
        warnings.warn ('it is deprecated to configure meta-data for analysis configuration manually, please read the configuration flags via the meta-data reader', category=deprecationWarningCategory, stacklevel=2)
 
    from AnalysisAlgorithmsConfig.ConfigAccumulator import ConfigAccumulator
 
    config = TextConfig(configPath)
 
    logCPAlgTextCfg.info("Configuration file read in:")
    config.printConfig()
 
    logCPAlgTextCfg.info("Default algorithms:")
    config.printAlgs(printOpts=True)
 
    logCPAlgTextCfg.info("Configuring algorithms based on YAML file:")
    configSeq = config.configure()
 
    # defaults are added to config as algs are configured
    logCPAlgTextCfg.info("Configuration used:")
    config.printConfig()
 
    # compile
    configAccumulator = ConfigAccumulator(algSeq=algSeq, dataType=dataType, isPhyslite=isPhyslite, geometry=geometry, autoconfigFromFlags=autoconfigFromFlags, flags=flags, noSystematics=noSystematics)
    configSeq.fullConfigure(configAccumulator)
 
    # blocks can be reordered during configSeq.fullConfigure
    logCPAlgTextCfg.info("ConfigBlocks and their configuration:")
    configSeq.printOptions()
 
    from AnaAlgorithm.DualUseConfig import isAthena, useComponentAccumulator
    if isAthena and useComponentAccumulator:
        return configAccumulator.CA
    else:
        return None
 
 
# Combine configuration files
#
# See the README for more info on how this works
#

printYaml()

def python.ConfigText.printYaml	(		d,
			sort = False,
			jsonFormat = False
	)

Prints a dictionary as YAML

Definition at line 30 of file ConfigText.py.

def printYaml(d, sort=False, jsonFormat=False):
    """Prints a dictionary as YAML"""
    print(yaml.dump(d, default_flow_style=jsonFormat, sort_keys=sort))
 

readYaml()

def python.ConfigText.readYaml

(

yamlPath

)

Loads YAML file into a dictionary

Definition at line 21 of file ConfigText.py.

def readYaml(yamlPath):
    """Loads YAML file into a dictionary"""
    if not os.path.isfile(yamlPath):
        raise ValueError(f"{yamlPath} is not a file.")
    with open(yamlPath, 'r') as f:
        textConfig = yaml.safe_load(f)
    return textConfig
 
 

Variable Documentation

logCPAlgTextCfg

python.ConfigText.logCPAlgTextCfg = logging.getLogger('CPAlgTextCfg')

Definition at line 18 of file ConfigText.py.