lsst.pipe.base.pipelineIR.ConfigIR Class Reference

Public Member Functions
Dict[str, Union[str, dict, List[str]]]	to_primitives (self)
ConfigIR	formatted (self, ParametersIR parameters)
Generator["ConfigIR", None, None]	maybe_merge (self, "ConfigIR" other_config)
def	__eq__ (self, object other)

Public Attributes
	file

Static Public Attributes
	default_factory
	dict

Detailed Description

Intermediate representation of configurations read from a pipeline yaml
file.

Definition at line 211 of file pipelineIR.py.

Member Function Documentation

__eq__()

def lsst.pipe.base.pipelineIR.ConfigIR.__eq__	(		self,
		object	other
	)

Definition at line 316 of file pipelineIR.py.

    def __eq__(self, other: object):
        if not isinstance(other, ConfigIR):
            return False
        elif all(getattr(self, attr) == getattr(other, attr) for attr in
                 ("python", "dataId", "file", "rest")):
            return True
        else:
            return False
 
 
@dataclass

formatted()

ConfigIR lsst.pipe.base.pipelineIR.ConfigIR.formatted	(		self,
		ParametersIR	parameters
	)

Returns a new ConfigIR object that is formatted according to the
specified parameters

Parameters
----------
parameters : ParametersIR
    Object that contains variable mappings used in substitution.

Returns
-------
config : ConfigIR
    A new ConfigIR object formatted with the input parameters

Definition at line 250 of file pipelineIR.py.

    def formatted(self, parameters: ParametersIR) -> ConfigIR:
        """Returns a new ConfigIR object that is formatted according to the
        specified parameters
 
        Parameters
        ----------
        parameters : ParametersIR
            Object that contains variable mappings used in substitution.
 
        Returns
        -------
        config : ConfigIR
            A new ConfigIR object formatted with the input parameters
        """
        new_config = copy.deepcopy(self)
        for key, value in new_config.rest.items():
            if not isinstance(value, str):
                continue
            match = re.match("parameters[.](.*)", value)
            if match and match.group(1) in parameters:
                new_config.rest[key] = parameters[match.group(1)]
            if match and match.group(1) not in parameters:
                warnings.warn(f"config {key} contains value {match.group(0)} which is formatted like a "
                              "Pipeline parameter but was not found within the Pipeline, if this was not "
                              "intentional, check for a typo")
        return new_config
 

maybe_merge()

Generator["ConfigIR", None, None] lsst.pipe.base.pipelineIR.ConfigIR.maybe_merge	(		self,
		"ConfigIR"	other_config
	)

Merges another instance of a `ConfigIR` into this instance if
possible. This function returns a generator that is either self
if the configs were merged, or self, and other_config if that could
not be merged.

Parameters
----------
other_config : `ConfigIR`
    An instance of `ConfigIR` to merge into this instance.

Returns
-------
Generator : `ConfigIR`
    A generator containing either self, or self and other_config if
    the configs could be merged or not respectively.

Definition at line 277 of file pipelineIR.py.

    def maybe_merge(self, other_config: "ConfigIR") -> Generator["ConfigIR", None, None]:
        """Merges another instance of a `ConfigIR` into this instance if
        possible. This function returns a generator that is either self
        if the configs were merged, or self, and other_config if that could
        not be merged.
 
        Parameters
        ----------
        other_config : `ConfigIR`
            An instance of `ConfigIR` to merge into this instance.
 
        Returns
        -------
        Generator : `ConfigIR`
            A generator containing either self, or self and other_config if
            the configs could be merged or not respectively.
        """
        # Verify that the config blocks can be merged
        if self.dataId != other_config.dataId or self.python or other_config.python or\
                self.file or other_config.file:
            yield from (self, other_config)
            return
 
        # create a set of all keys, and verify two keys do not have different
        # values
        key_union = self.rest.keys() & other_config.rest.keys()
        for key in key_union:
            if self.rest[key] != other_config.rest[key]:
                yield from (self, other_config)
                return
        self.rest.update(other_config.rest)
 
        # Combine the lists of override files to load
        self_file_set = set(self.file)
        other_file_set = set(other_config.file)
        self.file = list(self_file_set.union(other_file_set))
 
        yield self
 

to_primitives()

Dict[str, Union[str, dict, List[str]]] lsst.pipe.base.pipelineIR.ConfigIR.to_primitives

(

self

)

Convert to a representation used in yaml serialization

Definition at line 236 of file pipelineIR.py.

    def to_primitives(self) -> Dict[str, Union[str, dict, List[str]]]:
        """Convert to a representation used in yaml serialization
        """
        accumulate = {}
        for name in ("python", "dataId", "file"):
            # if this attribute is thruthy add it to the accumulation
            # dictionary
            if getattr(self, name):
                accumulate[name] = getattr(self, name)
        # Add the dictionary containing the rest of the config keys to the
        # # accumulated dictionary
        accumulate.update(self.rest)
        return accumulate
 

Member Data Documentation

default_factory

lsst.pipe.base.pipelineIR.ConfigIR.default_factory

static

Definition at line 225 of file pipelineIR.py.

dict

lsst.pipe.base.pipelineIR.ConfigIR.dict

static

Definition at line 230 of file pipelineIR.py.

file

lsst.pipe.base.pipelineIR.ConfigIR.file

Definition at line 312 of file pipelineIR.py.

---

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

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-0.4.3/Linux64/pipe_base/22.0.1+94e66cc9ed/python/lsst/pipe/base/pipelineIR.py