lsst.pipe.base.pipeline.PipelineDatasetTypes Class Reference

Public Member Functions
PipelineDatasetTypes	fromPipeline (cls, pipeline, *Registry registry)

Detailed Description

An immutable struct that classifies the dataset types used in a
`Pipeline`.

Definition at line 769 of file pipeline.py.

Member Function Documentation

fromPipeline()

PipelineDatasetTypes lsst.pipe.base.pipeline.PipelineDatasetTypes.fromPipeline	(		cls,
			pipeline,
		*Registry	registry
	)

Extract and classify the dataset types from all tasks in a
`Pipeline`.

Parameters
----------
pipeline: `Pipeline`
    An ordered collection of tasks that can be run together.
registry: `Registry`
    Registry used to construct normalized `DatasetType` objects and
    retrieve those that are incomplete.

Returns
-------
types: `PipelineDatasetTypes`
    The dataset types used by this `Pipeline`.

Raises
------
ValueError
    Raised if Tasks are inconsistent about which datasets are marked
    prerequisite.  This indicates that the Tasks cannot be run as part
    of the same `Pipeline`.

Definition at line 835 of file pipeline.py.

    def fromPipeline(cls, pipeline, *, registry: Registry) -> PipelineDatasetTypes:
        """Extract and classify the dataset types from all tasks in a
        `Pipeline`.
 
        Parameters
        ----------
        pipeline: `Pipeline`
            An ordered collection of tasks that can be run together.
        registry: `Registry`
            Registry used to construct normalized `DatasetType` objects and
            retrieve those that are incomplete.
 
        Returns
        -------
        types: `PipelineDatasetTypes`
            The dataset types used by this `Pipeline`.
 
        Raises
        ------
        ValueError
            Raised if Tasks are inconsistent about which datasets are marked
            prerequisite.  This indicates that the Tasks cannot be run as part
            of the same `Pipeline`.
        """
        allInputs = NamedValueSet()
        allOutputs = NamedValueSet()
        allInitInputs = NamedValueSet()
        allInitOutputs = NamedValueSet()
        prerequisites = NamedValueSet()
        byTask = dict()
        if isinstance(pipeline, Pipeline):
            pipeline = pipeline.toExpandedPipeline()
        for taskDef in pipeline:
            thisTask = TaskDatasetTypes.fromTaskDef(taskDef, registry=registry)
            allInitInputs |= thisTask.initInputs
            allInitOutputs |= thisTask.initOutputs
            allInputs |= thisTask.inputs
            prerequisites |= thisTask.prerequisites
            allOutputs |= thisTask.outputs
            byTask[taskDef.label] = thisTask
        if not prerequisites.isdisjoint(allInputs):
            raise ValueError("{} marked as both prerequisites and regular inputs".format(
                {dt.name for dt in allInputs & prerequisites}
            ))
        if not prerequisites.isdisjoint(allOutputs):
            raise ValueError("{} marked as both prerequisites and outputs".format(
                {dt.name for dt in allOutputs & prerequisites}
            ))
        # Make sure that components which are marked as inputs get treated as
        # intermediates if there is an output which produces the composite
        # containing the component
        intermediateComponents = NamedValueSet()
        intermediateComposites = NamedValueSet()
        outputNameMapping = {dsType.name: dsType for dsType in allOutputs}
        for dsType in allInputs:
            # get the name of a possible component
            name, component = dsType.nameAndComponent()
            # if there is a component name, that means this is a component
            # DatasetType, if there is an output which produces the parent of
            # this component, treat this input as an intermediate
            if component is not None:
                if name in outputNameMapping:
                    if outputNameMapping[name].dimensions != dsType.dimensions:
                        raise ValueError(f"Component dataset type {dsType.name} has different "
                                         f"dimensions ({dsType.dimensions}) than its parent "
                                         f"({outputNameMapping[name].dimensions}).")
                    composite = DatasetType(name, dsType.dimensions, outputNameMapping[name].storageClass,
                                            universe=registry.dimensions)
                    intermediateComponents.add(dsType)
                    intermediateComposites.add(composite)
 
        def checkConsistency(a: NamedValueSet, b: NamedValueSet):
            common = a.names & b.names
            for name in common:
                if a[name] != b[name]:
                    raise ValueError(f"Conflicting definitions for dataset type: {a[name]} != {b[name]}.")
 
        checkConsistency(allInitInputs, allInitOutputs)
        checkConsistency(allInputs, allOutputs)
        checkConsistency(allInputs, intermediateComposites)
        checkConsistency(allOutputs, intermediateComposites)
 
        def frozen(s: NamedValueSet) -> NamedValueSet:
            s.freeze()
            return s
 
        return cls(
            initInputs=frozen(allInitInputs - allInitOutputs),
            initIntermediates=frozen(allInitInputs & allInitOutputs),
            initOutputs=frozen(allInitOutputs - allInitInputs),
            inputs=frozen(allInputs - allOutputs - intermediateComponents),
            intermediates=frozen(allInputs & allOutputs | intermediateComponents),
            outputs=frozen(allOutputs - allInputs - intermediateComposites),
            prerequisites=frozen(prerequisites),
            byTask=MappingProxyType(byTask),  # MappingProxyType -> frozen view of dict for immutability
        )

---

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