lsst.pipe.base.connections.PipelineTaskConnectionsMetaclass Class Reference

Inheritance diagram for lsst.pipe.base.connections.PipelineTaskConnectionsMetaclass:

Public Member Functions
def	__prepare__ (name, bases, **kwargs)
def	__new__ (cls, name, bases, dct, **kwargs)
def	__init__ (cls, name, bases, dct, **kwargs)

Detailed Description

Metaclass used in the declaration of PipelineTaskConnections classes

Definition at line 96 of file connections.py.

Constructor & Destructor Documentation

__init__()

def lsst.pipe.base.connections.PipelineTaskConnectionsMetaclass.__init__	(		cls,
			name,
			bases,
			dct,
		**	kwargs
	)

Definition at line 187 of file connections.py.

    def __init__(cls, name, bases, dct, **kwargs):
        # This overrides the default init to drop the kwargs argument. Python
        # metaclasses will have this argument set if any kwargs are passes at
        # class construction time, but should be consumed before calling
        # __init__ on the type metaclass. This is in accordance with python
        # documentation on metaclasses
        super().__init__(name, bases, dct)
 
 

Member Function Documentation

__new__()

def lsst.pipe.base.connections.PipelineTaskConnectionsMetaclass.__new__	(		cls,
			name,
			bases,
			dct,
		**	kwargs
	)

Definition at line 110 of file connections.py.

    def __new__(cls, name, bases, dct, **kwargs):
        dimensionsValueError = TypeError("PipelineTaskConnections class must be created with a dimensions "
                                         "attribute which is an iterable of dimension names")
 
        if name != 'PipelineTaskConnections':
            # Verify that dimensions are passed as a keyword in class
            # declaration
            if 'dimensions' not in kwargs:
                for base in bases:
                    if hasattr(base, 'dimensions'):
                        kwargs['dimensions'] = base.dimensions
                        break
                if 'dimensions' not in kwargs:
                    raise dimensionsValueError
            try:
                if isinstance(kwargs['dimensions'], str):
                    raise TypeError("Dimensions must be iterable of dimensions, got str,"
                                    "possibly omitted trailing comma")
                if not isinstance(kwargs['dimensions'], typing.Iterable):
                    raise TypeError("Dimensions must be iterable of dimensions")
                dct['dimensions'] = set(kwargs['dimensions'])
            except TypeError as exc:
                raise dimensionsValueError from exc
            # Lookup any python string templates that may have been used in the
            # declaration of the name field of a class connection attribute
            allTemplates = set()
            stringFormatter = string.Formatter()
            # Loop over all connections
            for obj in dct['allConnections'].values():
                nameValue = obj.name
                # add all the parameters to the set of templates
                for param in stringFormatter.parse(nameValue):
                    if param[1] is not None:
                        allTemplates.add(param[1])
 
            # look up any template from base classes and merge them all
            # together
            mergeDict = {}
            for base in bases[::-1]:
                if hasattr(base, 'defaultTemplates'):
                    mergeDict.update(base.defaultTemplates)
            if 'defaultTemplates' in kwargs:
                mergeDict.update(kwargs['defaultTemplates'])
 
            if len(mergeDict) > 0:
                kwargs['defaultTemplates'] = mergeDict
 
            # Verify that if templated strings were used, defaults were
            # supplied as an argument in the declaration of the connection
            # class
            if len(allTemplates) > 0 and 'defaultTemplates' not in kwargs:
                raise TypeError("PipelineTaskConnection class contains templated attribute names, but no "
                                "defaut templates were provided, add a dictionary attribute named "
                                "defaultTemplates which contains the mapping between template key and value")
            if len(allTemplates) > 0:
                # Verify all templates have a default, and throw if they do not
                defaultTemplateKeys = set(kwargs['defaultTemplates'].keys())
                templateDifference = allTemplates.difference(defaultTemplateKeys)
                if templateDifference:
                    raise TypeError(f"Default template keys were not provided for {templateDifference}")
                # Verify that templates do not share names with variable names
                # used for a connection, this is needed because of how
                # templates are specified in an associated config class.
                nameTemplateIntersection = allTemplates.intersection(set(dct['allConnections'].keys()))
                if len(nameTemplateIntersection) > 0:
                    raise TypeError(f"Template parameters cannot share names with Class attributes"
                                    f" (conflicts are {nameTemplateIntersection}).")
            dct['defaultTemplates'] = kwargs.get('defaultTemplates', {})
 
        # Convert all the connection containers into frozensets so they cannot
        # be modified at the class scope
        for connectionName in ("inputs", "prerequisiteInputs", "outputs", "initInputs", "initOutputs"):
            dct[connectionName] = frozenset(dct[connectionName])
        # our custom dict type must be turned into an actual dict to be used in
        # type.__new__
        return super().__new__(cls, name, bases, dict(dct))
 

__prepare__()

def lsst.pipe.base.connections.PipelineTaskConnectionsMetaclass.__prepare__	(		name,
			bases,
		**	kwargs
	)

Definition at line 99 of file connections.py.

    def __prepare__(name, bases, **kwargs):  # noqa: 805
        # Create an instance of our special dict to catch and track all
        # variables that are instances of connectionTypes.BaseConnection
        # Copy any existing connections from a parent class
        dct = PipelineTaskConnectionDict()
        for base in bases:
            if isinstance(base, PipelineTaskConnectionsMetaclass):
                for name, value in base.allConnections.items():
                    dct[name] = value
        return dct
 

---

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/connections.py