lsst.pipe.base.pipelineIR.PipelineIR Class Reference

Public Member Functions
def	__init__ (self, loaded_yaml)
PipelineIR	subset_from_labels (self, Set[str] labelSpecifier)
def	from_string (cls, str pipeline_string)
PipelineIR	from_file (cls, str filename)
PipelineIR	from_uri (cls, Union[str, ButlerURI] uri)
def	to_file (self, str filename)
def	write_to_uri (self, Union[ButlerURI, str] uri)
Dict[str, Any]	to_primitives (self)
str	__str__ (self)
str	__repr__ (self)
def	__eq__ (self, object other)

Public Attributes
	description
	instrument
	contracts
	parameters
	labeled_subsets
	imports
	tasks

Detailed Description

Intermediate representation of a pipeline definition

Parameters
----------
loaded_yaml : `dict`
    A dictionary which matches the structure that would be produced by a
    yaml reader which parses a pipeline definition document

Raises
------
ValueError :
    - If a pipeline is declared without a description
    - If no tasks are declared in a pipeline, and no pipelines are to be
      inherited
    - If more than one instrument is specified
    - If more than one inherited pipeline share a label

Definition at line 459 of file pipelineIR.py.

Constructor & Destructor Documentation

__init__()

def lsst.pipe.base.pipelineIR.PipelineIR.__init__	(		self,
			loaded_yaml
	)

Definition at line 477 of file pipelineIR.py.

    def __init__(self, loaded_yaml):
        # Check required fields are present
        if "description" not in loaded_yaml:
            raise ValueError("A pipeline must be declared with a description")
        if "tasks" not in loaded_yaml and len({"imports", "inherits"} - loaded_yaml.keys()) == 2:
            raise ValueError("A pipeline must be declared with one or more tasks")
 
        # These steps below must happen in this call order
 
        # Process pipeline description
        self.description = loaded_yaml.pop("description")
 
        # Process tasks
        self._read_tasks(loaded_yaml)
 
        # Process instrument keys
        inst = loaded_yaml.pop("instrument", None)
        if isinstance(inst, list):
            raise ValueError("Only one top level instrument can be defined in a pipeline")
        self.instrument = inst
 
        # Process any contracts
        self._read_contracts(loaded_yaml)
 
        # Process any defined parameters
        self._read_parameters(loaded_yaml)
 
        # Process any named label subsets
        self._read_labeled_subsets(loaded_yaml)
 
        # Process any inherited pipelines
        self._read_imports(loaded_yaml)
 
        # verify named subsets, must be done after inheriting
        self._verify_labeled_subsets()
 

Member Function Documentation

__eq__()

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

Definition at line 895 of file pipelineIR.py.

        return str(self)
 
    def __eq__(self, other: object):
        if not isinstance(other, PipelineIR):
            return False
        elif all(getattr(self, attr) == getattr(other, attr) for attr in
                 ("contracts", "tasks", "instrument")):
            return True
        else:

__repr__()

str lsst.pipe.base.pipelineIR.PipelineIR.__repr__

(

self

)

Instance formatting as how it would look in yaml representation

Definition at line 890 of file pipelineIR.py.

        return yaml.dump(self.to_primitives(), sort_keys=False)
 
    def __repr__(self) -> str:
        """Instance formatting as how it would look in yaml representation
        """

__str__()

str lsst.pipe.base.pipelineIR.PipelineIR.__str__

(

self

)

Instance formatting as how it would look in yaml representation

Definition at line 885 of file pipelineIR.py.

        return accumulate
 
    def __str__(self) -> str:
        """Instance formatting as how it would look in yaml representation
        """

from_file()

PipelineIR lsst.pipe.base.pipelineIR.PipelineIR.from_file	(		cls,
		str	filename
	)

Create a `PipelineIR` object from the document specified by the
input path.

Parameters
----------
filename : `str`
    Location of document to use in creating a `PipelineIR` object.

Returns
-------
pipelineIR : `PipelineIR`
    The loaded pipeline

Note
----
This method is deprecated, please use from_uri

Definition at line 798 of file pipelineIR.py.

                version="v21.0,", category=FutureWarning)  # type: ignore
    def from_file(cls, filename: str) -> PipelineIR:
        """Create a `PipelineIR` object from the document specified by the
        input path.
 
        Parameters
        ----------
        filename : `str`
            Location of document to use in creating a `PipelineIR` object.
 
        Returns
        -------
        pipelineIR : `PipelineIR`
            The loaded pipeline
 
        Note
        ----
        This method is deprecated, please use from_uri
        """
        return cls.from_uri(filename)

from_string()

def lsst.pipe.base.pipelineIR.PipelineIR.from_string	(		cls,
		str	pipeline_string
	)

Create a `PipelineIR` object from a string formatted like a pipeline
document

Parameters
----------
pipeline_string : `str`
    A string that is formatted according like a pipeline document

Definition at line 784 of file pipelineIR.py.

    def from_string(cls, pipeline_string: str):
        """Create a `PipelineIR` object from a string formatted like a pipeline
        document
 
        Parameters
        ----------
        pipeline_string : `str`
            A string that is formatted according like a pipeline document
        """
        loaded_yaml = yaml.load(pipeline_string, Loader=PipelineYamlLoader)
        return cls(loaded_yaml)
 

from_uri()

PipelineIR lsst.pipe.base.pipelineIR.PipelineIR.from_uri	(		cls,
		Union[str, ButlerURI]	uri
	)

Create a `PipelineIR` object from the document specified by the
input uri.

Parameters
----------
uri: `str` or `ButlerURI`
    Location of document to use in creating a `PipelineIR` object.

Returns
-------
pipelineIR : `PipelineIR`
    The loaded pipeline

Definition at line 819 of file pipelineIR.py.

    @classmethod
    def from_uri(cls, uri: Union[str, ButlerURI]) -> PipelineIR:
        """Create a `PipelineIR` object from the document specified by the
        input uri.
 
        Parameters
        ----------
        uri: `str` or `ButlerURI`
            Location of document to use in creating a `PipelineIR` object.
 
        Returns
        -------
        pipelineIR : `PipelineIR`
            The loaded pipeline
        """
        loaded_uri = ButlerURI(uri)
        # With ButlerURI we have the choice of always using a local file or
        # reading in the bytes directly. Reading in bytes can be more
        # efficient for reasonably-sized files when the resource is remote.
        # For now use the local file variant. For a local file as_local() does
        # nothing.
        with loaded_uri.as_local() as local:
            # explicitly read here, there was some issue with yaml trying
            # to read the ButlerURI itself (I think because it only
            # pretends to be conformant to the io api)
            loaded_yaml = yaml.load(local.read(), Loader=PipelineYamlLoader)
            return cls(loaded_yaml)

subset_from_labels()

PipelineIR lsst.pipe.base.pipelineIR.PipelineIR.subset_from_labels	(		self,
		Set[str]	labelSpecifier
	)

Subset a pipelineIR to contain only labels specified in
labelSpecifier.

Parameters
----------
labelSpecifier : `set` of `str`
    Set containing labels that describes how to subset a pipeline.

Returns
-------
pipeline : `PipelineIR`
    A new pipelineIR object that is a subset of the old pipelineIR

Raises
------
ValueError
    Raised if there is an issue with specified labels

Notes
-----
This method attempts to prune any contracts that contain labels which
are not in the declared subset of labels. This pruning is done using a
string based matching due to the nature of contracts and may prune more
than it should. Any labeled subsets defined that no longer have all
members of the subset present in the pipeline will be removed from the
resulting pipeline.

Definition at line 718 of file pipelineIR.py.

    def subset_from_labels(self, labelSpecifier: Set[str]) -> PipelineIR:
        """Subset a pipelineIR to contain only labels specified in
        labelSpecifier.
 
        Parameters
        ----------
        labelSpecifier : `set` of `str`
            Set containing labels that describes how to subset a pipeline.
 
        Returns
        -------
        pipeline : `PipelineIR`
            A new pipelineIR object that is a subset of the old pipelineIR
 
        Raises
        ------
        ValueError
            Raised if there is an issue with specified labels
 
        Notes
        -----
        This method attempts to prune any contracts that contain labels which
        are not in the declared subset of labels. This pruning is done using a
        string based matching due to the nature of contracts and may prune more
        than it should. Any labeled subsets defined that no longer have all
        members of the subset present in the pipeline will be removed from the
        resulting pipeline.
        """
 
        pipeline = copy.deepcopy(self)
 
        # update the label specifier to expand any named subsets
        toRemove = set()
        toAdd = set()
        for label in labelSpecifier:
            if label in pipeline.labeled_subsets:
                toRemove.add(label)
                toAdd.update(pipeline.labeled_subsets[label].subset)
        labelSpecifier.difference_update(toRemove)
        labelSpecifier.update(toAdd)
        # verify all the labels are in the pipeline
        if not labelSpecifier.issubset(pipeline.tasks.keys()
                                       | pipeline.labeled_subsets):
            difference = labelSpecifier.difference(pipeline.tasks.keys())
            raise ValueError("Not all supplied labels (specified or named subsets) are in the pipeline "
                             f"definition, extra labels: {difference}")
        # copy needed so as to not modify while iterating
        pipeline_labels = set(pipeline.tasks.keys())
        # Remove the labels from the pipelineIR, and any contracts that contain
        # those labels (see docstring on _remove_contracts for why this may
        # cause issues)
        for label in pipeline_labels:
            if label not in labelSpecifier:
                pipeline.tasks.pop(label)
                pipeline._remove_contracts(label)
 
        # create a copy of the object to iterate over
        labeled_subsets = copy.copy(pipeline.labeled_subsets)
        # remove any labeled subsets that no longer have a complete set
        for label, labeled_subset in labeled_subsets.items():
            if labeled_subset.subset - pipeline.tasks.keys():
                pipeline.labeled_subsets.pop(label)
 
        return pipeline
 

to_file()

def lsst.pipe.base.pipelineIR.PipelineIR.to_file	(		self,
		str	filename
	)

Serialize this `PipelineIR` object into a yaml formatted string and
write the output to a file at the specified path.

Parameters
----------
filename : `str`
    Location of document to write a `PipelineIR` object.

Definition at line 847 of file pipelineIR.py.

    @deprecated(reason="This has been replaced with `write_to_uri`. will be removed after v23",
                version="v21.0,", category=FutureWarning)  # type: ignore
    def to_file(self, filename: str):
        """Serialize this `PipelineIR` object into a yaml formatted string and
        write the output to a file at the specified path.
 
        Parameters
        ----------
        filename : `str`
            Location of document to write a `PipelineIR` object.
        """
 

to_primitives()

Dict[str, Any] lsst.pipe.base.pipelineIR.PipelineIR.to_primitives

(

self

)

Convert to a representation used in yaml serialization

Definition at line 870 of file pipelineIR.py.

        butlerUri.write(yaml.dump(self.to_primitives(), sort_keys=False).encode())
 
    def to_primitives(self) -> Dict[str, Any]:
        """Convert to a representation used in yaml serialization
        """
        accumulate = {"description": self.description}
        if self.instrument is not None:
            accumulate['instrument'] = self.instrument
        if self.parameters:
            accumulate['parameters'] = self.parameters.to_primitives()
        accumulate['tasks'] = {m: t.to_primitives() for m, t in self.tasks.items()}
        if len(self.contracts) > 0:
            accumulate['contracts'] = [c.to_primitives() for c in self.contracts]
        if self.labeled_subsets:
            accumulate['subsets'] = {k: v.to_primitives() for k, v in self.labeled_subsets.items()}

write_to_uri()

def lsst.pipe.base.pipelineIR.PipelineIR.write_to_uri	(		self,
		Union[ButlerURI, str]	uri
	)

Serialize this `PipelineIR` object into a yaml formatted string and
write the output to a file at the specified uri.

Parameters
----------
uri: `str` or `ButlerURI`
    Location of document to write a `PipelineIR` object.

Definition at line 858 of file pipelineIR.py.

        self.write_to_uri(filename)
 
    def write_to_uri(self, uri: Union[ButlerURI, str]):
        """Serialize this `PipelineIR` object into a yaml formatted string and
        write the output to a file at the specified uri.
 
        Parameters
        ----------
        uri: `str` or `ButlerURI`
            Location of document to write a `PipelineIR` object.
        """
        butlerUri = ButlerURI(uri)

Member Data Documentation

contracts

lsst.pipe.base.pipelineIR.PipelineIR.contracts

Definition at line 525 of file pipelineIR.py.

description

lsst.pipe.base.pipelineIR.PipelineIR.description

Definition at line 487 of file pipelineIR.py.

imports

lsst.pipe.base.pipelineIR.PipelineIR.imports

Definition at line 607 of file pipelineIR.py.

instrument

lsst.pipe.base.pipelineIR.PipelineIR.instrument

Definition at line 496 of file pipelineIR.py.

labeled_subsets

lsst.pipe.base.pipelineIR.PipelineIR.labeled_subsets

Definition at line 556 of file pipelineIR.py.

parameters

lsst.pipe.base.pipelineIR.PipelineIR.parameters

Definition at line 544 of file pipelineIR.py.

tasks

lsst.pipe.base.pipelineIR.PipelineIR.tasks

Definition at line 660 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