wfcommons.wfinstances

wfcommons.wfinstances.instance

class wfcommons.wfinstances.instance.Instance(input_instance: str, schema_file: Optional[str] = None, logger: Optional[logging.Logger] = None)

Bases: object

Representation of one execution of one workflow on a set of machines

Instance(input_instance = 'instance.json')
Parameters

  • input_instance (str) – The JSON instance.

  • schema_file (str) – The path to the JSON schema that defines the instance. If no schema file is provided, it will look for a local copy of the WfCommons schema, and if not available it will fetch the latest schema from the WfCommons schema GitHub repository.

  • logger (Logger) – The logger where to log information/warning or errors.

draw(output: Optional[str] = None, extension: str = 'pdf') → None

Produce an image or a pdf file representing the instance.

Parameters

  • output (str) – Name of the output file.

  • extension – Type of the file extension (pdf, png, or svg).

leaves() → List[str]

Get the leaves of the workflow (i.e., the tasks without any successors).

Returns

List of leaves

Return type

List[str]

roots() → List[str]

Get the roots of the workflow (i.e., the tasks without any predecessors).

Returns

List of roots

Return type

List[str]

write_dot(output: Optional[str] = None) → None

Write a dot file of the instance.

Parameters

output (str) – The output dot file name (optional).

wfcommons.wfinstances.instance_analyzer

class wfcommons.wfinstances.instance_analyzer.InstanceAnalyzer(logger: Optional[logging.Logger] = None)

Bases: object

Set of tools for analyzing collections of instances.

Parameters

logger (Logger) – The logger where to log information/warning or errors (optional).

append_instance(instance: wfcommons.wfinstances.instance.Instance) → None

Append a workflow instance object to the instance analyzer.

instance = Instance(input_instance = 'instance.json', schema = 'schema.json')
instance_analyzer = InstanceAnalyzer()
instance_analyzer.append_instance(instance)
Parameters

instance (Instance) – A workflow instance object.

build_summary(tasks_list: List[str], include_raw_data: Optional[bool] = True) → Dict[str, Dict[str, Any]]

Analyzes appended instances and produce a summary of the analysis per task prefix.

workflow_tasks = ['sG1IterDecon', 'wrapper_siftSTFByMisfit']
instances_summary = instance_analyzer.build_summary(workflow_tasks, include_raw_data=False)
Parameters

  • tasks_list (List[str]) – List of workflow tasks prefix (e.g., mProject, sol2sanger, add_replace)

  • include_raw_data (bool) – Whether to include the raw data in the instance summary.

Returns

A summary of the analysis of instances in the form of a dictionary in which keys are task prefixes.

Return type

Dict[str, Dict[str, Any]]

generate_all_fit_plots(outfile_prefix: Optional[str] = None) → None

Produce fit plots as images for each entry of the summary analysis. For entries in which there are no distribution (i.e., constant value), no plot will be generated.

Parameters

outfile_prefix (str) – Prefix to be attached to each generated plot file name (optional).

generate_fit_plots(instance_element: wfcommons.wfinstances.instance_analyzer.InstanceElement, outfile_prefix: Optional[str] = None) → None

Produce fit plots as images for each entry of an instance element generated by the summary analysis. For entries in which there are no distribution (i.e., constant value), no plot will be generated.

Parameters

  • instance_element (InstanceElement) – Workflow element for which the fit plots will be generated.

  • outfile_prefix (str) – Prefix to be attached to each generated plot file name (optional).

class wfcommons.wfinstances.instance_analyzer.InstanceElement(value)

Bases: wfcommons.utils.NoValue

An enumeration.

INPUT = ('input', 'Input File Size (bytes)')
OUTPUT = ('output', 'Input File Size (bytes)')
RUNTIME = ('runtime', 'Runtime (s)')

wfcommons.wfinstances.logs.makeflow

class wfcommons.wfinstances.logs.makeflow.MakeflowLogsParser(execution_dir: str, resource_monitor_logs_dir: str, description: Optional[str] = None, logger: Optional[logging.Logger] = None)

Bases: wfcommons.wfinstances.logs.abstract_logs_parser.LogsParser

Parse Makeflow submit directory to generate workflow instance.

Parameters

  • execution_dir (str) – Makeflow workflow execution directory (contains .mf and .makeflowlog files).

  • resource_monitor_logs_dir (str) – Resource Monitor log files directory.

  • description (str) – Workflow instance description.

  • logger (Logger) – The logger where to log information/warning or errors (optional).

build_workflow(workflow_name: Optional[str] = None) → wfcommons.common.workflow.Workflow

Create workflow instance based on the workflow execution logs.

Parameters

workflow_name (str) – The workflow name.

Returns

A workflow instance object.

Return type

Workflow

wfcommons.wfinstances.logs.pegasus

class wfcommons.wfinstances.logs.pegasus.PegasusLogsParser(submit_dir: str, description: Optional[str] = None, ignore_auxiliary: Optional[bool] = True, legacy: Optional[bool] = False, logger: Optional[logging.Logger] = None)

Bases: wfcommons.wfinstances.logs.abstract_logs_parser.LogsParser

Parse Pegasus submit directory to generate workflow instance.

Parameters

  • submit_dir (str) – Pegasus submit directory.

  • legacy (bool) – Whether the submit directory is from a Pegasus 4.x version.

  • description (str) – Workflow instance description.

  • ignore_auxiliary (bool) – Ignore auxiliary jobs.

  • logger (Logger) – The logger where to log information/warning or errors (optional).

build_workflow(workflow_name: Optional[str] = None) → wfcommons.common.workflow.Workflow

Create workflow instance based on the workflow execution logs.

Parameters

workflow_name (str) – The workflow name.

Returns

A workflow instance object.

Return type

Workflow