layers

HexFormatter

HexFormatter(separator=':', index_format='{:05X}', offset=0)

Bases: SimpleFormatter

A hex-based node formatter that produces names like LayerName:0000C.

Source code in ezdag/layers.py

def __init__(
    self, separator: str = ":", index_format: str = "{:05X}", offset: int = 0
) -> None:
    self.separator = separator
    self.index_format = index_format
    self.offset = offset

Layer dataclass

Defines a single layer (or set of related jobs) in an HTCondor DAG.

Stores submit configuration for a set of nodes as well as providing functionality to determine the parent-child relationships between nodes.

Parameters:

Name	Type	Description	Default
executable	str	The path of the executable to run.	required
name	str	The human-readable name of this node. Defaults to the basename of the executable if not given.	''
universe	str	The execution environment for a job. Defaults to 'vanilla'.	'vanilla'
log_dir	str	The directory in which logs will be written to. Defaults to ./logs.	'logs'
retries	int	The number of retries given for a job. Defaults to 3.	3
transfer_files	bool	Whether to leverage Condor file transfer for moving around files. On by default.	True
submit_description	Union[dict, Submit]	The submit descriptors representing this set of jobs.	field(default_factory=dict)
requirements	dict	The submit descriptors representing this set of jobs. Deprecated in favor for submit_description to avoid confusion, as 'requirements' refers to a specific submit descriptor. This option will be removed in a future release.	field(default_factory=dict)
nodes	list	The nodes representing the layer. Nodes can be passed upon instantiation or added to the layer after the fact via Layer.append(node), Layer.extend(nodes), or Layer += node.	field(default_factory=list)

has_dependencies property

has_dependencies

Check if any of the nodes in this layer have dependencies.

append

append(node)

Append a node to this layer.

Source code in ezdag/layers.py

def append(self, node: Node) -> None:
    """Append a node to this layer."""
    assert isinstance(node.inputs, list)
    assert isinstance(node.outputs, list)
    for input_ in node.inputs:
        self.inputs.setdefault(input_.name, []).append(input_.argument)
    for output in node.outputs:
        self.outputs.setdefault(output.name, []).append(output.argument)
    self.nodes.append(node)

command

command(node, readjust_paths=True)

Given a node, return the command that would be run.

Parameters:

Name	Type	Description	Default
node	Node	The node to return the command for.	required
readjust_paths	bool	Determines whether path locations are readjusted based on the command that would be run on the node's execute point. This only has an effect if using file transfer. Default is True.	True

Source code in ezdag/layers.py

def command(self, node, readjust_paths: bool = True):
    """Given a node, return the command that would be run.

    Parameters
    ----------
    node : Node
        The node to return the command for.
    readjust_paths : bool
        Determines whether path locations are readjusted based on
        the command that would be run on the node's execute point.
        This only has an effect if using file transfer. Default is True.

    """
    args = re.sub(r"\$\(((\w+?))\)", r"{\1}", self._arguments())
    # extract node variables
    node_vars = {arg.condor_name: arg.vars() for arg in node.arguments}
    for arg in node.inputs:
        if self.transfer_files and readjust_paths:
            node_vars[arg.condor_name] = arg.vars(basename=path.is_abs_or_url)
        else:
            node_vars[arg.condor_name] = arg.vars()
    for arg in node.outputs:
        basename = readjust_paths and self.transfer_files
        node_vars[arg.condor_name] = arg.vars(basename=basename)
    return self.executable + " " + args.format(**node_vars)

config

config(formatter=None)

Generates a layer configuration.

This configuration can be passed directly into an htcondor.dags.NodeLayer if desired.

Parameters:

Name	Type	Description	Default
formatter	NodeNameFormatter	Defines how the node names are defined and formatted. Defaults to a hex-based formatter with 5 digits.	None

Source code in ezdag/layers.py

def config(
    self,
    formatter: Optional[dags.NodeNameFormatter] = None,
) -> Dict[str, Any]:
    """Generates a layer configuration.

    This configuration can be passed directly into an
    htcondor.dags.NodeLayer if desired.

    Parameters
    ----------
    formatter : htcondor.dags.NodeNameFormatter
        Defines how the node names are defined and formatted. Defaults to a
        hex-based formatter with 5 digits.

    """
    # check that nodes are valid
    self.validate()

    # update submit description with defaults + other layer configuration
    submit_description = self._update_submit_defaults(self.submit_description)

    if not formatter:
        formatter = HexFormatter()
    return {
        "name": self.name,
        "submit_description": submit_description,
        "vars": self._vars(formatter),
        "retries": self.retries,
    }

extend

extend(nodes)

Append multiple nodes to this layer.

Source code in ezdag/layers.py

def extend(self, nodes: Iterable[Node]) -> None:
    """Append multiple nodes to this layer."""
    for node in nodes:
        self.append(node)

validate

validate()

Ensure all nodes in this layer are consistent with each other.

Source code in ezdag/layers.py

def validate(self) -> None:
    """Ensure all nodes in this layer are consistent with each other."""
    assert self.nodes, "at least one node must be connected to this layer"

    # check arg names across nodes are equal
    args = [arg.name for arg in self.nodes[0].arguments]
    for node in self.nodes[:-1]:
        assert args == [arg.name for arg in node.arguments]

    # check input/output names across nodes are equal
    inputs = [arg.name for arg in self.nodes[0].inputs]
    for node in self.nodes[:-1]:
        assert inputs == [arg.name for arg in node.inputs]
    outputs = [arg.name for arg in self.nodes[0].outputs]
    for node in self.nodes[:-1]:
        assert outputs == [arg.name for arg in node.outputs]

    # check meta-parameters (equality, name validity)
    variables = list(self.nodes[0].variables.keys())
    for node in self.nodes[:-1]:
        assert variables == list(node.variables.keys())
    for var in variables:
        if var in PROTECTED_CONDOR_VARS:
            raise ValueError(
                f"{var} is a protected condor name for node {self.name}"
            )

Node dataclass

Defines a single node (or job) in an HTCondor DAG.

Stores both the arguments used within a job as well as capturing any inputs and outputs the job uses/creates.

Parameters:

Name	Type	Description	Default
arguments	Union[Argument, Option, list]	The arguments the node uses which aren't I/O related.	field(default_factory=list)
inputs	Union[Argument, Option, list]	The arguments the node takes as inputs.	field(default_factory=list)
outputs	Union[Argument, Option, list]	The arguments the node takes as outputs.	field(default_factory=list)
variables	dict	Meta parameters that can be used within the submit description.	field(default_factory=dict)

provides property

provides

Returns:

Type	Description
list	The outputs this node provides when it completes.

requires property

requires

Returns:

Type	Description
list	The inputs this node explicitly depends on to run.