python.ConfigText.TextConfig Class Reference

Inheritance diagram for python.ConfigText.TextConfig:

Collaboration diagram for python.ConfigText.TextConfig:

Public Member Functions
def	__init__ (self, yamlPath=None, *config=None, addDefaultBlocks=True)
def	setConfig (self, config)
def	preprocessConfig (self, config, algs)
def	cleanupPlaceholders (self, config)
def	loadConfig (self, yamlPath=None, *configDict=None)
def	printConfig (self, sort=False, jsonFormat=False)
def	saveYaml (self, filePath='config.yaml', default_flow_style=False, **kwargs)
def	addBlock (self, name, **kwargs)
def	setOptions (self, **kwargs)
def	configure (self)

Private Member Functions
def	_addNewConfigBlocks (self, modulePath, functionName, algName, defaults=None, pos=None, superBlocks=None)
def	_configureAlg (self, block, blockConfig, configSeq=None, containerName=None, extraOptions=None)

Private Attributes
	_config
	__loadedYaml
	_last

Detailed Description

Definition at line 38 of file ConfigText.py.

Constructor & Destructor Documentation

__init__()

def python.ConfigText.TextConfig.__init__	(		self,
			yamlPath = None,
		*	config = None,
			addDefaultBlocks = True
	)

Definition at line 39 of file ConfigText.py.

    def __init__(self, yamlPath=None, *, config=None, addDefaultBlocks=True):
        super().__init__(addDefaultBlocks=False)
 
        if yamlPath and config:
            raise ValueError("Cannot specify both yamlPath and config. Use one or the other.")
 
        # Block to add new blocks to this object
        self.addAlgConfigBlock(algName="AddConfigBlocks", alg=self._addNewConfigBlocks,
            defaults={'self': self})
        # add default blocks
        if addDefaultBlocks:
            self.addDefaultAlgs()
        # load yaml
        self._config = {}
        # do not allow for loading multiple yaml files
        self.__loadedYaml = False
        if yamlPath is not None or config is not None:
            self.loadConfig(yamlPath, configDict=config)
        # last is used for setOptionValue when using addBlock
        self._last = None
 
 

Member Function Documentation

_addNewConfigBlocks()

def python.ConfigText.TextConfig._addNewConfigBlocks (   self,   modulePath,   functionName,   algName,   defaults = None,   pos = None,   superBlocks = None  )

private

Load <functionName> from <modulePath>

Definition at line 240 of file ConfigText.py.

    def _addNewConfigBlocks(self, modulePath, functionName,
        algName, defaults=None, pos=None, superBlocks=None):
        """
        Load <functionName> from <modulePath>
        """
        try:
            module = importlib.import_module(modulePath)
            fxn = getattr(module, functionName)
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError(f"{e}\nFailed to load {functionName} from {modulePath}")
        else:
            sys.modules[functionName] = fxn
        # add new algorithm to available algorithms
        self.addAlgConfigBlock(algName=algName, alg=fxn,
            defaults=defaults,
            superBlocks=superBlocks,
            pos=pos)
        return
 
 

_configureAlg()

def python.ConfigText.TextConfig._configureAlg (   self,   block,   blockConfig,   configSeq = None,   containerName = None,   extraOptions = None  )

private

Definition at line 260 of file ConfigText.py.

    def _configureAlg(self, block, blockConfig, configSeq=None, containerName=None,
                      extraOptions=None):
        if not isinstance(blockConfig, list):
            blockConfig = [blockConfig]
 
        for options in blockConfig:
            # Special case: propogate containerName down to subAlgs
            if 'containerName' in options:
                containerName = options['containerName']
            elif containerName is not None and 'containerName' not in options:
                options['containerName'] = containerName
            # will check which options are associated alg and not options
            logCPAlgTextCfg.info(f"Configuring {block.algName}")
            seq, funcOpts = block.makeConfig(options)
            if not seq._blocks:
                continue
            algOpts = seq.setOptions(options)
            # If containerName was not set explicitly, we can now retrieve
            # its default value
            if containerName is None:
                for opt in algOpts:
                    if 'name' in opt and opt['name'] == 'containerName':
                        containerName = opt.get('value', None)
                        break  # Exit the loop as we've found the key
 
            if configSeq is not None:
                configSeq += seq
 
            # propagate special extra options to subalgs
            if extraOptions is None:
                extraOptionsList = ["skipOnData", "skipOnMC", "onlyForDSIDs"]
                for i in algOpts:
                    if i['name'] in extraOptionsList and i['defaultValue'] != i['value']:
                        if extraOptions is None:
                            extraOptions = {}
                        extraOptions[i['name']] = i['value']
            else:
                algOpts = seq.setOptions(extraOptions.copy())
 
            # check to see if there are unused parameters
            algOpts = [i['name'] for i in algOpts]
            expectedOptions = set(funcOpts)
            expectedOptions |= set(algOpts)
            expectedOptions |= set(block.subAlgs)
 
            difference = set(options.keys()) - expectedOptions
            difference.discard('__placeholder__')
            if difference:
                difference = "\n".join(difference)
                raise ValueError(f"There are options set that are not used for "
                                 f"{block.algName}:\n{difference}\n"
                                 "Please check your configuration.")
 
            # check for sub-blocks and call this function recursively
            for alg in self._order.get(block.algName, []):
                if alg in options:
                    subAlg = block.subAlgs[alg]
                    self._configureAlg(subAlg, options[alg], configSeq, containerName, extraOptions)
        return configSeq
 
 

addBlock()

def python.ConfigText.TextConfig.addBlock	(		self,
			name,
		**	kwargs
	)

Create entry into dictionary representing the text configuration

Definition at line 177 of file ConfigText.py.

    def addBlock(self, name, **kwargs):
        """
        Create entry into dictionary representing the text configuration
        """
        def setEntry(name, config, opts):
            if '.' not in name:
                if name not in config:
                    config[name] = opts
                elif isinstance(config[name], list):
                    config[name].append(opts)
                else:
                    config[name] = [config[name], opts]
                # set last added block for setOptionValue
                self._last = opts
            else:
                name, rest = name[:name.index('.')], name[name.index('.') + 1:]
                config = config[name]
                if isinstance(config, list):
                    config = config[-1]
                setEntry(rest, config, opts)
            return
        setEntry(name, self._config, dict(kwargs))
        return
 
 

cleanupPlaceholders()

def python.ConfigText.TextConfig.cleanupPlaceholders	(		self,
			config
	)

Remove placeholder markers after initialization.

Definition at line 93 of file ConfigText.py.

    def cleanupPlaceholders(self, config):
        """
        Remove placeholder markers after initialization.
        """
        if not isinstance(config, dict):
            return
        if "__placeholder__" in config:
            del config["__placeholder__"]
        for key, value in config.items():
            self.cleanupPlaceholders(value)
 

configure()

def python.ConfigText.TextConfig.configure

(

self

)

Process YAML configuration file and confgure added algorithms.

Definition at line 213 of file ConfigText.py.

    def configure(self):
        """Process YAML configuration file and confgure added algorithms."""
        # make sure all blocks in yaml file are added (otherwise they would be ignored)
        for blockName in self._config:
            if blockName not in self._order[self.ROOTNAME]:
                if not blockName:
                    blockName = list(self._config[blockName].keys())[0]
                raise ValueError(f"Unkown block {blockName} in yaml file")
 
        # configure blocks
        configSeq = ConfigSequence()
        for blockName in self._order[self.ROOTNAME]:
            if blockName == "AddConfigBlocks":
                continue
 
            assert blockName in self._algs
 
            # order only applies to root blocks
            if blockName in self._config:
                blockConfig = self._config[blockName]
                alg = self._algs[blockName]
                self._configureAlg(alg, blockConfig, configSeq)
            else:
                continue
        return configSeq
 
 

loadConfig()

def python.ConfigText.TextConfig.loadConfig	(		self,
			yamlPath = None,
		*	configDict = None
	)

read a YAML file. Will combine with any config blocks added using python

Definition at line 104 of file ConfigText.py.

    def loadConfig(self, yamlPath=None, *, configDict=None):
        """
        read a YAML file. Will combine with any config blocks added using python
        """
        if self.__loadedYaml or isinstance(yamlPath, list):
            raise NotImplementedError("Mering multiple yaml files is not implemented.")
        self.__loadedYaml = True
 
        def merge(config, algs, path=''):
            """Add to config block-by-block"""
            if not isinstance(config, list):
                config = [config]
            # loop over list of blocks with same block name
            for blocks in config:
                # deal with case where empty dict is config
                if blocks == {} and path:
                    self.addBlock(path)
                    return
                # remove any subBlocks from block config
                subBlocks = {}
                for blockName in algs:
                    if blockName in blocks:
                        subBlocks[blockName] = blocks.pop(blockName)
                # anything left should be a block and it's configuration
                if blocks:
                    self.addBlock(path, **blocks)
                # add in any subBlocks
                for subName, subBlock in subBlocks.items():
                    newPath = f'{path}.{subName}' if path else subName
                    merge(subBlock, algs[subName].subAlgs, newPath)
            return
 
        logCPAlgTextCfg.info(f'loading {yamlPath}')
        if configDict is not None:
            # if configDict is provided, use it directly
            config = configDict
        else:
            config = readYaml(yamlPath)
        # check if blocks are defined in yaml file
        if "AddConfigBlocks" in config:
           self._configureAlg(self._algs["AddConfigBlocks"], config["AddConfigBlocks"])
 
        # Preprocess the configuration dictionary (see !76767)
        self.preprocessConfig(config, self._algs)
 
        merge(config, self._algs)
 
        # Cleanup placeholders (see !76767)
        self.cleanupPlaceholders(config)
 
        return
 
 

preprocessConfig()

def python.ConfigText.TextConfig.preprocessConfig	(		self,
			config,
			algs
	)

Preprocess the configuration dictionary.
Ensure blocks with only sub-blocks are initialized with an empty dictionary.

Definition at line 69 of file ConfigText.py.

    def preprocessConfig(self, config, algs):
        """
        Preprocess the configuration dictionary.
        Ensure blocks with only sub-blocks are initialized with an empty dictionary.
        """
        def processNode(node, algs):
            if not isinstance(node, dict):
                return  # Base case: not a dictionary
            for blockName, blockContent in list(node.items()):
                # If the block name is recognized in algs
                if blockName in algs:
                    # If the block only defines sub-blocks, initialize it
                    if isinstance(blockContent, dict) and not any(
                            key in algs[blockName].options for key in blockContent
                    ):
                        # Ensure parent block is initialized as an empty dictionary
                        node[blockName] = {'__placeholder__': True, **blockContent}
                    # Recurse into sub-blocks
                    processNode(node[blockName], algs[blockName].subAlgs)
 
        # Start processing from the root of the configuration
        processNode(config, algs)
 

printConfig()

def python.ConfigText.TextConfig.printConfig	(		self,
			sort = False,
			jsonFormat = False
	)

Print YAML configuration file.

Definition at line 157 of file ConfigText.py.

    def printConfig(self, sort=False, jsonFormat=False):
        """Print YAML configuration file."""
        if self._config is None:
            raise ValueError("No configuration has been loaded.")
        printYaml(self._config, sort, jsonFormat)
        return
 
 

saveYaml()

def python.ConfigText.TextConfig.saveYaml	(		self,
			filePath = 'config.yaml',
			default_flow_style = False,
		**	kwargs
	)

Convert dictionary representation to yaml and save

Definition at line 165 of file ConfigText.py.

    def saveYaml(self, filePath='config.yaml', default_flow_style=False,
            **kwargs):
        """
        Convert dictionary representation to yaml and save
        """
        logCPAlgTextCfg.info(f"Saving configuration to {filePath}")
        config = self._config
        with open(filePath, 'w') as outfile:
            yaml.dump(config, outfile, default_flow_style=False, **kwargs)
        return
 
 

setConfig()

def python.ConfigText.TextConfig.setConfig	(		self,
			config
	)

Print YAML configuration file.

Definition at line 61 of file ConfigText.py.

    def setConfig(self, config):
        """Print YAML configuration file."""
        if self._config:
            raise ValueError("Configuration has already been loaded.")
        self._config = config
        return
 

setOptions()

def python.ConfigText.TextConfig.setOptions	(		self,
		**	kwargs
	)

Set option(s) for the lsat block that was added. If an option
was added previously, will update value

Definition at line 202 of file ConfigText.py.

    def setOptions(self, **kwargs):
        """
        Set option(s) for the lsat block that was added. If an option
        was added previously, will update value
        """
        if self._last is None:
            raise TypeError("Cannot set options before adding a block")
        # points to dict with opts for last added block
        self._last.update(**kwargs)
 
 

Member Data Documentation

__loadedYaml

python.ConfigText.TextConfig.__loadedYaml

private

Definition at line 54 of file ConfigText.py.

_config

python.ConfigText.TextConfig._config

private

Definition at line 52 of file ConfigText.py.

_last

python.ConfigText.TextConfig._last

private

Definition at line 58 of file ConfigText.py.

---

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

• ConfigText.py