Coverage for python/lsst/ctrl/bps/generic_workflow.py: 96%
562 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-26 09:18 +0000
1# This file is part of ctrl_bps.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28"""Class definitions for a Generic Workflow Graph."""
30__all__ = [
31 "GenericWorkflow",
32 "GenericWorkflowExec",
33 "GenericWorkflowFile",
34 "GenericWorkflowGroup",
35 "GenericWorkflowJob",
36 "GenericWorkflowLazyGroup",
37 "GenericWorkflowNode",
38 "GenericWorkflowNodeType",
39 "GenericWorkflowNoopJob",
40]
43import dataclasses
44import itertools
45import logging
46import pickle
47from collections import Counter, defaultdict
48from collections.abc import Iterable, Iterator
49from enum import IntEnum, auto
50from typing import IO, Any, BinaryIO, Literal, cast, overload
52from networkx import DiGraph, topological_sort
53from networkx.algorithms.dag import is_directed_acyclic_graph
55from lsst.utils.iteration import ensure_iterable
57from .bps_draw import draw_networkx_dot
58from .bps_utils import subset_dimension_values
60_LOG = logging.getLogger(__name__)
63@dataclasses.dataclass(slots=True)
64class GenericWorkflowFile:
65 """Information about a file that may be needed by various workflow
66 management services.
67 """
69 name: str
70 """Lookup key (logical file name) of file/directory. Must be unique
71 within run.
72 """
74 src_uri: str | None = None # don't know that need ResourcePath
75 """Original location of file/directory.
76 """
78 wms_transfer: bool = False
79 """Whether the WMS should ignore file or not. Default is False.
80 """
82 job_access_remote: bool = False
83 """Whether the job can remotely access file (using separately specified
84 file access protocols). Default is False.
85 """
87 job_shared: bool = False
88 """Whether job requires its own copy of this file. Default is False.
89 """
91 def __hash__(self) -> int:
92 return hash(self.name)
95@dataclasses.dataclass(slots=True)
96class GenericWorkflowExec:
97 """Information about an executable that may be needed by various workflow
98 management services.
99 """
101 name: str
102 """Lookup key (logical file name) of executable. Must be unique
103 within run.
104 """
106 src_uri: str | None = None # don't know that need ResourcePath
107 """Original location of executable.
108 """
110 transfer_executable: bool = False
111 """Whether the WMS/plugin is responsible for staging executable to
112 location usable by job.
113 """
115 def __hash__(self) -> int:
116 return hash(self.name)
119class GenericWorkflowNodeType(IntEnum):
120 """Type of valid types for nodes in the GenericWorkflow."""
122 NOOP = auto()
123 """Does nothing, but enforces special dependencies."""
125 PAYLOAD = auto()
126 """Typical workflow job."""
128 GROUP = auto()
129 """A special group (subdag) of jobs."""
131 LAZY_GROUP = auto()
132 """When run will generate sub-workflow of jobs."""
135@dataclasses.dataclass(slots=True)
136class GenericWorkflowNode:
137 """Base class for nodes in the GenericWorkflow."""
139 name: str
140 """Name of node. Must be unique within workflow."""
142 label: str
143 """"Primary user-facing label for job. Does not need to be unique and
144 may be used for summary reports or to group nodes."""
146 def __hash__(self) -> int:
147 return hash(self.name)
149 @property
150 def node_type(self) -> GenericWorkflowNodeType:
151 """Type of node."""
152 raise NotImplementedError(f"{type(self).__name__} needs to override node_type.")
155@dataclasses.dataclass(slots=True)
156class GenericWorkflowNoopJob(GenericWorkflowNode):
157 """Job that does no work. Used for special dependencies."""
159 @property
160 def node_type(self) -> GenericWorkflowNodeType:
161 """Indicate this is a noop job."""
162 return GenericWorkflowNodeType.NOOP
165@dataclasses.dataclass(slots=True)
166class GenericWorkflowJob(GenericWorkflowNode):
167 """Information about a job that may be needed by various workflow
168 management services.
169 """
171 quanta_counts: Counter[str] = dataclasses.field(default_factory=Counter)
172 """Counts of quanta per task label in job.
173 """
175 tags: dict[str, Any] = dataclasses.field(default_factory=dict)
176 """Other key/value pairs for job that user may want to use as a filter.
177 """
179 executable: GenericWorkflowExec | None = None
180 """Executable for job.
181 """
183 arguments: str | None = None
184 """Command line arguments for job.
185 """
187 cmdvals: dict[str, Any] = dataclasses.field(default_factory=dict)
188 """Values for variables in cmdline when using lazy command line creation.
189 """
191 memory_multiplier: float | None = None
192 """Memory growth rate between retries.
193 """
195 request_memory: int | None = None # MB
196 """Max memory (in MB) that the job is expected to need.
197 """
199 request_memory_max: int | None = None # MB
200 """Max memory (in MB) that the job should ever use.
201 """
203 request_cpus: int | None = None # cores
204 """Max number of cpus that the job is expected to need.
205 """
207 request_disk: int | None = None # MB
208 """Max amount of job scratch disk (in MB) that the job is expected to need.
209 """
211 request_walltime: str | None = None # minutes
212 """Max amount of time (in seconds) that the job is expected to need.
213 """
215 compute_site: str | None = None
216 """Key to look up site-specific information for running the job.
217 """
219 accounting_group: str | None = None
220 """Name of the accounting group to use.
221 """
223 accounting_user: str | None = None
224 """Name of the user to use for accounting purposes.
225 """
227 mail_to: str | None = None
228 """Comma separated list of email addresses for emailing job status.
229 """
231 when_to_mail: str | None = None
232 """WMS-specific terminology for when to email job status.
233 """
235 number_of_retries: int | None = None
236 """Number of times to automatically retry a failed job.
237 """
239 retry_unless_exit: int | list[int] | None = None
240 """Exit code(s) for job that means to not automatically retry.
241 """
243 abort_on_value: int | None = None
244 """Job exit value for signals to abort the entire workflow.
245 """
247 abort_return_value: int | None = None
248 """Exit value to use when aborting the entire workflow.
249 """
251 priority: str | None = None
252 """Initial priority of job in WMS-format.
253 """
255 category: str | None = None
256 """WMS-facing label of job within single workflow (e.g., can be used for
257 throttling jobs within a single workflow).
258 """
260 concurrency_limit: str | None = None
261 """Names of concurrency limits that the WMS plugin can appropriately
262 translate to limit the number of this job across all running workflows.
263 """
265 queue: str | None = None
266 """Name of queue to use. Different WMS can translate this concept
267 differently.
268 """
270 pre_cmdline: str | None = None
271 """Command line to be executed prior to executing job.
272 """
274 post_cmdline: str | None = None
275 """Command line to be executed after job executes.
277 Should be executed regardless of exit status.
278 """
280 preemptible: bool | None = None
281 """The flag indicating whether the job can be preempted.
282 """
284 profile: dict[str, Any] = dataclasses.field(default_factory=dict)
285 """Nested dictionary of WMS-specific key/value pairs with primary key being
286 WMS key (e.g., pegasus, condor, panda).
287 """
289 attrs: dict[str, Any] = dataclasses.field(default_factory=dict)
290 """Key/value pairs of job attributes (for WMS that have attributes in
291 addition to commands).
292 """
294 environment: dict[str, Any] = dataclasses.field(default_factory=dict)
295 """Environment variable names and values to be explicitly set inside job.
296 """
298 compute_cloud: str | None = None
299 """Key to look up cloud-specific information for running the job.
300 """
302 @property
303 def node_type(self) -> GenericWorkflowNodeType:
304 """Indicate this is a payload job."""
305 return GenericWorkflowNodeType.PAYLOAD
308class GenericWorkflow(DiGraph):
309 """A generic representation of a workflow used to submit to specific
310 workflow management systems.
312 Parameters
313 ----------
314 name : `str`
315 Name of generic workflow.
316 incoming_graph_data : `~typing.Any`, optional
317 Data used to initialized graph that is passed through to DiGraph
318 constructor. Can be any type supported by networkx.DiGraph.
319 **attr : `dict`
320 Keyword arguments passed through to DiGraph constructor.
321 """
323 def __init__(self, name: str, incoming_graph_data: Any | None = None, **attr: Any) -> None:
324 super().__init__(incoming_graph_data, **attr)
325 self._name = name
326 self.run_attrs: dict[str, str] = {}
327 self._job_labels = GenericWorkflowLabels()
328 self._files: dict[str, GenericWorkflowFile] = {}
329 self._executables: dict[str, GenericWorkflowExec] = {}
330 self._inputs: dict[
331 str, list[GenericWorkflowFile]
332 ] = {} # mapping job.names to list of GenericWorkflowFile
333 self._outputs: dict[
334 str, list[GenericWorkflowFile]
335 ] = {} # mapping job.names to list of GenericWorkflowFile
336 self.run_id = None
337 self._final: GenericWorkflowJob | GenericWorkflow | None = None
339 # Starting from ver. 3.6 of NetworkX, the DiGraph class defines its custom
340 # __new__ method that explicitly defines arguments it accepts. As a result,
341 # we need to override it to let our subclass use different ones.
342 #
343 # Notes
344 # -----
345 # Most likely overriding __new__ in this manner will prevent us from using
346 # different graph backends with our subclass. However, since we are not
347 # using any backends, this should not be a problem at the moment.
348 def __new__(cls, *args, **kwargs) -> "GenericWorkflow":
349 return object.__new__(cls)
351 @property
352 def name(self) -> str:
353 """Retrieve name of generic workflow.
355 Returns
356 -------
357 name : `str`
358 Name of generic workflow.
359 """
360 return self._name
362 @property
363 def quanta_counts(self) -> Counter[str]:
364 """Count of quanta per task label (`collections.Counter`)."""
365 qcounts: Counter[str] = Counter()
366 for job_name in self:
367 gwjob = self.get_job(job_name)
368 if hasattr(gwjob, "quanta_counts"):
369 qcounts += gwjob.quanta_counts
370 return qcounts
372 @property
373 def labels(self) -> list[str]:
374 """Job labels (`list` [`str`], read-only)."""
375 return self._job_labels.labels
377 def regenerate_labels(self) -> None:
378 """Regenerate the list of job labels."""
379 self._job_labels = GenericWorkflowLabels()
380 for job_name in self:
381 job = self.get_job(job_name)
382 if job.node_type == GenericWorkflowNodeType.PAYLOAD:
383 job = cast(GenericWorkflowJob, job)
384 parents_labels: list[str] = []
385 for parent_name in self.predecessors(job.name):
386 parent_job = self.get_job(parent_name)
387 if parent_job.node_type == GenericWorkflowNodeType.PAYLOAD:
388 # parent_job = cast(GenericWorkflowJob, parent_job)
389 parents_labels.append(parent_job.label)
390 children_labels: list[str] = []
391 for child_name in self.successors(job.name):
392 child_job = self.get_job(child_name)
393 if child_job.node_type == GenericWorkflowNodeType.PAYLOAD:
394 # child_job = cast(GenericWorkflowJob, child_job)
395 children_labels.append(child_job.label)
396 self._job_labels.add_job(job, parents_labels, children_labels)
398 @property
399 def job_counts(self) -> Counter[str]:
400 """Count of jobs per job label (`collections.Counter`)."""
401 jcounts = self._job_labels.job_counts
403 # Final is separate
404 final = self.get_final()
405 if final:
406 if isinstance(final, GenericWorkflow): 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 jcounts.update(final.job_counts)
408 else:
409 jcounts[final.label] += 1
411 return jcounts
413 def __iter__(self) -> Iterator[str]:
414 """Return iterator of job names in topologically sorted order."""
415 return topological_sort(self)
417 @overload
418 def get_files(self, data: Literal[False], transfer_only: bool = True) -> list[str]: ... 418 ↛ exitline 418 didn't return from function 'get_files' because
420 @overload
421 def get_files(self, data: Literal[True], transfer_only: bool = True) -> list[GenericWorkflowFile]: ... 421 ↛ exitline 421 didn't return from function 'get_files' because
423 def get_files(
424 self, data: bool = False, transfer_only: bool = True
425 ) -> list[GenericWorkflowFile] | list[str]:
426 """Retrieve files from generic workflow.
428 Need API in case change way files are stored (e.g., make
429 workflow a bipartite graph with jobs and files nodes).
431 Parameters
432 ----------
433 data : `bool`, optional
434 Whether to return the file data as well as the file object name
435 (The default is `False`).
436 transfer_only : `bool`, optional
437 Whether to only return files for which a workflow management system
438 would be responsible for transferring.
440 Returns
441 -------
442 files : `list` [`lsst.ctrl.bps.GenericWorkflowFile`] or `list` [`str`]
443 File names or objects from generic workflow meeting specifications.
444 """
445 files: list[Any] = [] # Any for mypy to allow the different append lines.
446 for filename, file in self._files.items():
447 if not transfer_only or file.wms_transfer:
448 if not data:
449 files.append(filename)
450 else:
451 files.append(file)
452 return files
454 def add_job(
455 self,
456 job: GenericWorkflowNode,
457 parent_names: str | list[str] | None = None,
458 child_names: str | list[str] | None = None,
459 ) -> None:
460 """Add job to generic workflow.
462 Parameters
463 ----------
464 job : `lsst.ctrl.bps.GenericWorkflowNode`
465 Job to add to the generic workflow.
466 parent_names : `str` | `list` [`str`], optional
467 Names of jobs that are parents of given job.
468 child_names : `str` | `list` [`str`], optional
469 Names of jobs that are children of given job.
470 """
471 _LOG.debug("job: %s (%s)", job.name, job.label)
472 _LOG.debug("parent_names: %s", parent_names)
473 _LOG.debug("child_names: %s", child_names)
474 if not isinstance(job, GenericWorkflowNode):
475 raise RuntimeError(f"Invalid type for job to be added to GenericWorkflowGraph ({type(job)}).")
476 if self.has_node(job.name):
477 raise RuntimeError(f"Job {job.name} already exists in GenericWorkflowGraph.")
478 super().add_node(job.name, job=job)
479 self.add_job_relationships(parent_names, job.name)
480 self.add_job_relationships(job.name, child_names)
481 if job.node_type in [GenericWorkflowNodeType.PAYLOAD, GenericWorkflowNodeType.LAZY_GROUP]:
482 job = cast(GenericWorkflowJob, job)
483 self.add_executable(job.executable)
484 self._job_labels.add_job(
485 job,
486 [self.get_job(p).label for p in self.predecessors(job.name)],
487 [self.get_job(p).label for p in self.successors(job.name)],
488 )
490 def add_node(self, node_for_adding: GenericWorkflowNode, **attr: Any) -> None:
491 """Override networkx function to call more specific add_job function.
493 Parameters
494 ----------
495 node_for_adding : `lsst.ctrl.bps.GenericWorkflowJob`
496 Job to be added to generic workflow.
497 **attr : `~typing.Any`
498 Needed to match original networkx function, but not used.
499 """
500 self.add_job(node_for_adding)
502 def add_job_relationships(
503 self, parents: str | list[str] | None, children: str | list[str] | None
504 ) -> None:
505 """Add dependencies between parent and child jobs. All parents will
506 be connected to all children.
508 Parameters
509 ----------
510 parents : `str` or `list` [`str`], optional
511 Parent job names.
512 children : `str` or `list` [`str`], optional
513 Children job names.
514 """
515 # Allow this to be a noop if no parents or no children
516 if parents is not None and children is not None:
517 self.add_edges_from(itertools.product(ensure_iterable(parents), ensure_iterable(children)))
518 self._job_labels.add_job_relationships(
519 [self.get_job(n).label for n in ensure_iterable(parents)],
520 [self.get_job(n).label for n in ensure_iterable(children)],
521 )
523 def add_edges_from(self, ebunch_to_add: Iterable[tuple[str, str]], **attr: Any) -> None:
524 """Add several edges between jobs in the generic workflow.
526 Parameters
527 ----------
528 ebunch_to_add : Iterable [`tuple` [`str`, `str`]]
529 Iterable of job name pairs between which a dependency should be
530 saved.
531 **attr : `~typing.Any`
532 Data can be assigned using keyword arguments (not currently used).
533 """
534 for edge_to_add in ebunch_to_add:
535 self.add_edge(edge_to_add[0], edge_to_add[1], **attr)
537 def add_edge(self, u_of_edge: str, v_of_edge: str, **attr: Any) -> None:
538 """Add edge connecting jobs in workflow.
540 Parameters
541 ----------
542 u_of_edge : `str`
543 Name of parent job.
544 v_of_edge : `str`
545 Name of child job.
546 **attr
547 Attributes to save with edge.
548 """
549 if u_of_edge not in self:
550 raise RuntimeError(f"{u_of_edge} not in GenericWorkflow")
551 if v_of_edge not in self:
552 raise RuntimeError(f"{v_of_edge} not in GenericWorkflow")
553 super().add_edge(u_of_edge, v_of_edge, **attr)
555 def get_job(self, job_name: str) -> GenericWorkflowNode:
556 """Retrieve job by name from workflow.
558 Parameters
559 ----------
560 job_name : `str`
561 Name of job to retrieve.
563 Returns
564 -------
565 job : `lsst.ctrl.bps.GenericWorkflowNode`
566 Job matching given job_name.
567 """
568 return self.nodes[job_name]["job"]
570 def del_job(self, job_name: str) -> None:
571 """Delete job from generic workflow leaving connected graph.
573 Parameters
574 ----------
575 job_name : `str`
576 Name of job to delete from workflow.
577 """
578 job = self.get_job(job_name)
580 # Remove from job labels
581 if isinstance(job, GenericWorkflowJob): 581 ↛ 585line 581 didn't jump to line 585 because the condition on line 581 was always true
582 self._job_labels.del_job(job)
584 # Connect all parent jobs to all children jobs.
585 parents = list(self.predecessors(job_name))
586 children = list(self.successors(job_name))
587 self.add_job_relationships(parents, children)
589 # Delete job node (which deletes edges).
590 self.remove_node(job_name)
592 def add_job_inputs(self, job_name: str, files: GenericWorkflowFile | list[GenericWorkflowFile]) -> None:
593 """Add files as inputs to specified job.
595 Parameters
596 ----------
597 job_name : `str`
598 Name of job to which inputs should be added.
599 files : `lsst.ctrl.bps.GenericWorkflowFile` or \
600 `list` [`lsst.ctrl.bps.GenericWorkflowFile`]
601 File object(s) to be added as inputs to the specified job.
602 """
603 self._inputs.setdefault(job_name, [])
604 for file in ensure_iterable(files):
605 # Save the central copy
606 if file.name not in self._files:
607 self._files[file.name] = file
609 # Save the job reference to the file
610 self._inputs[job_name].append(file)
612 def get_file(self, name: str) -> GenericWorkflowFile:
613 """Retrieve a file object by name.
615 Parameters
616 ----------
617 name : `str`
618 Name of file object.
620 Returns
621 -------
622 gwfile : `lsst.ctrl.bps.GenericWorkflowFile`
623 File matching given name.
624 """
625 return self._files[name]
627 def add_file(self, gwfile: GenericWorkflowFile) -> None:
628 """Add file object.
630 Parameters
631 ----------
632 gwfile : `lsst.ctrl.bps.GenericWorkflowFile`
633 File object to add to workflow.
634 """
635 if gwfile.name not in self._files:
636 self._files[gwfile.name] = gwfile
637 else:
638 _LOG.debug("Skipped add_file for existing file %s", gwfile.name)
640 @overload
641 def get_job_inputs( 641 ↛ exitline 641 didn't return from function 'get_job_inputs' because
642 self, job_name: str, data: Literal[False], transfer_only: bool = False
643 ) -> list[str]: ...
645 @overload
646 def get_job_inputs( 646 ↛ exitline 646 didn't return from function 'get_job_inputs' because
647 self, job_name: str, data: Literal[True], transfer_only: bool = False
648 ) -> list[GenericWorkflowFile]: ...
650 def get_job_inputs(
651 self, job_name: str, data: bool = True, transfer_only: bool = False
652 ) -> list[GenericWorkflowFile] | list[str]:
653 """Return the input files for the given job.
655 Parameters
656 ----------
657 job_name : `str`
658 Name of the job.
659 data : `bool`, optional
660 Whether to return the file data as well as the file object name.
661 transfer_only : `bool`, optional
662 Whether to only return files for which a workflow management system
663 would be responsible for transferring.
665 Returns
666 -------
667 inputs : `list` [`lsst.ctrl.bps.GenericWorkflowFile`] or `list` [`str`]
668 Input files for the given job. If no input files for the job,
669 returns an empty list.
670 """
671 inputs: list[Any] = [] # Any for mypy to allow the different append lines.
672 if job_name in self._inputs:
673 for gwfile in self._inputs[job_name]:
674 if not transfer_only or gwfile.wms_transfer:
675 if not data:
676 inputs.append(gwfile.name)
677 else:
678 inputs.append(gwfile)
679 return inputs
681 def add_job_outputs(self, job_name: str, files: list[GenericWorkflowFile]) -> None:
682 """Add output files to a job.
684 Parameters
685 ----------
686 job_name : `str`
687 Name of job to which the files should be added as outputs.
688 files : `list` [`lsst.ctrl.bps.GenericWorkflowFile`]
689 File objects to be added as outputs for specified job.
690 """
691 self._outputs.setdefault(job_name, [])
693 for file_ in ensure_iterable(files):
694 # Save the central copy
695 if file_.name not in self._files:
696 self._files[file_.name] = file_
698 # Save the job reference to the file
699 self._outputs[job_name].append(file_)
701 @overload
702 def get_job_outputs( 702 ↛ exitline 702 didn't return from function 'get_job_outputs' because
703 self, job_name: str, data: Literal[False], transfer_only: bool = False
704 ) -> list[str]: ...
706 @overload
707 def get_job_outputs( 707 ↛ exitline 707 didn't return from function 'get_job_outputs' because
708 self, job_name: str, data: Literal[True], transfer_only: bool = False
709 ) -> list[GenericWorkflowFile]: ...
711 def get_job_outputs(
712 self, job_name: str, data: bool = True, transfer_only: bool = False
713 ) -> list[GenericWorkflowFile] | list[str]:
714 """Return the output files for the given job.
716 Parameters
717 ----------
718 job_name : `str`
719 Name of the job.
720 data : `bool`
721 Whether to return the file data as well as the file object name.
722 It defaults to `True` thus returning file data as well.
723 transfer_only : `bool`
724 Whether to only return files for which a workflow management system
725 would be responsible for transferring. It defaults to `False` thus
726 returning all output files.
728 Returns
729 -------
730 outputs : `list` [`lsst.ctrl.bps.GenericWorkflowFile`] or \
731 `list` [`str`]
732 Output files for the given job. If no output files for the job,
733 returns an empty list.
734 """
735 outputs: list[Any] = [] # Any for mypy to allow the different append lines.
736 if not data: 736 ↛ 737line 736 didn't jump to line 737 because the condition on line 736 was never true
737 outputs = cast(list[str], outputs)
738 else:
739 outputs = cast(list[GenericWorkflowFile], outputs)
741 if job_name in self._outputs:
742 for gwfile in self._outputs[job_name]:
743 if not transfer_only or gwfile.wms_transfer: 743 ↛ 742line 743 didn't jump to line 742 because the condition on line 743 was always true
744 if not data: 744 ↛ 745line 744 didn't jump to line 745 because the condition on line 744 was never true
745 outputs.append(gwfile.name)
746 else:
747 outputs.append(gwfile)
748 return outputs
750 def draw(self, stream: str | IO[str], format_: str = "dot") -> None:
751 """Output generic workflow in a visualization format.
753 Parameters
754 ----------
755 stream : `str` or `io.BufferedIOBase`
756 Stream to which the visualization should be written.
757 format_ : `str`, optional
758 Which visualization format to use. It defaults to the format for
759 the dot program.
760 """
761 draw_funcs = {"dot": draw_networkx_dot}
762 if format_ in draw_funcs: 762 ↛ 763line 762 didn't jump to line 763 because the condition on line 762 was never true
763 draw_funcs[format_](self, stream)
764 else:
765 raise RuntimeError(f"Unknown draw format ({format_})")
767 def save(self, stream: str | IO[bytes], format_: str = "pickle") -> None:
768 """Save the generic workflow in a format that is loadable.
770 Parameters
771 ----------
772 stream : `str` or `io.BufferedIOBase`
773 Stream to pass to the format-specific writer. Accepts anything
774 that the writer accepts.
775 format_ : `str`, optional
776 Format in which to write the data. It defaults to pickle format.
777 """
778 if format_ == "pickle":
779 stream = cast(BinaryIO, stream)
780 pickle.dump(self, stream)
781 else:
782 raise RuntimeError(f"Unknown format ({format_})")
784 @classmethod
785 def load(cls, stream: str | IO[bytes], format_: str = "pickle") -> "GenericWorkflow":
786 """Load a GenericWorkflow from the given stream.
788 Parameters
789 ----------
790 stream : `str` or `io.BufferedIOBase`
791 Stream to pass to the format-specific loader. Accepts anything that
792 the loader accepts.
793 format_ : `str`, optional
794 Format of data to expect when loading from stream. It defaults
795 to pickle format.
797 Returns
798 -------
799 generic_workflow : `lsst.ctrl.bps.GenericWorkflow`
800 Generic workflow loaded from the given stream.
801 """
802 if format_ == "pickle":
803 stream = cast(BinaryIO, stream)
804 object_ = pickle.load(stream)
805 assert isinstance(object_, GenericWorkflow) # for mypy
806 return object_
808 raise RuntimeError(f"Unknown format ({format_})")
810 def validate(self) -> None:
811 """Run checks to ensure that the generic workflow graph is valid."""
812 # Make sure a directed acyclic graph
813 assert is_directed_acyclic_graph(self)
815 def add_workflow_source(self, workflow: "GenericWorkflow") -> None:
816 """Add given workflow as new source to this workflow.
818 Parameters
819 ----------
820 workflow : `lsst.ctrl.bps.GenericWorkflow`
821 The given workflow.
822 """
823 # Find source nodes in self.
824 self_sources = [n for n in self if self.in_degree(n) == 0]
825 _LOG.debug("self_sources = %s", self_sources)
827 # Find sink nodes of workflow.
828 new_sinks = [n for n in workflow if workflow.out_degree(n) == 0]
829 _LOG.debug("new sinks = %s", new_sinks)
831 # Add new workflow nodes to self graph and make new edges.
832 self.add_nodes_from(workflow.nodes(data=True))
833 self.add_edges_from(workflow.edges())
834 for source in self_sources:
835 for sink in new_sinks:
836 self.add_edge(sink, source)
838 # Add separately stored info
839 for job_name in workflow:
840 job = self.get_job(job_name)
841 # Add job labels
842 if isinstance(job, GenericWorkflowJob): 842 ↛ 852line 842 didn't jump to line 852 because the condition on line 842 was always true
843 self._job_labels.add_job(
844 job,
845 [self.get_job(p).label for p in self.predecessors(job.name)],
846 [self.get_job(p).label for p in self.successors(job.name)],
847 )
848 # Executables are stored separately so copy them.
849 self.add_executable(job.executable)
851 # Files are stored separately so copy them.
852 self.add_job_inputs(job_name, workflow.get_job_inputs(job_name, data=True))
853 self.add_job_outputs(job_name, workflow.get_job_outputs(job_name, data=True))
855 def add_final(self, final: "GenericWorkflowJob | GenericWorkflow") -> None:
856 """Add special final job/workflow to the generic workflow.
858 Parameters
859 ----------
860 final : `lsst.ctrl.bps.GenericWorkflowJob` or \
861 `lsst.ctrl.bps.GenericWorkflow`
862 Information needed to execute the special final job(s), the
863 job(s) to be executed after all jobs that can be executed
864 have been executed regardless of exit status of any of the
865 jobs.
866 """
867 if not isinstance(final, GenericWorkflowJob) and not isinstance(final, GenericWorkflow): 867 ↛ 868line 867 didn't jump to line 868 because the condition on line 867 was never true
868 raise TypeError("Invalid type for GenericWorkflow final ({type(final)})")
870 self._final = final
871 if isinstance(final, GenericWorkflowJob): 871 ↛ exitline 871 didn't return from function 'add_final' because the condition on line 871 was always true
872 self.add_executable(final.executable)
874 def get_final(self) -> "GenericWorkflowJob | GenericWorkflow | None":
875 """Return job/workflow to be executed after all jobs that can be
876 executed have been executed regardless of exit status of any of
877 the jobs.
879 Returns
880 -------
881 final : `lsst.ctrl.bps.GenericWorkflowJob` or \
882 `lsst.ctrl.bps.GenericWorkflow`
883 Information needed to execute final job(s).
884 """
885 return self._final
887 def add_executable(self, executable: GenericWorkflowExec | None) -> None:
888 """Add executable to workflow's list of executables.
890 Parameters
891 ----------
892 executable : `lsst.ctrl.bps.GenericWorkflowExec`
893 Executable object to be added to workflow.
894 """
895 if executable is not None:
896 self._executables[executable.name] = executable
897 else:
898 _LOG.warning("executable not specified (None); cannot add to the workflow's list of executables")
900 @overload
901 def get_executables(self, data: Literal[False], transfer_only: bool = False) -> list[str]: ... 901 ↛ exitline 901 didn't return from function 'get_executables' because
903 @overload
904 def get_executables( 904 ↛ exitline 904 didn't return from function 'get_executables' because
905 self, data: Literal[True], transfer_only: bool = False
906 ) -> list[GenericWorkflowExec]: ...
908 def get_executables(
909 self, data: bool = False, transfer_only: bool = True
910 ) -> list[GenericWorkflowExec] | list[str]:
911 """Retrieve executables from generic workflow.
913 Parameters
914 ----------
915 data : `bool`, optional
916 Whether to return the executable data as well as the exec object
917 name (The defaults is False).
918 transfer_only : `bool`, optional
919 Whether to only return executables for which transfer_executable
920 is True.
922 Returns
923 -------
924 execs : `list` [`lsst.ctrl.bps.GenericWorkflowExec`] or `list` [`str`]
925 Filtered executable names or objects from generic workflow.
926 """
927 execs: list[Any] = [] # This and cast lines for mypy
928 if not data:
929 execs = cast(list[str], execs)
930 else:
931 execs = cast(list[GenericWorkflowExec], execs)
933 for name, executable in self._executables.items():
934 if not transfer_only or executable.transfer_executable: 934 ↛ 933line 934 didn't jump to line 933 because the condition on line 934 was always true
935 if not data:
936 execs.append(name)
937 else:
938 execs.append(executable)
939 return execs
941 def get_jobs_by_label(self, label: str) -> list[GenericWorkflowJob]:
942 """Retrieve jobs by label from workflow.
944 Parameters
945 ----------
946 label : `str`
947 Label of jobs to retrieve.
949 Returns
950 -------
951 jobs : list[`lsst.ctrl.bps.GenericWorkflowNode`]
952 Jobs having given label.
953 """
954 return self._job_labels.get_jobs_by_label(label)
956 def _check_job_ordering_config(self, ordering_config: dict[str, Any]) -> dict[str, DiGraph]:
957 """Check configuration related to job ordering.
959 Parameters
960 ----------
961 ordering_config : `dict` [`str`, `~typing.Any`]
962 Job ordering configuration to check.
964 Returns
965 -------
966 group_to_label_subgraph : `dict` [`str`, `network.DiGraph`]
967 Mapping of group name to a graph of the job labels in the group.
968 """
969 group_to_label_subgraph = {}
970 job_label_to_group: dict[str, str] = {} # Checking label appears only in one group
971 for group, group_vals in ordering_config.items():
972 implementation = group_vals.get("implementation", "group")
973 if implementation not in ["noop", "group"]:
974 raise RuntimeError(f"Invalid implementation for {group}: {implementation}")
975 ordering_type = group_vals.get("ordering_type", "sort")
976 if ordering_type != "sort":
977 raise RuntimeError(f"Invalid ordering_type for {group}: {ordering_type}")
978 if "dimensions" not in group_vals:
979 raise KeyError(f"Missing dimensions entry in ordering group {group}")
981 job_labels = [x.strip() for x in group_vals["labels"].split(",")]
982 _LOG.debug("group %s: job_labels=%s", group, job_labels)
983 unused_labels = []
984 for job_label in job_labels:
985 if job_label not in self._job_labels.labels:
986 unused_labels.append(job_label)
987 elif job_label in job_label_to_group:
988 raise RuntimeError(
989 f"Job label {job_label} appears in more than one job ordering group "
990 f"({group} {job_label_to_group[job_label]})"
991 )
992 else:
993 job_label_to_group[job_label] = group
995 if unused_labels:
996 _LOG.info("Workflow job labels = %s", ",".join(self._job_labels.labels))
997 _LOG.warning(
998 "Job label(s) (%s) from job ordering group %s does not exist in workflow.",
999 ",".join(unused_labels),
1000 group,
1001 )
1003 label_subgraph = self._job_labels.subgraph(job_labels)
1004 group_to_label_subgraph[group] = label_subgraph
1006 return group_to_label_subgraph
1008 def _group_jobs_by_values(
1009 self, group: str, group_config: dict[str, Any], label_subgraph: DiGraph
1010 ) -> dict[tuple[Any, ...], list[str]]:
1011 """Create job mapping of special sortable dimension key
1012 to job name by comparing dimension values.
1014 Parameters
1015 ----------
1016 group : `str`
1017 Name of group for which creating mapping.
1018 group_config : `dict` [`str`, `~typing.Any`]
1019 Config for group for which creating mapping.
1020 label_subgraph : `networkx.DiGraph`
1021 The graph of job labels to be used in mapping.
1023 Returns
1024 -------
1025 dims_to_jobs : `dict` [`tuple` [`~typing.Any`, ...], `list` [`str`]]
1026 Mapping of dimensions to job names.
1027 """
1028 dims_to_jobs: dict[tuple[Any, ...], list[str]] = {}
1029 for job_label in label_subgraph:
1030 jobs = self.get_jobs_by_label(job_label)
1031 for job in jobs:
1032 job_dim_values = subset_dimension_values(
1033 f"Job {job.name}",
1034 f"order group {group}",
1035 group_config["dimensions"],
1036 job.tags,
1037 group_config.get("equalDimensions", None),
1038 )
1039 job_dims = []
1040 for dim in [d.strip() for d in group_config["dimensions"].split(",")]:
1041 job_dims.append(job_dim_values[dim])
1042 dims_job_list = dims_to_jobs.setdefault(tuple(job_dims), [])
1043 dims_job_list.append(job.name)
1044 return dims_to_jobs
1046 def _group_jobs_by_dependencies(
1047 self, group: str, group_config: dict[str, Any], label_subgraph: DiGraph
1048 ) -> dict[tuple[Any, ...], list[str]]:
1049 """Create job mapping of special sortable dimension key
1050 to job name by following dependencies.
1052 Parameters
1053 ----------
1054 group : `str`
1055 Name of group for which creating mapping.
1056 group_config : `dict` [`str`, `~typing.Any`]
1057 Config for group for which creating mapping.
1058 label_subgraph : `networkx.DiGraph`
1059 The graph of job labels to be used in mapping.
1061 Returns
1062 -------
1063 dims_to_jobs : `dict` [`tuple` [`~typing.Any`, ...], `list` [`str`]]
1064 Mapping of dimensions to job names.
1065 """
1066 method = group_config["findDependencyMethod"]
1067 dim_labels = list(topological_sort(label_subgraph))
1068 match method:
1069 case "source":
1070 find_potential_jobs = self.successors
1071 case "sink":
1072 find_potential_jobs = self.predecessors
1073 dim_labels.reverse()
1074 case _:
1075 raise RuntimeError(f"Invalid findDependencyMethod ({method})")
1077 jobs_seen: set[str] = set()
1078 dims_to_jobs: dict[tuple[Any, ...], list[str]] = {}
1079 for label in dim_labels:
1080 jobs = self.get_jobs_by_label(label)
1081 for job in jobs:
1082 if job.name not in jobs_seen:
1083 jobs_seen.add(job.name)
1084 job_dim_values = subset_dimension_values(
1085 f"Job {job.name}",
1086 f"order group {group}",
1087 group_config["dimensions"],
1088 job.tags,
1089 group_config.get("equalDimensions", None),
1090 )
1091 job_dims = []
1092 for dim in [d.strip() for d in group_config["dimensions"].split(",")]:
1093 job_dims.append(job_dim_values[dim])
1094 dims_job_list = dims_to_jobs.setdefault(tuple(job_dims), [])
1095 dims_job_list.append(job.name)
1096 # Use dependencies to find other quantum to add
1097 # Note: in testing, using the following code was faster
1098 # than using networkx descendants and ancestors functions
1099 # While traversing the subgraph, nodes may appear
1100 # repeatedly in potential_jobs.
1101 jobs_to_use = [job]
1102 while jobs_to_use:
1103 job_to_use = jobs_to_use.pop()
1105 potential_job_names = find_potential_jobs(job_to_use.name)
1106 for potential_job_name in potential_job_names:
1107 potential_job = cast(GenericWorkflowJob, self.get_job(potential_job_name))
1108 if potential_job.label in label_subgraph:
1109 if potential_job.name not in dims_job_list:
1110 _LOG.debug(
1111 "Adding potential job %s (%s) to group %s",
1112 potential_job.name,
1113 potential_job.label,
1114 group,
1115 )
1116 dims_job_list.append(potential_job.name)
1117 jobs_to_use.append(potential_job)
1118 jobs_seen.add(potential_job.name)
1119 else:
1120 _LOG.debug(
1121 "label (%s) not in ordered_tasks. Not adding potential quantum %s",
1122 potential_job.label,
1123 potential_job.name,
1124 )
1125 return dims_to_jobs
1127 def _update_by_group_sort(
1128 self,
1129 group: str,
1130 dims_to_jobs: dict[tuple[Any, ...], list[str]],
1131 blocking: bool = False,
1132 ) -> None:
1133 """Update portion of workflow for special job ordering using sort.
1135 Parameters
1136 ----------
1137 group : `str`
1138 Ordering group label used for job name and messages.
1139 dims_to_jobs: `dict` [`tuple` [`~typing.Any`, ...], `list` [`str`]]
1140 Mapping of special dimension keys to workflow job names.
1141 The sort for the special ordering is over the keys.
1142 blocking: `bool`
1143 Whether a failure in a group blocks execution of remaining groups.
1144 """
1145 group_job_names: list[str] = []
1146 for dim_key, job_list in sorted(dims_to_jobs.items()):
1147 _LOG.debug("group %s: dim_key=%s", group, dim_key)
1148 group_job_name = f"group_{group}_{'=='.join([str(dk) for dk in dim_key])}"
1149 prev_name = group_job_names[-1] if group_job_names else None
1150 group_job_names.append(group_job_name)
1152 self._replace_subgraph_with_job_group(group_job_name, group, job_list, blocking)
1154 # ordering between groups
1155 if prev_name:
1156 self.add_edge(prev_name, group_job_name)
1158 def _replace_subgraph_with_job_group(
1159 self, group_name: str, group_label: str, job_list: list[str], blocking: bool
1160 ) -> None:
1161 """Update portion of workflow for special job ordering using groups.
1163 Parameters
1164 ----------
1165 group_name : `str`
1166 Ordering group name.
1167 group_label : `str`
1168 Ordering group label.
1169 job_list: `list` [`str`]
1170 List of job names to put in the group
1171 blocking: `bool`
1172 Whether a failure in a group blocks execution of remaining groups.
1173 """
1174 job_group = GenericWorkflowGroup(group_name, group_label, blocking=blocking)
1175 self.add_node(job_group)
1177 # Add jobs, files, and executables first
1178 # then add edges later to avoid order issues
1179 for job_name in job_list:
1180 job = cast(GenericWorkflowJob, self.get_job(job_name))
1181 job_group.add_job(job)
1182 files = self.get_job_inputs(job_name, data=True)
1183 job_group.add_job_inputs(job_name, files)
1184 files = self.get_job_outputs(job_name, data=True)
1185 job_group.add_job_outputs(job_name, files)
1186 job_group.add_executable(job.executable)
1188 # Can't remove edges while looping through edge view,
1189 # so save to remove after loops
1190 edges_to_remove: list[tuple[str, str]] = []
1191 for job_name in job_list:
1192 in_edges = self.in_edges(job_name)
1193 for u, v in in_edges:
1194 if u in job_list:
1195 job_group.add_edge(u, v)
1196 else:
1197 self.add_edge(u, group_name)
1198 edges_to_remove.append((u, v))
1200 out_edges = self.out_edges(job_name)
1201 for u, v in out_edges:
1202 if v in job_list:
1203 job_group.add_edge(u, v)
1204 else:
1205 self.add_edge(group_name, v)
1206 edges_to_remove.append((u, v))
1208 # Remove edges collected earlier
1209 self.remove_edges_from(edges_to_remove)
1211 # Remove nodes from main GenericWorkflow
1212 self.remove_nodes_from(job_list)
1214 def add_special_job_ordering(self, ordering: dict[str, Any]) -> None:
1215 """Add special nodes and dependencies to enforce given ordering.
1217 Parameters
1218 ----------
1219 ordering : `dict` [`str`, `~typing.Any`]
1220 Description of the job ordering to enforce.
1221 """
1222 group_to_label_subgraph = self._check_job_ordering_config(ordering)
1224 for group, group_vals in ordering.items():
1225 if "findDependencyMethod" in group_vals:
1226 job_grouping_func = self._group_jobs_by_dependencies
1227 else:
1228 job_grouping_func = self._group_jobs_by_values
1230 dims = [x.strip() for x in group_vals["dimensions"].split(",")]
1231 _LOG.debug("group %s: dims=%s", group, dims)
1232 job_groups = job_grouping_func(group, group_vals, group_to_label_subgraph[group])
1234 # Update the workflow
1235 implementation = group_vals.get("implementation", "group")
1236 ordering_type = group_vals.get("ordering_type", "sort")
1237 match (implementation, ordering_type):
1238 case ("noop", "sort"):
1239 self._update_by_noop_sort(group, job_groups)
1240 case ("group", "sort"): 1240 ↛ 1243line 1240 didn't jump to line 1243 because the pattern on line 1240 always matched
1241 blocking = group_vals.get("blocking", False)
1242 self._update_by_group_sort(group, job_groups, blocking)
1243 case _:
1244 raise RuntimeError(
1245 f"Invalid implementation, ordering_type pair for group ({implementation},"
1246 f" {ordering_type})"
1247 )
1249 def _update_by_noop_sort(self, group: str, dims_to_jobs: dict[tuple[Any, ...], list[str]]) -> None:
1250 """Update portion of workflow for special ordering of jobs using sort.
1252 Parameters
1253 ----------
1254 group : `str`
1255 Ordering group label used for job name and messages.
1256 dims_to_jobs: `dict` [`tuple`[`str`,...], `list` [`str`]]
1257 Mapping of special dimension keys to workflow job names.
1258 The sort for the special ordering is over the keys.
1259 """
1260 noop_names: list[str] = []
1261 prev_name: str | None = None
1262 for dim_key, job_list in sorted(dims_to_jobs.items()):
1263 _LOG.debug("group %s: dim_key=%s", group, dim_key)
1264 noop_name = f"noop_{group}_{'=='.join([str(dk) for dk in dim_key])}"
1265 if noop_names:
1266 prev_name = noop_names[-1]
1267 noop_names.append(noop_name)
1269 self._update_single_noop(job_list, noop_name, group, prev_name)
1271 # As implemented, loop adds one last NOOP job with
1272 # 0 children. Remove it as not useful.
1273 assert self.out_degree(noop_names[-1]) == 0
1274 self.remove_node(noop_names[-1])
1276 def _update_single_noop(
1277 self, job_list: list[str], order_node_name: str, order_label: str, prev_order_name: str | None
1278 ) -> None:
1279 """Update the workflow around the single job making special NOOP
1280 jobs when necessary.
1282 Parameters
1283 ----------
1284 job_list : `list` [`str`]
1285 Current jobs involved in special ordering
1286 order_node_name : `str`
1287 Name for the order NOOP job. If it does not exist in workflow, a
1288 new NOOP job with this name will be created and added.
1289 order_label : `str`
1290 Label for the order NOOP job.
1291 prev_order_name: `str` or None
1292 Name of the previous order NOOP job used to add edge as predecessor
1293 of current job.
1294 """
1295 if order_node_name not in self: 1295 ↛ 1300line 1295 didn't jump to line 1300 because the condition on line 1295 was always true
1296 _LOG.debug("Adding new ordering node %s", order_node_name)
1297 order_node = GenericWorkflowNoopJob(order_node_name, order_label)
1298 self.add_job(order_node)
1300 subgraph = DiGraph(self).subgraph(job_list)
1301 sinks = [n for n in job_list if subgraph.out_degree(n) == 0]
1302 for job_name in sinks:
1303 self.add_edge(job_name, order_node_name)
1305 if prev_order_name:
1306 sources = [n for n in job_list if subgraph.in_degree(n) == 0]
1307 for job_name in sources:
1308 _LOG.debug("Adding edge %s to %s", prev_order_name, job_name)
1309 self.add_edge(prev_order_name, job_name)
1312@dataclasses.dataclass(slots=True)
1313class GenericWorkflowGroup(GenericWorkflowNode, GenericWorkflow):
1314 """Node representing a group of jobs. Used for special dependencies.
1316 Parameters
1317 ----------
1318 name : `str`
1319 Name of node. Must be unique within workflow.
1320 label : `str`
1321 Primary user-facing label for job. Does not need to be unique and
1322 may be used for summary reports or to group nodes.
1323 blocking : `bool`
1324 Whether a failure inside group prunes executions of remaining groups.
1325 """
1327 blocking: bool = False
1328 """Whether a failure inside group prunes executions of remaining groups."""
1330 @property
1331 def node_type(self) -> GenericWorkflowNodeType:
1332 """Indicate this is a group job."""
1333 return GenericWorkflowNodeType.GROUP
1335 def __init__(self, name: str, label: str, blocking: bool = False) -> None:
1336 """Initialize each parent class."""
1337 _LOG.debug("%s %s %s", name, label, blocking)
1338 GenericWorkflowNode.__init__(self, name, label)
1339 GenericWorkflow.__init__(self, name)
1340 self.blocking = blocking
1343@dataclasses.dataclass(slots=True)
1344class GenericWorkflowLazyGroup(GenericWorkflowJob):
1345 """Node representing a group of jobs to be generated when run."""
1347 # Docstring inherited.
1349 @property
1350 def node_type(self) -> GenericWorkflowNodeType:
1351 """Indicate this is a lazy group job."""
1352 return GenericWorkflowNodeType.LAZY_GROUP
1355class GenericWorkflowLabels:
1356 """Label-oriented representation of the GenericWorkflowJobs."""
1358 def __init__(self) -> None:
1359 self._label_graph = DiGraph() # Dependency graph of job labels
1360 self._label_to_jobs: defaultdict[str, list[GenericWorkflowJob]] = defaultdict(
1361 list
1362 ) # mapping job label to list of GenericWorkflowJob
1364 @property
1365 def labels(self) -> list[str]:
1366 """List of job labels (`list` [`str`], read-only)."""
1367 return list(topological_sort(self._label_graph))
1369 @property
1370 def job_counts(self) -> Counter[str]:
1371 """Count of jobs per job label (`collections.Counter`)."""
1372 return Counter({label: len(self._label_to_jobs[label]) for label in self.labels})
1374 def get_jobs_by_label(self, label: str) -> list[GenericWorkflowJob]:
1375 """Retrieve jobs by label from workflow.
1377 Parameters
1378 ----------
1379 label : `str`
1380 Label of jobs to retrieve.
1382 Returns
1383 -------
1384 jobs : list[`lsst.ctrl.bps.GenericWorkflowJob`]
1385 Jobs having given label.
1386 """
1387 return self._label_to_jobs[label]
1389 def add_job(self, job: GenericWorkflowJob, parent_labels: list[str], child_labels: list[str]) -> None:
1390 """Add job's label to labels.
1392 Parameters
1393 ----------
1394 job : `lsst.ctrl.bps.GenericWorkflowJob`
1395 The job to add to the job labels.
1396 parent_labels : `list` [`str`]
1397 Parent job labels.
1398 child_labels : `list` [`str`]
1399 Children job labels.
1400 """
1401 _LOG.debug("job: %s (%s)", job.name, job.label)
1402 _LOG.debug("parent_labels: %s", parent_labels)
1403 _LOG.debug("child_labels: %s", child_labels)
1404 self._label_to_jobs[job.label].append(job)
1405 self._label_graph.add_node(job.label)
1406 for parent in parent_labels:
1407 self._label_graph.add_edge(parent, job.label)
1408 for child in child_labels:
1409 self._label_graph.add_edge(job.label, child)
1411 def add_job_relationships(self, parent_labels: list[str], children_labels: list[str]) -> None:
1412 """Add dependencies between parent and child job labels.
1413 All parents will be connected to all children.
1415 Parameters
1416 ----------
1417 parent_labels : `list` [`str`]
1418 Parent job labels.
1419 children_labels : `list` [`str`]
1420 Children job labels.
1421 """
1422 # Since labels, must ensure not adding edge from label to itself.
1423 edges = [
1424 e
1425 for e in itertools.product(ensure_iterable(parent_labels), ensure_iterable(children_labels))
1426 if e[0] != e[1]
1427 ]
1429 self._label_graph.add_edges_from(edges)
1431 def del_job(self, job: GenericWorkflowJob) -> None:
1432 """Delete job and its label from job labels.
1434 Parameters
1435 ----------
1436 job : `lsst.ctrl.bps.GenericWorkflowJob`
1437 The job to delete from the job labels.
1438 """
1439 self._label_to_jobs[job.label].remove(job)
1440 # Don't leave keys around if removed last job
1441 if not self._label_to_jobs[job.label]:
1442 del self._label_to_jobs[job.label]
1444 parents = self._label_graph.predecessors(job.label)
1445 children = self._label_graph.successors(job.label)
1446 self._label_graph.remove_node(job.label)
1447 self._label_graph.add_edges_from(
1448 itertools.product(ensure_iterable(parents), ensure_iterable(children))
1449 )
1451 def subgraph(self, labels: Iterable[str]) -> DiGraph:
1452 """Create subgraph of workflow label graph with given labels.
1454 Parameters
1455 ----------
1456 labels : Iterable [`str`]
1457 Labels to appear in subgraph.
1459 Returns
1460 -------
1461 subgraph : `networkx.DiGraph`
1462 Subgraph of workflow label graph with given labels.
1463 """
1464 return self._label_graph.subgraph(labels)