lsst.pipe.base.graph.graph.QuantumGraph Class Reference

Public Member Functions
def	__init__ (self, Mapping[TaskDef, Set[Quantum]] quanta)
nx.DiGraph	taskGraph (self)
nx.DiGraph	graph (self)
Iterable[QuantumNode]	inputQuanta (self)
Iterable[QuantumNode]	outputQuanta (self)
Tuple[DatasetTypeName,...]	allDatasetTypes (self)
bool	isConnected (self)
QuantumNode	getQuantumNodeByNodeId (self, NodeId nodeId)
FrozenSet[Quantum]	getQuantaForTask (self, TaskDef taskDef)
FrozenSet[QuantumNode]	getNodesForTask (self, TaskDef taskDef)
Iterable[TaskDef]	findTasksWithInput (self, DatasetTypeName datasetTypeName)
Optional[TaskDef]	findTaskWithOutput (self, DatasetTypeName datasetTypeName)
Iterable[TaskDef]	tasksWithDSType (self, DatasetTypeName datasetTypeName)
List[TaskDef]	findTaskDefByName (self, str taskName)
Optional[TaskDef]	findTaskDefByLabel (self, str label)
Set[Quantum]	findQuantaWithDSType (self, DatasetTypeName datasetTypeName)
bool	checkQuantumInGraph (self, Quantum quantum)
def	writeDotGraph (self, Union[str, io.BufferedIOBase] output)
_T	subset (_T self, Union[QuantumNode, Iterable[QuantumNode]] nodes)
Tuple[_T,...]	subsetToConnected (_T self)
Set[QuantumNode]	determineInputsToQuantumNode (self, QuantumNode node)
Set[QuantumNode]	determineOutputsOfQuantumNode (self, QuantumNode node)
_T	determineConnectionsOfQuantumNode (_T self, QuantumNode node)
_T	determineAncestorsOfQuantumNode (_T self, QuantumNode node)
List[Tuple[QuantumNode, QuantumNode]]	findCycle (self)
def	saveUri (self, uri)
QuantumGraph	loadUri (cls, Union[ButlerURI, str] uri, DimensionUniverse universe, Optional[Iterable[int]] nodes=None, Optional[BuildId] graphID=None)
def	save (self, io.IO[bytes] file)
QuantumGraph	load (cls, io.IO[bytes] file, DimensionUniverse universe, Optional[Iterable[int]] nodes=None, Optional[BuildId] graphID=None)
Generator[TaskDef, None, None]	iterTaskGraph (self)
def	graphID (self)
Generator[QuantumNode, None, None]	__iter__ (self)
int	__len__ (self)
bool	__contains__ (self, QuantumNode node)
dict	__getstate__ (self)
def	__setstate__ (self, dict state)
bool	__eq__ (self, object other)

Detailed Description

QuantumGraph is a directed acyclic graph of `QuantumNode` objects

This data structure represents a concrete workflow generated from a
`Pipeline`.

Parameters
----------
quanta : Mapping of `TaskDef` to sets of `Quantum`
    This maps tasks (and their configs) to the sets of data they are to
    process.

Definition at line 73 of file graph.py.

Constructor & Destructor Documentation

__init__()

def lsst.pipe.base.graph.graph.QuantumGraph.__init__	(		self,
		Mapping[TaskDef, Set[Quantum]]	quanta
	)

Definition at line 85 of file graph.py.

    def __init__(self, quanta: Mapping[TaskDef, Set[Quantum]]):
        self._buildGraphs(quanta)
 

Member Function Documentation

__contains__()

bool lsst.pipe.base.graph.graph.QuantumGraph.__contains__	(		self,
		QuantumNode	node
	)

Definition at line 843 of file graph.py.

    def __contains__(self, node: QuantumNode) -> bool:
        return self._connectedQuanta.has_node(node)
 

__eq__()

bool lsst.pipe.base.graph.graph.QuantumGraph.__eq__	(		self,
		object	other
	)

Definition at line 867 of file graph.py.

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, QuantumGraph):
            return False
        if len(self) != len(other):
            return False
        for node in self:
            if node not in other:
                return False
            if self.determineInputsToQuantumNode(node) != other.determineInputsToQuantumNode(node):
                return False
            if self.determineOutputsOfQuantumNode(node) != other.determineOutputsOfQuantumNode(node):
                return False
        return list(self.taskGraph) == list(other.taskGraph)

__getstate__()

dict lsst.pipe.base.graph.graph.QuantumGraph.__getstate__

(

self

)

Stores a compact form of the graph as a list of graph nodes, and a
tuple of task labels and task configs. The full graph can be
reconstructed with this information, and it preseves the ordering of
the graph ndoes.

Definition at line 846 of file graph.py.

    def __getstate__(self) -> dict:
        """Stores a compact form of the graph as a list of graph nodes, and a
        tuple of task labels and task configs. The full graph can be
        reconstructed with this information, and it preseves the ordering of
        the graph ndoes.
        """
        return {"nodesList": list(self)}
 

__iter__()

Generator[QuantumNode, None, None] lsst.pipe.base.graph.graph.QuantumGraph.__iter__

(

self

)

Definition at line 837 of file graph.py.

    def __iter__(self) -> Generator[QuantumNode, None, None]:
        yield from nx.topological_sort(self._connectedQuanta)
 

__len__()

int lsst.pipe.base.graph.graph.QuantumGraph.__len__

(

self

)

Definition at line 840 of file graph.py.

    def __len__(self) -> int:
        return self._count
 

__setstate__()

def lsst.pipe.base.graph.graph.QuantumGraph.__setstate__	(		self,
		dict	state
	)

Reconstructs the state of the graph from the information persisted
in getstate.

Definition at line 854 of file graph.py.

    def __setstate__(self, state: dict):
        """Reconstructs the state of the graph from the information persisted
        in getstate.
        """
        quanta: DefaultDict[TaskDef, Set[Quantum]] = defaultdict(set)
        quantumToNodeId: Dict[Quantum, NodeId] = {}
        quantumNode: QuantumNode
        for quantumNode in state['nodesList']:
            quanta[quantumNode.taskDef].add(quantumNode.quantum)
            quantumToNodeId[quantumNode.quantum] = quantumNode.nodeId
        _buildId = quantumNode.nodeId.buildId if state['nodesList'] else None  # type: ignore
        self._buildGraphs(quanta, _quantumToNodeId=quantumToNodeId, _buildId=_buildId)
 

allDatasetTypes()

Tuple[DatasetTypeName, ...] lsst.pipe.base.graph.graph.QuantumGraph.allDatasetTypes

(

self

)

Return all the `DatasetTypeName` objects that are contained inside
the graph.

Returns
-------
tuple of `DatasetTypeName`
    All the data set type names that are present in the graph

Definition at line 213 of file graph.py.

    def allDatasetTypes(self) -> Tuple[DatasetTypeName, ...]:
        """Return all the `DatasetTypeName` objects that are contained inside
        the graph.
 
        Returns
        -------
        tuple of `DatasetTypeName`
            All the data set type names that are present in the graph
        """
        return tuple(self._datasetDict.keys())
 

checkQuantumInGraph()

bool lsst.pipe.base.graph.graph.QuantumGraph.checkQuantumInGraph	(		self,
		Quantum	quantum
	)

Check if specified quantum appears in the graph as part of a node.

Parameters
----------
quantum : `Quantum`
    The quantum to search for

Returns
-------
`bool`
    The result of searching for the quantum

Definition at line 438 of file graph.py.

    def checkQuantumInGraph(self, quantum: Quantum) -> bool:
        """Check if specified quantum appears in the graph as part of a node.
 
        Parameters
        ----------
        quantum : `Quantum`
            The quantum to search for
 
        Returns
        -------
        `bool`
            The result of searching for the quantum
        """
        for qset in self._quanta.values():
            if quantum in qset:
                return True
        return False
 

determineAncestorsOfQuantumNode()

_T lsst.pipe.base.graph.graph.QuantumGraph.determineAncestorsOfQuantumNode	(	_T	self,
		QuantumNode	node
	)

Return a graph of the specified node and all the ancestor nodes
directly reachable by walking edges.

Parameters
----------
node : `QuantumNode`
    The node for which all ansestors are to be determined

Returns
-------
graph of `QuantumNode`
    Graph of node and all of its ansestors

Definition at line 555 of file graph.py.

    def determineAncestorsOfQuantumNode(self: _T, node: QuantumNode) -> _T:
        """Return a graph of the specified node and all the ancestor nodes
        directly reachable by walking edges.
 
        Parameters
        ----------
        node : `QuantumNode`
            The node for which all ansestors are to be determined
 
        Returns
        -------
        graph of `QuantumNode`
            Graph of node and all of its ansestors
        """
        predecessorNodes = nx.ancestors(self._connectedQuanta, node)
        predecessorNodes.add(node)
        return self.subset(predecessorNodes)
 

determineConnectionsOfQuantumNode()

_T lsst.pipe.base.graph.graph.QuantumGraph.determineConnectionsOfQuantumNode	(	_T	self,
		QuantumNode	node
	)

Return a graph of `QuantumNode` that are direct inputs and outputs
of a specified node.

Parameters
----------
node : `QuantumNode`
    The node of the graph for which connected nodes are to be
    determined.

Returns
-------
graph : graph of `QuantumNode`
    All the nodes that are directly connected to specified node

Definition at line 536 of file graph.py.

    def determineConnectionsOfQuantumNode(self: _T, node: QuantumNode) -> _T:
        """Return a graph of `QuantumNode` that are direct inputs and outputs
        of a specified node.
 
        Parameters
        ----------
        node : `QuantumNode`
            The node of the graph for which connected nodes are to be
            determined.
 
        Returns
        -------
        graph : graph of `QuantumNode`
            All the nodes that are directly connected to specified node
        """
        nodes = self.determineInputsToQuantumNode(node).union(self.determineOutputsOfQuantumNode(node))
        nodes.add(node)
        return self.subset(nodes)
 

determineInputsToQuantumNode()

Set[QuantumNode] lsst.pipe.base.graph.graph.QuantumGraph.determineInputsToQuantumNode	(		self,
		QuantumNode	node
	)

Return a set of `QuantumNode` that are direct inputs to a specified
node.

Parameters
----------
node : `QuantumNode`
    The node of the graph for which inputs are to be determined

Returns
-------
set of `QuantumNode`
    All the nodes that are direct inputs to specified node

Definition at line 504 of file graph.py.

    def determineInputsToQuantumNode(self, node: QuantumNode) -> Set[QuantumNode]:
        """Return a set of `QuantumNode` that are direct inputs to a specified
        node.
 
        Parameters
        ----------
        node : `QuantumNode`
            The node of the graph for which inputs are to be determined
 
        Returns
        -------
        set of `QuantumNode`
            All the nodes that are direct inputs to specified node
        """
        return set(pred for pred in self._connectedQuanta.predecessors(node))
 

determineOutputsOfQuantumNode()

Set[QuantumNode] lsst.pipe.base.graph.graph.QuantumGraph.determineOutputsOfQuantumNode	(		self,
		QuantumNode	node
	)

Return a set of `QuantumNode` that are direct outputs of a specified
node.

Parameters
----------
node : `QuantumNode`
    The node of the graph for which outputs are to be determined

Returns
-------
set of `QuantumNode`
    All the nodes that are direct outputs to specified node

Definition at line 520 of file graph.py.

    def determineOutputsOfQuantumNode(self, node: QuantumNode) -> Set[QuantumNode]:
        """Return a set of `QuantumNode` that are direct outputs of a specified
        node.
 
        Parameters
        ----------
        node : `QuantumNode`
            The node of the graph for which outputs are to be determined
 
        Returns
        -------
        set of `QuantumNode`
            All the nodes that are direct outputs to specified node
        """
        return set(succ for succ in self._connectedQuanta.successors(node))
 

findCycle()

List[Tuple[QuantumNode, QuantumNode]] lsst.pipe.base.graph.graph.QuantumGraph.findCycle

(

self

)

Check a graph for the presense of cycles and returns the edges of
any cycles found, or an empty list if there is no cycle.

Returns
-------
result : list of tuple of `QuantumNode`, `QuantumNode`
    A list of any graph edges that form a cycle, or an empty list if
    there is no cycle. Empty list to so support if graph.find_cycle()
    syntax as an empty list is falsy.

Definition at line 573 of file graph.py.

    def findCycle(self) -> List[Tuple[QuantumNode, QuantumNode]]:
        """Check a graph for the presense of cycles and returns the edges of
        any cycles found, or an empty list if there is no cycle.
 
        Returns
        -------
        result : list of tuple of `QuantumNode`, `QuantumNode`
            A list of any graph edges that form a cycle, or an empty list if
            there is no cycle. Empty list to so support if graph.find_cycle()
            syntax as an empty list is falsy.
        """
        try:
            return nx.find_cycle(self._connectedQuanta)
        except nx.NetworkXNoCycle:
            return []
 

findQuantaWithDSType()

Set[Quantum] lsst.pipe.base.graph.graph.QuantumGraph.findQuantaWithDSType	(		self,
		DatasetTypeName	datasetTypeName
	)

Return all the `Quantum` that contain a specified `DatasetTypeName`.

Parameters
----------
datasetTypeName : `str`
    The name of the dataset type to search for as a string,
    can also accept a `DatasetTypeName` which is a `NewType` of str for
    type safety in static type checking.

Returns
-------
result : `set` of `QuantumNode` objects
    A `set` of `QuantumNode`s that contain specified `DatasetTypeName`

Raises
------
KeyError
    Raised if the `DatasetTypeName` is not part of the `QuantumGraph`

Definition at line 412 of file graph.py.

    def findQuantaWithDSType(self, datasetTypeName: DatasetTypeName) -> Set[Quantum]:
        """Return all the `Quantum` that contain a specified `DatasetTypeName`.
 
        Parameters
        ----------
        datasetTypeName : `str`
            The name of the dataset type to search for as a string,
            can also accept a `DatasetTypeName` which is a `NewType` of str for
            type safety in static type checking.
 
        Returns
        -------
        result : `set` of `QuantumNode` objects
            A `set` of `QuantumNode`s that contain specified `DatasetTypeName`
 
        Raises
        ------
        KeyError
            Raised if the `DatasetTypeName` is not part of the `QuantumGraph`
 
        """
        tasks = self._datasetDict.getAll(datasetTypeName)
        result: Set[Quantum] = set()
        result = result.union(*(self._quanta[task] for task in tasks))
        return result
 

findTaskDefByLabel()

Optional[TaskDef] lsst.pipe.base.graph.graph.QuantumGraph.findTaskDefByLabel	(		self,
		str	label
	)

Determine which `TaskDef` objects in this graph are associated
with a `str` representing a tasks label.

Parameters
----------
taskName : str
    Name of a task to search for

Returns
-------
result : `TaskDef`
    `TaskDef` objects that has the specified label.

Definition at line 393 of file graph.py.

    def findTaskDefByLabel(self, label: str) -> Optional[TaskDef]:
        """Determine which `TaskDef` objects in this graph are associated
        with a `str` representing a tasks label.
 
        Parameters
        ----------
        taskName : str
            Name of a task to search for
 
        Returns
        -------
        result : `TaskDef`
            `TaskDef` objects that has the specified label.
        """
        for task in self._quanta.keys():
            if label == task.label:
                return task
        return None
 

findTaskDefByName()

List[TaskDef] lsst.pipe.base.graph.graph.QuantumGraph.findTaskDefByName	(		self,
		str	taskName
	)

Determine which `TaskDef` objects in this graph are associated
with a `str` representing a task name (looks at the taskName property
of `TaskDef` objects).

Returns a list of `TaskDef` objects as a `PipelineTask` may appear
multiple times in a graph with different labels.

Parameters
----------
taskName : str
    Name of a task to search for

Returns
-------
result : list of `TaskDef`
    List of the `TaskDef` objects that have the name specified.
    Multiple values are returned in the case that a task is used
    multiple times with different labels.

Definition at line 366 of file graph.py.

    def findTaskDefByName(self, taskName: str) -> List[TaskDef]:
        """Determine which `TaskDef` objects in this graph are associated
        with a `str` representing a task name (looks at the taskName property
        of `TaskDef` objects).
 
        Returns a list of `TaskDef` objects as a `PipelineTask` may appear
        multiple times in a graph with different labels.
 
        Parameters
        ----------
        taskName : str
            Name of a task to search for
 
        Returns
        -------
        result : list of `TaskDef`
            List of the `TaskDef` objects that have the name specified.
            Multiple values are returned in the case that a task is used
            multiple times with different labels.
        """
        results = []
        for task in self._quanta.keys():
            split = task.taskName.split('.')
            if split[-1] == taskName:
                results.append(task)
        return results
 

findTasksWithInput()

Iterable[TaskDef] lsst.pipe.base.graph.graph.QuantumGraph.findTasksWithInput	(		self,
		DatasetTypeName	datasetTypeName
	)

Find all tasks that have the specified dataset type name as an
input.

Parameters
----------
datasetTypeName : `str`
    A string representing the name of a dataset type to be queried,
    can also accept a `DatasetTypeName` which is a `NewType` of str for
    type safety in static type checking.

Returns
-------
tasks : iterable of `TaskDef`
    `TaskDef` objects that have the specified `DatasetTypeName` as an
    input, list will be empty if no tasks use specified
    `DatasetTypeName` as an input.

Raises
------
KeyError
    Raised if the `DatasetTypeName` is not part of the `QuantumGraph`

Definition at line 289 of file graph.py.

    def findTasksWithInput(self, datasetTypeName: DatasetTypeName) -> Iterable[TaskDef]:
        """Find all tasks that have the specified dataset type name as an
        input.
 
        Parameters
        ----------
        datasetTypeName : `str`
            A string representing the name of a dataset type to be queried,
            can also accept a `DatasetTypeName` which is a `NewType` of str for
            type safety in static type checking.
 
        Returns
        -------
        tasks : iterable of `TaskDef`
            `TaskDef` objects that have the specified `DatasetTypeName` as an
            input, list will be empty if no tasks use specified
            `DatasetTypeName` as an input.
 
        Raises
        ------
        KeyError
            Raised if the `DatasetTypeName` is not part of the `QuantumGraph`
        """
        return (c for c in self._datasetDict.getInputs(datasetTypeName))
 

findTaskWithOutput()

Optional[TaskDef] lsst.pipe.base.graph.graph.QuantumGraph.findTaskWithOutput	(		self,
		DatasetTypeName	datasetTypeName
	)

Find all tasks that have the specified dataset type name as an
output.

Parameters
----------
datasetTypeName : `str`
    A string representing the name of a dataset type to be queried,
    can also accept a `DatasetTypeName` which is a `NewType` of str for
    type safety in static type checking.

Returns
-------
`TaskDef` or `None`
    `TaskDef` that outputs `DatasetTypeName` as an output or None if
    none of the tasks produce this `DatasetTypeName`.

Raises
------
KeyError
    Raised if the `DatasetTypeName` is not part of the `QuantumGraph`

Definition at line 314 of file graph.py.

    def findTaskWithOutput(self, datasetTypeName: DatasetTypeName) -> Optional[TaskDef]:
        """Find all tasks that have the specified dataset type name as an
        output.
 
        Parameters
        ----------
        datasetTypeName : `str`
            A string representing the name of a dataset type to be queried,
            can also accept a `DatasetTypeName` which is a `NewType` of str for
            type safety in static type checking.
 
        Returns
        -------
        `TaskDef` or `None`
            `TaskDef` that outputs `DatasetTypeName` as an output or None if
            none of the tasks produce this `DatasetTypeName`.
 
        Raises
        ------
        KeyError
            Raised if the `DatasetTypeName` is not part of the `QuantumGraph`
        """
        return self._datasetDict.getOutput(datasetTypeName)
 

getNodesForTask()

FrozenSet[QuantumNode] lsst.pipe.base.graph.graph.QuantumGraph.getNodesForTask	(		self,
		TaskDef	taskDef
	)

Return all the `QuantumNodes` associated with a `TaskDef`.

Parameters
----------
taskDef : `TaskDef`
    The `TaskDef` for which `Quantum` are to be queried

Returns
-------
frozenset of `QuantumNodes`
    The `frozenset` of `QuantumNodes` that is associated with the
    specified `TaskDef`.

Definition at line 273 of file graph.py.

    def getNodesForTask(self, taskDef: TaskDef) -> FrozenSet[QuantumNode]:
        """Return all the `QuantumNodes` associated with a `TaskDef`.
 
        Parameters
        ----------
        taskDef : `TaskDef`
            The `TaskDef` for which `Quantum` are to be queried
 
        Returns
        -------
        frozenset of `QuantumNodes`
            The `frozenset` of `QuantumNodes` that is associated with the
            specified `TaskDef`.
        """
        return frozenset(self._taskToQuantumNode[taskDef])
 

getQuantaForTask()

FrozenSet[Quantum] lsst.pipe.base.graph.graph.QuantumGraph.getQuantaForTask	(		self,
		TaskDef	taskDef
	)

Return all the `Quantum` associated with a `TaskDef`.

Parameters
----------
taskDef : `TaskDef`
    The `TaskDef` for which `Quantum` are to be queried

Returns
-------
frozenset of `Quantum`
    The `set` of `Quantum` that is associated with the specified
    `TaskDef`.

Definition at line 257 of file graph.py.

    def getQuantaForTask(self, taskDef: TaskDef) -> FrozenSet[Quantum]:
        """Return all the `Quantum` associated with a `TaskDef`.
 
        Parameters
        ----------
        taskDef : `TaskDef`
            The `TaskDef` for which `Quantum` are to be queried
 
        Returns
        -------
        frozenset of `Quantum`
            The `set` of `Quantum` that is associated with the specified
            `TaskDef`.
        """
        return frozenset(self._quanta[taskDef])
 

getQuantumNodeByNodeId()

QuantumNode lsst.pipe.base.graph.graph.QuantumGraph.getQuantumNodeByNodeId	(		self,
		NodeId	nodeId
	)

Lookup a `QuantumNode` from an id associated with the node.

Parameters
----------
nodeId : `NodeId`
    The number associated with a node

Returns
-------
node : `QuantumNode`
    The node corresponding with input number

Raises
------
IndexError
    Raised if the requested nodeId is not in the graph.
IncompatibleGraphError
    Raised if the nodeId was built with a different graph than is not
    this instance (or a graph instance that produced this instance
    through and operation such as subset)

Definition at line 231 of file graph.py.

    def getQuantumNodeByNodeId(self, nodeId: NodeId) -> QuantumNode:
        """Lookup a `QuantumNode` from an id associated with the node.
 
        Parameters
        ----------
        nodeId : `NodeId`
            The number associated with a node
 
        Returns
        -------
        node : `QuantumNode`
            The node corresponding with input number
 
        Raises
        ------
        IndexError
            Raised if the requested nodeId is not in the graph.
        IncompatibleGraphError
            Raised if the nodeId was built with a different graph than is not
            this instance (or a graph instance that produced this instance
            through and operation such as subset)
        """
        if nodeId.buildId != self._buildId:
            raise IncompatibleGraphError("This node was built from a different, incompatible, graph instance")
        return self._nodeIdMap[nodeId]
 

graph()

nx.DiGraph lsst.pipe.base.graph.graph.QuantumGraph.graph

(

self

)

Return a graph representing the relations between all the
`QuantumNode` objects. Largely it should be preferred to iterate
over, and use methods of this class, but sometimes direct access to
the networkx object may be helpful

Returns
-------
graph : `networkx.Digraph`
    Internal datastructure that holds relations of `QuantumNode`
    objects

Definition at line 172 of file graph.py.

    def graph(self) -> nx.DiGraph:
        """Return a graph representing the relations between all the
        `QuantumNode` objects. Largely it should be preferred to iterate
        over, and use methods of this class, but sometimes direct access to
        the networkx object may be helpful
 
        Returns
        -------
        graph : `networkx.Digraph`
            Internal datastructure that holds relations of `QuantumNode`
            objects
        """
        return self._connectedQuanta
 

graphID()

def lsst.pipe.base.graph.graph.QuantumGraph.graphID

(

self

)

Returns the ID generated by the graph at construction time

Definition at line 832 of file graph.py.

    def graphID(self):
        """Returns the ID generated by the graph at construction time
        """
        return self._buildId
 

inputQuanta()

Iterable[QuantumNode] lsst.pipe.base.graph.graph.QuantumGraph.inputQuanta

(

self

)

Make a `list` of all `QuantumNode` objects that are 'input' nodes
to the graph, meaning those nodes to not depend on any other nodes in
the graph.

Returns
-------
inputNodes : iterable of `QuantumNode`
    A list of nodes that are inputs to the graph

Definition at line 187 of file graph.py.

    def inputQuanta(self) -> Iterable[QuantumNode]:
        """Make a `list` of all `QuantumNode` objects that are 'input' nodes
        to the graph, meaning those nodes to not depend on any other nodes in
        the graph.
 
        Returns
        -------
        inputNodes : iterable of `QuantumNode`
            A list of nodes that are inputs to the graph
        """
        return (q for q, n in self._connectedQuanta.in_degree if n == 0)
 

isConnected()

bool lsst.pipe.base.graph.graph.QuantumGraph.isConnected

(

self

)

Return True if all of the nodes in the graph are connected, ignores
directionality of connections.

Definition at line 225 of file graph.py.

    def isConnected(self) -> bool:
        """Return True if all of the nodes in the graph are connected, ignores
        directionality of connections.
        """
        return nx.is_weakly_connected(self._connectedQuanta)
 

iterTaskGraph()

Generator[TaskDef, None, None] lsst.pipe.base.graph.graph.QuantumGraph.iterTaskGraph

(

self

)

Iterate over the `taskGraph` attribute in topological order

Yields
------
taskDef : `TaskDef`
    `TaskDef` objects in topological order

Definition at line 821 of file graph.py.

    def iterTaskGraph(self) -> Generator[TaskDef, None, None]:
        """Iterate over the `taskGraph` attribute in topological order
 
        Yields
        ------
        taskDef : `TaskDef`
            `TaskDef` objects in topological order
        """
        yield from nx.topological_sort(self.taskGraph)
 

load()

QuantumGraph lsst.pipe.base.graph.graph.QuantumGraph.load	(		cls,
		io.IO[bytes]	file,
		DimensionUniverse	universe,
		Optional[Iterable[int]]	nodes = None,
		Optional[BuildId]	graphID = None
	)

Read QuantumGraph from a file that was made by `save`.

Parameters
----------
file : `io.IO` of bytes
    File with pickle data open in binary mode.
universe: `~lsst.daf.butler.DimensionUniverse`
    DimensionUniverse instance, not used by the method itself but
    needed to ensure that registry data structures are initialized.
nodes: iterable of `int` or None
    Numbers that correspond to nodes in the graph. If specified, only
    these nodes will be loaded. Defaults to None, in which case all
    nodes will be loaded.
graphID : `str` or `None`
    If specified this ID is verified against the loaded graph prior to
    loading any Nodes. This defaults to None in which case no
    validation is done.

Returns
-------
graph : `QuantumGraph`
    Resulting QuantumGraph instance.

Raises
------
TypeError
    Raised if pickle contains instance of a type other than
    QuantumGraph.
ValueError
    Raised if one or more of the nodes requested is not in the
    `QuantumGraph` or if graphID parameter does not match the graph
    being loaded or if the supplied uri does not point at a valid
    `QuantumGraph` save file.

Notes
-----
Reading Quanta from pickle requires existence of singleton
DimensionUniverse which is usually instantiated during Registry
initialization. To make sure that DimensionUniverse exists this method
accepts dummy DimensionUniverse argument.

Definition at line 764 of file graph.py.

             ) -> QuantumGraph:
        """Read QuantumGraph from a file that was made by `save`.
 
        Parameters
        ----------
        file : `io.IO` of bytes
            File with pickle data open in binary mode.
        universe: `~lsst.daf.butler.DimensionUniverse`
            DimensionUniverse instance, not used by the method itself but
            needed to ensure that registry data structures are initialized.
        nodes: iterable of `int` or None
            Numbers that correspond to nodes in the graph. If specified, only
            these nodes will be loaded. Defaults to None, in which case all
            nodes will be loaded.
        graphID : `str` or `None`
            If specified this ID is verified against the loaded graph prior to
            loading any Nodes. This defaults to None in which case no
            validation is done.
 
        Returns
        -------
        graph : `QuantumGraph`
            Resulting QuantumGraph instance.
 
        Raises
        ------
        TypeError
            Raised if pickle contains instance of a type other than
            QuantumGraph.
        ValueError
            Raised if one or more of the nodes requested is not in the
            `QuantumGraph` or if graphID parameter does not match the graph
            being loaded or if the supplied uri does not point at a valid
            `QuantumGraph` save file.
 
        Notes
        -----
        Reading Quanta from pickle requires existence of singleton
        DimensionUniverse which is usually instantiated during Registry
        initialization. To make sure that DimensionUniverse exists this method
        accepts dummy DimensionUniverse argument.
        """
        # Try to see if the file handle contains pickle data, this will be
        # removed in the future
        try:
            qgraph = pickle.load(file)
            warnings.warn("Pickle graphs are deprecated, please re-save your graph with the save method")
        except pickle.UnpicklingError:
            with LoadHelper(file) as loader:  # type: ignore # needed because we don't have Protocols yet
                qgraph = loader.load(nodes, graphID)
        if not isinstance(qgraph, QuantumGraph):
            raise TypeError(f"QuantumGraph pickle file has contains unexpected object type: {type(qgraph)}")
        return qgraph
 

loadUri()

QuantumGraph lsst.pipe.base.graph.graph.QuantumGraph.loadUri	(		cls,
		Union[ButlerURI, str]	uri,
		DimensionUniverse	universe,
		Optional[Iterable[int]]	nodes = None,
		Optional[BuildId]	graphID = None
	)

Read `QuantumGraph` from a URI.

Parameters
----------
uri : `ButlerURI` or `str`
    URI from where to load the graph.
universe: `~lsst.daf.butler.DimensionUniverse`
    DimensionUniverse instance, not used by the method itself but
    needed to ensure that registry data structures are initialized.
nodes: iterable of `int` or None
    Numbers that correspond to nodes in the graph. If specified, only
    these nodes will be loaded. Defaults to None, in which case all
    nodes will be loaded.
graphID : `str` or `None`
    If specified this ID is verified against the loaded graph prior to
    loading any Nodes. This defaults to None in which case no
    validation is done.

Returns
-------
graph : `QuantumGraph`
    Resulting QuantumGraph instance.

Raises
------
TypeError
    Raised if pickle contains instance of a type other than
    QuantumGraph.
ValueError
    Raised if one or more of the nodes requested is not in the
    `QuantumGraph` or if graphID parameter does not match the graph
    being loaded or if the supplied uri does not point at a valid
    `QuantumGraph` save file.


Notes
-----
Reading Quanta from pickle requires existence of singleton
DimensionUniverse which is usually instantiated during Registry
initialization. To make sure that DimensionUniverse exists this method
accepts dummy DimensionUniverse argument.

Definition at line 604 of file graph.py.

                ) -> QuantumGraph:
        """Read `QuantumGraph` from a URI.
 
        Parameters
        ----------
        uri : `ButlerURI` or `str`
            URI from where to load the graph.
        universe: `~lsst.daf.butler.DimensionUniverse`
            DimensionUniverse instance, not used by the method itself but
            needed to ensure that registry data structures are initialized.
        nodes: iterable of `int` or None
            Numbers that correspond to nodes in the graph. If specified, only
            these nodes will be loaded. Defaults to None, in which case all
            nodes will be loaded.
        graphID : `str` or `None`
            If specified this ID is verified against the loaded graph prior to
            loading any Nodes. This defaults to None in which case no
            validation is done.
 
        Returns
        -------
        graph : `QuantumGraph`
            Resulting QuantumGraph instance.
 
        Raises
        ------
        TypeError
            Raised if pickle contains instance of a type other than
            QuantumGraph.
        ValueError
            Raised if one or more of the nodes requested is not in the
            `QuantumGraph` or if graphID parameter does not match the graph
            being loaded or if the supplied uri does not point at a valid
            `QuantumGraph` save file.
 
 
        Notes
        -----
        Reading Quanta from pickle requires existence of singleton
        DimensionUniverse which is usually instantiated during Registry
        initialization. To make sure that DimensionUniverse exists this method
        accepts dummy DimensionUniverse argument.
        """
        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 pickle files when the resource
        # is remote. For now use the local file variant. For a local file
        # as_local() does nothing.
 
        if uri.getExtension() in (".pickle", ".pkl"):
            with uri.as_local() as local, open(local.ospath, "rb") as fd:
                warnings.warn("Pickle graphs are deprecated, please re-save your graph with the save method")
                qgraph = pickle.load(fd)
        elif uri.getExtension() in ('.qgraph'):
            with LoadHelper(uri) as loader:
                qgraph = loader.load(nodes, graphID)
        else:
            raise ValueError("Only know how to handle files saved as `pickle`, `pkl`, or `qgraph`")
        if not isinstance(qgraph, QuantumGraph):
            raise TypeError(f"QuantumGraph save file contains unexpected object type: {type(qgraph)}")
        return qgraph
 

outputQuanta()

Iterable[QuantumNode] lsst.pipe.base.graph.graph.QuantumGraph.outputQuanta

(

self

)

Make a `list` of all `QuantumNode` objects that are 'output' nodes
to the graph, meaning those nodes have no nodes that depend them in
the graph.

Returns
-------
outputNodes : iterable of `QuantumNode`
    A list of nodes that are outputs of the graph

Definition at line 200 of file graph.py.

    def outputQuanta(self) -> Iterable[QuantumNode]:
        """Make a `list` of all `QuantumNode` objects that are 'output' nodes
        to the graph, meaning those nodes have no nodes that depend them in
        the graph.
 
        Returns
        -------
        outputNodes : iterable of `QuantumNode`
            A list of nodes that are outputs of the graph
        """
        return [q for q, n in self._connectedQuanta.out_degree if n == 0]
 

save()

def lsst.pipe.base.graph.graph.QuantumGraph.save	(		self,
		io.IO[bytes]	file
	)

Save QuantumGraph to a file.

Presently we store QuantumGraph in pickle format, this could
potentially change in the future if better format is found.

Parameters
----------
file : `io.BufferedIOBase`
    File to write pickle data open in binary mode.

Definition at line 670 of file graph.py.

    def save(self, file: io.IO[bytes]):
        """Save QuantumGraph to a file.
 
        Presently we store QuantumGraph in pickle format, this could
        potentially change in the future if better format is found.
 
        Parameters
        ----------
        file : `io.BufferedIOBase`
            File to write pickle data open in binary mode.
        """
        buffer = self._buildSaveObject()
        file.write(buffer)  # type: ignore # Ignore because bytearray is safe to use in place of bytes
 

saveUri()

def lsst.pipe.base.graph.graph.QuantumGraph.saveUri	(		self,
			uri
	)

Save `QuantumGraph` to the specified URI.

Parameters
----------
uri : `ButlerURI` or `str`
    URI to where the graph should be saved.

Definition at line 589 of file graph.py.

    def saveUri(self, uri):
        """Save `QuantumGraph` to the specified URI.
 
        Parameters
        ----------
        uri : `ButlerURI` or `str`
            URI to where the graph should be saved.
        """
        buffer = self._buildSaveObject()
        butlerUri = ButlerURI(uri)
        if butlerUri.getExtension() not in (".qgraph"):
            raise TypeError(f"Can currently only save a graph in qgraph format not {uri}")
        butlerUri.write(buffer)  # type: ignore  # Ignore because bytearray is safe to use in place of bytes
 

subset()

_T lsst.pipe.base.graph.graph.QuantumGraph.subset	(	_T	self,
		Union[QuantumNode, Iterable[QuantumNode]]	nodes
	)

Create a new graph object that contains the subset of the nodes
specified as input. Node number is preserved.

Parameters
----------
nodes : `QuantumNode` or iterable of `QuantumNode`

Returns
-------
graph : instance of graph type
    An instance of the type from which the subset was created

Definition at line 466 of file graph.py.

    def subset(self: _T, nodes: Union[QuantumNode, Iterable[QuantumNode]]) -> _T:
        """Create a new graph object that contains the subset of the nodes
        specified as input. Node number is preserved.
 
        Parameters
        ----------
        nodes : `QuantumNode` or iterable of `QuantumNode`
 
        Returns
        -------
        graph : instance of graph type
            An instance of the type from which the subset was created
        """
        if not isinstance(nodes, Iterable):
            nodes = (nodes, )
        quantumSubgraph = self._connectedQuanta.subgraph(nodes).nodes
        quantumMap = defaultdict(set)
 
        node: QuantumNode
        for node in quantumSubgraph:
            quantumMap[node.taskDef].add(node.quantum)
        # Create an empty graph, and then populate it with custom mapping
        newInst = type(self)({})
        newInst._buildGraphs(quantumMap, _quantumToNodeId={n.quantum: n.nodeId for n in nodes},
                             _buildId=self._buildId)
        return newInst
 

subsetToConnected()

Tuple[_T, ...] lsst.pipe.base.graph.graph.QuantumGraph.subsetToConnected

(

_T

self

)

Generate a list of subgraphs where each is connected.

Returns
-------
result : list of `QuantumGraph`
    A list of graphs that are each connected

Definition at line 493 of file graph.py.

    def subsetToConnected(self: _T) -> Tuple[_T, ...]:
        """Generate a list of subgraphs where each is connected.
 
        Returns
        -------
        result : list of `QuantumGraph`
            A list of graphs that are each connected
        """
        return tuple(self.subset(connectedSet)
                     for connectedSet in nx.weakly_connected_components(self._connectedQuanta))
 

taskGraph()

nx.DiGraph lsst.pipe.base.graph.graph.QuantumGraph.taskGraph

(

self

)

Return a graph representing the relations between the tasks inside
the quantum graph.

Returns
-------
taskGraph : `networkx.Digraph`
    Internal datastructure that holds relations of `TaskDef` objects

Definition at line 160 of file graph.py.

    def taskGraph(self) -> nx.DiGraph:
        """Return a graph representing the relations between the tasks inside
        the quantum graph.
 
        Returns
        -------
        taskGraph : `networkx.Digraph`
            Internal datastructure that holds relations of `TaskDef` objects
        """
        return self._taskGraph
 

tasksWithDSType()

Iterable[TaskDef] lsst.pipe.base.graph.graph.QuantumGraph.tasksWithDSType	(		self,
		DatasetTypeName	datasetTypeName
	)

Find all tasks that are associated with the specified dataset type
name.

Parameters
----------
datasetTypeName : `str`
    A string representing the name of a dataset type to be queried,
    can also accept a `DatasetTypeName` which is a `NewType` of str for
    type safety in static type checking.

Returns
-------
result : iterable of `TaskDef`
    `TaskDef` objects that are associated with the specified
    `DatasetTypeName`

Raises
------
KeyError
    Raised if the `DatasetTypeName` is not part of the `QuantumGraph`

Definition at line 338 of file graph.py.

    def tasksWithDSType(self, datasetTypeName: DatasetTypeName) -> Iterable[TaskDef]:
        """Find all tasks that are associated with the specified dataset type
        name.
 
        Parameters
        ----------
        datasetTypeName : `str`
            A string representing the name of a dataset type to be queried,
            can also accept a `DatasetTypeName` which is a `NewType` of str for
            type safety in static type checking.
 
        Returns
        -------
        result : iterable of `TaskDef`
            `TaskDef` objects that are associated with the specified
            `DatasetTypeName`
 
        Raises
        ------
        KeyError
            Raised if the `DatasetTypeName` is not part of the `QuantumGraph`
        """
        results = self.findTasksWithInput(datasetTypeName)
        output = self.findTaskWithOutput(datasetTypeName)
        if output is not None:
            results = chain(results, (output,))
        return results
 

writeDotGraph()

def lsst.pipe.base.graph.graph.QuantumGraph.writeDotGraph	(		self,
		Union[str, io.BufferedIOBase]	output
	)

Write out the graph as a dot graph.

Parameters
----------
output : str or `io.BufferedIOBase`
    Either a filesystem path to write to, or a file handle object

Definition at line 456 of file graph.py.

    def writeDotGraph(self, output: Union[str, io.BufferedIOBase]):
        """Write out the graph as a dot graph.
 
        Parameters
        ----------
        output : str or `io.BufferedIOBase`
            Either a filesystem path to write to, or a file handle object
        """
        write_dot(self._connectedQuanta, output)
 

---

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