lsst.pipe.base.pipeline.TaskDatasetTypes Class Reference

Public Member Functions
TaskDatasetTypes	fromTaskDef (cls, TaskDef taskDef, *Registry registry)

Detailed Description

An immutable struct that extracts and classifies the dataset types used
by a `PipelineTask`

Definition at line 605 of file pipeline.py.

Member Function Documentation

fromTaskDef()

TaskDatasetTypes lsst.pipe.base.pipeline.TaskDatasetTypes.fromTaskDef	(		cls,
		TaskDef	taskDef,
		*Registry	registry
	)

Extract and classify the dataset types from a single `PipelineTask`.

Parameters
----------
taskDef: `TaskDef`
    An instance of a `TaskDef` class for a particular `PipelineTask`.
registry: `Registry`
    Registry used to construct normalized `DatasetType` objects and
    retrieve those that are incomplete.

Returns
-------
types: `TaskDatasetTypes`
    The dataset types used by this task.

Definition at line 658 of file pipeline.py.

    def fromTaskDef(cls, taskDef: TaskDef, *, registry: Registry) -> TaskDatasetTypes:
        """Extract and classify the dataset types from a single `PipelineTask`.
 
        Parameters
        ----------
        taskDef: `TaskDef`
            An instance of a `TaskDef` class for a particular `PipelineTask`.
        registry: `Registry`
            Registry used to construct normalized `DatasetType` objects and
            retrieve those that are incomplete.
 
        Returns
        -------
        types: `TaskDatasetTypes`
            The dataset types used by this task.
        """
        def makeDatasetTypesSet(connectionType: str, freeze: bool = True) -> NamedValueSet[DatasetType]:
            """Constructs a set of true `DatasetType` objects
 
            Parameters
            ----------
            connectionType : `str`
                Name of the connection type to produce a set for, corresponds
                to an attribute of type `list` on the connection class instance
            freeze : `bool`, optional
                If `True`, call `NamedValueSet.freeze` on the object returned.
 
            Returns
            -------
            datasetTypes : `NamedValueSet`
                A set of all datasetTypes which correspond to the input
                connection type specified in the connection class of this
                `PipelineTask`
 
            Notes
            -----
            This function is a closure over the variables ``registry`` and
            ``taskDef``.
            """
            datasetTypes = NamedValueSet()
            for c in iterConnections(taskDef.connections, connectionType):
                dimensions = set(getattr(c, 'dimensions', set()))
                if "skypix" in dimensions:
                    try:
                        datasetType = registry.getDatasetType(c.name)
                    except LookupError as err:
                        raise LookupError(
                            f"DatasetType '{c.name}' referenced by "
                            f"{type(taskDef.connections).__name__} uses 'skypix' as a dimension "
                            f"placeholder, but does not already exist in the registry.  "
                            f"Note that reference catalog names are now used as the dataset "
                            f"type name instead of 'ref_cat'."
                        ) from err
                    rest1 = set(registry.dimensions.extract(dimensions - set(["skypix"])).names)
                    rest2 = set(dim.name for dim in datasetType.dimensions
                                if not isinstance(dim, SkyPixDimension))
                    if rest1 != rest2:
                        raise ValueError(f"Non-skypix dimensions for dataset type {c.name} declared in "
                                         f"connections ({rest1}) are inconsistent with those in "
                                         f"registry's version of this dataset ({rest2}).")
                else:
                    # Component dataset types are not explicitly in the
                    # registry.  This complicates consistency checks with
                    # registry and requires we work out the composite storage
                    # class.
                    registryDatasetType = None
                    try:
                        registryDatasetType = registry.getDatasetType(c.name)
                    except KeyError:
                        compositeName, componentName = DatasetType.splitDatasetTypeName(c.name)
                        parentStorageClass = DatasetType.PlaceholderParentStorageClass \
                            if componentName else None
                        datasetType = c.makeDatasetType(
                            registry.dimensions,
                            parentStorageClass=parentStorageClass
                        )
                        registryDatasetType = datasetType
                    else:
                        datasetType = c.makeDatasetType(
                            registry.dimensions,
                            parentStorageClass=registryDatasetType.parentStorageClass
                        )
 
                    if registryDatasetType and datasetType != registryDatasetType:
                        raise ValueError(f"Supplied dataset type ({datasetType}) inconsistent with "
                                         f"registry definition ({registryDatasetType}) "
                                         f"for {taskDef.label}.")
                datasetTypes.add(datasetType)
            if freeze:
                datasetTypes.freeze()
            return datasetTypes
 
        # optionally add output dataset for metadata
        outputs = makeDatasetTypesSet("outputs", freeze=False)
        if taskDef.metadataDatasetName is not None:
            # Metadata is supposed to be of the PropertySet type, its
            # dimensions correspond to a task quantum
            dimensions = registry.dimensions.extract(taskDef.connections.dimensions)
            outputs |= {DatasetType(taskDef.metadataDatasetName, dimensions, "PropertySet")}
        outputs.freeze()
 
        return cls(
            initInputs=makeDatasetTypesSet("initInputs"),
            initOutputs=makeDatasetTypesSet("initOutputs"),
            inputs=makeDatasetTypesSet("inputs"),
            prerequisites=makeDatasetTypesSet("prerequisiteInputs"),
            outputs=outputs,
        )
 
 
@dataclass(frozen=True)

---

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