dags

DAG

DAG(name='workflow', formatter=None, *args, **kwargs)

Bases: DAG

Defines a DAGMan workflow including the execution graph and configuration.

Parameters:

Name	Type	Description	Default
name	str	The name of the DAG workflow, used for files written to disk and for DAG submission when calling write() and submit(). Defaults to "workflow".	'workflow'
formatter	NodeNameFormatter	Defines how the node names are defined and formatted. Defaults to a hex-based formatter with 5 digits.	None
*args		Any positional arguments that htcondor.dags.DAG accepts	()
**kwargs		Any keyword arguments that htcondor.dags.DAG accepts	{}

Source code in ezdag/dags.py

def __init__(
    self,
    name: str = "workflow",
    formatter: dags.NodeNameFormatter | None = None,
    *args,
    **kwargs,
) -> None:
    super().__init__(*args, **kwargs)
    self.name = name
    self._node_layers: dict[str, dags.NodeLayer] = {}
    self._ordered_layers: list[dags.NodeLayer] = []
    self._layers: dict[str, Layer] = {}
    self._provides: dict[str, tuple[str, int]] = {}
    self._all_edges: dict[str, dict[str, set[tuple[int, int]]]] = {}
    if formatter:
        self.formatter = formatter
    else:
        self.formatter = HexFormatter()
    self._dag_path: str | None = None
    self._layer_count: Counter[str] = Counter()

attach

attach(layer)

Attach a layer of related job nodes to this DAG.

Parameters:

Name	Type	Description	Default
layer	Layer	The layer to attach.	required

Source code in ezdag/dags.py

def attach(self, layer: Layer) -> None:
    """Attach a layer of related job nodes to this DAG.

    Parameters
    ----------
    layer
        The layer to attach.

    """
    self._layer_count[layer.name] += 1

    # layer naming - append number for duplicate layers
    layer_config = layer.config(self.formatter)
    if self._layer_count[layer.name] > 1:
        layer_name = f"{layer.name}{self._layer_count[layer.name]}"
        layer_config["name"] = layer_name
    else:
        layer_name = layer.name
    self._layers[layer_name] = layer

    # determine parent-child relationships and connect accordingly
    all_edges = defaultdict(set)
    if layer.has_dependencies:
        # determine edges
        for child_idx, node in enumerate(layer.nodes):
            for input_ in node.requires:
                if input_ in self._provides:
                    parent_name, parent_idx = self._provides[input_]
                    all_edges[parent_name].add((parent_idx, child_idx))
        self._all_edges[layer_name] = all_edges

        if not all_edges:
            node_layer = self.layer(**layer_config)
            self._node_layers[layer_name] = node_layer
            self._ordered_layers.append(node_layer)

        # determine edge type and connect
        for num, (parent, edges) in enumerate(all_edges.items()):
            edge = self._get_edge_type(parent, layer_name, edges)
            if num == 0:
                node_layer = self._node_layers[parent].child_layer(
                    **layer_config, edge=edge
                )
                self._node_layers[layer_name] = node_layer
                self._ordered_layers.append(node_layer)
            else:
                self._node_layers[layer_name].add_parents(
                    self._node_layers[parent], edge=edge
                )

    else:
        node_layer = self.layer(**layer_config)
        self._node_layers[layer_name] = node_layer
        self._ordered_layers.append(node_layer)

    # register any data products the layer provides
    for idx, node in enumerate(layer.nodes):
        for output in node.provides:
            self._provides[output] = (layer_name, idx)

submit

submit(path=None, *, write_script=False, **kwargs)

Submit the DAG via HTCondor.

If the DAG has not already been written to disk, do so as well. This is equivalent to calling write() prior to submission, making use of the path and write_script arguments for doing so. See DAG.write for more information.

Parameters:

Name	Type	Description	Default
path	Path	The directory to write the DAG files to. Defaults to the current working directory.	None
write_script	bool	Also write out the list of commands for each node to disk. Defaults to false.	False
**kwargs		Any keyword arguments that condor_submit_dag accepts. See htcondor.Submit.from_dag for more information.	{}

Returns:

Type	Description
SubmitResult	The submit result containing the cluster ID and ClassAd of the submitted DAG.

Source code in ezdag/dags.py

def submit(
    self, path: Path | None = None, *, write_script: bool = False, **kwargs
) -> htcondor.SubmitResult:
    """Submit the DAG via HTCondor.

    If the DAG has not already been written to disk, do so as well.
    This is equivalent to calling write() prior to submission, making
    use of the `path` and `write_script` arguments for doing so. See
    DAG.write for more information.

    Parameters
    ----------
    path : Path
        The directory to write the DAG files to. Defaults to the current working
        directory.
    write_script : bool
        Also write out the list of commands for each node to disk. Defaults
        to false.
    **kwargs
        Any keyword arguments that `condor_submit_dag` accepts. See
        [htcondor.Submit.from_dag](https://htcondor.readthedocs.io/en/latest/apis/python-bindings/api/version2/htcondor2/submit.html#htcondor2.Submit.from_dag)
        for more information.

    Returns
    -------
    htcondor.SubmitResult
        The submit result containing the cluster ID and ClassAd of the
        submitted DAG.

    """
    if not path:
        path = Path.cwd()

    # write DAG to disk if not already done
    if not self._dag_path:
        self.write(path, write_script=write_script)
        self._dag_path = str(path / f"{self.name}.dag")

    # submit the DAG
    submit_kwargs = {"UseDagDir": True, **kwargs}
    dag_submit = htcondor.Submit.from_dag(self._dag_path, submit_kwargs)
    return htcondor.Schedd().submit(dag_submit)

visualize

visualize(path=None, image_format='svg')

Visualize a DAG.

Note: This requires graphviz to be installed.

Parameters:

Name	Type	Description	Default
path	Path	The directory to write the graph diagram to. Defaults to the current working directory as {workflow}.png.	None
image_format	str	The output file format to use. Defaults to 'svg'.	'svg'

Source code in ezdag/dags.py

def visualize(self, path: Path | None = None, image_format: str = "svg") -> None:
    """Visualize a DAG.

    Note: This requires graphviz to be installed.

    Parameters
    ----------
    path : Path
        The directory to write the graph diagram to. Defaults to the
        current working directory as {workflow}.png.
    image_format : str
        The output file format to use. Defaults to 'svg'.

    """
    try:
        import graphviz
    except ImportError as exc:
        msg = "graphviz needs to be installed to visualize DAGs"
        raise ImportError(msg) from exc

    if not path:
        path = Path.cwd()

    # create the graph
    graph = graphviz.Digraph()
    for layer in self.walk(dags.WalkOrder("BREADTH")):
        for idx in range(len(layer.vars)):
            node_name = self.formatter.generate(layer.name, idx)
            graph.node(node_name.replace(":", "-"))
        for parent, edges in self._all_edges[layer.name].items():
            for parent_idx, child_idx in edges:
                parent_name = self.formatter.generate(parent, parent_idx)
                child_name = self.formatter.generate(layer.name, child_idx)
                graph.edge(
                    parent_name.replace(":", "-"),
                    child_name.replace(":", "-"),
                )

    # write to disk
    graph.render(
        filename=self.name,
        directory=str(path),
        format=image_format,
        cleanup=True,
    )

write

write(path=None, *, write_script=False)

Write out the given DAG to the given directory.

This includes the DAG description file itself, as well as any associated submit descriptions and log directories.

Also optionally writes out the list of commands for each node, which represents commands that would be run on the execute point, after taking into account file location changes where the job would be run if file transfer is enabled.

Parameters:

Name	Type	Description	Default
path	Path	The directory to write the DAG files to. Defaults to the current working directory.	None
write_script	bool	Also write out the list of commands for each node to disk. Defaults to false.	False

Source code in ezdag/dags.py

def write(self, path: Path | None = None, *, write_script: bool = False) -> None:
    """Write out the given DAG to the given directory.

    This includes the DAG description file itself, as well as any
    associated submit descriptions and log directories.

    Also optionally writes out the list of commands for each node, which
    represents commands that would be run on the execute point, after
    taking into account file location changes where the job would be run if
    file transfer is enabled.

    Parameters
    ----------
    path : Path
        The directory to write the DAG files to. Defaults to the current working
        directory.
    write_script : bool
        Also write out the list of commands for each node to disk. Defaults
        to false.

    """
    if not path:
        path = Path.cwd()

    dag_file = f"{self.name}.dag"
    self._write_dag(dag_file, path=path)
    self._dag_path = str(path / dag_file)
    if write_script:
        self._write_script(f"{self.name}.sh", path=path)

EdgeConnector

EdgeConnector(indices)

Bases: BaseEdge

This edge connects individual nodes in layers given an explicit mapping.

Source code in ezdag/dags.py

def __init__(self, indices) -> None:
    self.indices = indices