wfcommons.common¶

wfcommons.common.file¶

class wfcommons.common.file.File(file_id: str, size: int, logger: Logger | None = None)[source]¶

Bases: object

Representation of a file.

Parameters:

file_id (str) – The id of the file.
size (int) – File size in bytes.
logger (Optional[Logger]) – The logger where to log information/warning or errors.

as_dict() → Dict[str, int | str][source]¶

A JSON representation of the file.

Returns:: A JSON object representation of the file.
Return type:: Dict[str, Union[str, int]]

wfcommons.common.task¶

class wfcommons.common.task.TaskType(value)[source]¶

Bases: NoValue

Task type.

COMPUTE = 'compute'¶

AUXILIARY = 'auxiliary'¶

TRANSFER = 'transfer'¶

SUBWORKFLOW = 'subworkflow'¶

Bases: object

Representation of a task.

Parameters:

name (str) – The name of the task.
task_id (str) – Task unique ID (e.g., ID0000001).
runtime (float) – Task runtime in seconds.
input_files (Optional[List[File]]) – List of input files used by the task.
output_files (Optional[List[File]]) – List of output files used by the task.
cores (float) – Number of cores required by the task.
category (Optional[str]) – Task category (can be used, for example, to define tasks that use the same program).
machines (Optional[List[Machine]]) – Machines on which is the task has been executed.
program (Optional[str]) – Program name.
args (Optional[List[str]]) – List of task arguments.
avg_cpu (Optional[float]) – Average CPU utilization in %.
bytes_read (Optional[int]) – Total bytes read in KB.
bytes_written (Optional[int]) – Total bytes written in KB.
memory (Optional[int]) – Memory (resident set) size of the process in bytes.
energy (Optional[int]) – Total energy consumption in kWh.
avg_power (Optional[float]) – Average power consumption in W.
priority (Optional[int]) – Task priority.
task_type (TaskType) – The type of the task.
logger (Optional[Logger]) – The logger where to log information/warning or errors.

logger: Logger¶

name: str¶

task_id: str¶

runtime: float¶

cores: float | None¶

category: str | None¶

program: str | None¶

args: List[str]¶

avg_cpu: float | None¶

bytes_read: int | None¶

bytes_written: int | None¶

memory: int | None¶

energy: int | None¶

avg_power: float | None¶

input_files: List[File]¶

output_files: List[File]¶

machines: List[Machine] | None¶

priority: int | None¶

type: TaskType | None¶

launch_dir: str | None¶

start_time: str | None¶

specification_as_dict() → Dict[source]¶

A JSON representation of the task specification.

Returns:: A JSON object representation of the task.
Return type:: Dict

execution_as_dict() → Dict[source]¶

A JSON representation of the task execution.

Returns:: A JSON object representation of the task.
Return type:: Dict

wfcommons.common.machine¶

class wfcommons.common.machine.MachineSystem(value)[source]¶

Bases: NoValue

Machine system type.

LINUX = 'linux'¶

MACOS = 'macos'¶

WINDOWS = 'windows'¶

Bases: object

Representation of one compute machine.

Parameters:

name (str) – Machine node name.
cpu (Dict[str, Union[int, str]]) –
A dictionary containing information about the CPU specification. Must at least contains two fields: count (number of CPU cores) and speed (CPU speed of each core in MHz).
```
cpu = {
    'coreCount': 48,
    'speedInMHz': 1200,
    'vendor': 'Vendor Name'
}
```
system (MachineSystem) – Machine system (linux, macos, windows).
architecture (str) – Machine architecture (e.g., x86_64, ppc).
memory (int) – Total machine’s RAM memory in bytes.
release (str) – Machine release.
hashcode (str) – MD5 Hashcode for the Machine.
logger (Logger) – The logger where to log information/warning or errors.

logger: Logger¶

name: str¶

system: MachineSystem¶

architecture: str¶

memory: int¶

release: str¶

cpu_cores: int¶

cpu_speed: int¶

cpu_flops: int¶

cpu_vendor: str¶

as_dict() → Dict[str, int | str][source]¶

A JSON representation of the machine.

Returns:: A JSON object representation of the machine.
Return type:: Dict[str, Union[int, str]]

wfcommons.common.workflow¶

class wfcommons.common.workflow.Workflow(*args, backend=None, **kwargs)[source]¶

Bases: DiGraph

Representation of a workflow. The workflow representation is an extension of the NetworkX DiGraph class.

Parameters:

name (str) – Workflow name.
description (Optional[str]) – Workflow instance description.
runtime_system_name (Optional[str]) – WMS name.
runtime_system_version (Optional[str]) – WMS version.
runtime_system_url (Optional[str]) – URL for the WMS website.
executed_at (Optional[str]) – Workflow start timestamp in the ISO 8601 format.
makespan (Optional[float]) – Workflow makespan in seconds.
author_name (Optional[str]) – Author name.
author_email (Optional[str]) – Author email.
author_institution (Optional[str]) – Author institution.
author_country (Optional[str]) – Author country (preferably country code, ISO ALPHA-2 code).

add_task(task: Task) → None[source]¶

Add a Task to the workflow.

Parameters:: task (Task) – A Task object.

add_dependency(parent: str, child: str) → None[source]¶

Add a dependency between tasks.

Parameters:

parent (str) – Parent task id.
child (str) – Child task id.

write_json(json_file_path: Path | None = None) → None[source]¶

Write a JSON file of the workflow instance.

Parameters:: json_file_path (Optional[pathlib.Path]) – JSON output file name.

generate_json() → None[source]¶

Generate a JSON representation of the workflow instance.

Returns:: A JSON representation of the workflow instance.
Return type:: dict

write_dot(dot_file_path: Path | None = None) → None[source]¶

Write a dot file of the workflow instance.

Parameters:: dot_file_path (Optional[pathlib.Path]) – DOT output file name.

read_dot(dot_file_path: Path | None = None) → None[source]¶

Read a dot file of the workflow instance.

Parameters:: dot_file_path (Optional[pathlib.Path]) – DOT input file name.

to_nx_digraph() → DiGraph[source]¶

roots() → List[str][source]¶

leaves() → List[str][source]¶