python.ConfigText Namespace Reference

Classes
class	TextConfig
class	TextConfigWarning

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

Variables
	logCPAlgTextCfg = logging.getLogger('CPAlgTextCfg')

Function Documentation

_find_fragment()

python.ConfigText._find_fragment ( fragment_path, config_path )

protected

Definition at line 467 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()

python.ConfigText._load_fragment ( pathlib.Path fragment_path )

protected

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 450 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()

python.ConfigText._merge_dicts ( local, fragment )

protected

Definition at line 480 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()

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 381 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()

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

Definition at line 333 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()
 
    return configAccumulator.CA if isAthena else None
 
 
# Combine configuration files
#
# See the README for more info on how this works
#

printYaml()

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

Prints a dictionary as YAML

Definition at line 31 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()

python.ConfigText.readYaml

(

yamlPath

)

Loads YAML file into a dictionary

Definition at line 22 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 19 of file ConfigText.py.