cwltool.cwlprov.provenance_profile

Module Contents

Classes

ProvenanceProfile

Provenance profile.

Functions

copy_job_order(job, job_order_object)

Create copy of job object for provenance.

cwltool.cwlprov.provenance_profile.copy_job_order(job, job_order_object)

Create copy of job object for provenance.

Parameters:

• job (Union[cwltool.process.Process, cwltool.utils.JobsType]) –

• job_order_object (cwltool.utils.CWLObjectType) –

Return type:

cwltool.utils.CWLObjectType

class cwltool.cwlprov.provenance_profile.ProvenanceProfile(research_object, full_name, host_provenance, user_provenance, orcid, fsaccess, run_uuid=None)

Provenance profile.

Populated as the workflow runs.

Parameters:

• research_object (cwltool.cwlprov.ro.ResearchObject) –

• full_name (str) –

• host_provenance (bool) –

• user_provenance (bool) –

• orcid (str) –

• fsaccess (cwltool.stdfsaccess.StdFsAccess) –

• run_uuid (Optional[uuid.UUID]) –

__str__()

Represent this Provenvance profile as a string.

Return type:

str

generate_prov_doc()

Add basic namespaces.

Return type:

Tuple[str, prov.model.ProvDocument]

evaluate(process, job, job_order_object, research_obj)

Evaluate the nature of job.

Parameters:

• process (cwltool.process.Process) –

• job (cwltool.utils.JobsType) –

• job_order_object (cwltool.utils.CWLObjectType) –

• research_obj (cwltool.cwlprov.ro.ResearchObject) –

Return type:

None

record_process_start(process, job, process_run_id=None)

Parameters:

• process (cwltool.process.Process) –

• job (cwltool.utils.JobsType) –

• process_run_id (Optional[str]) –

Return type:

Optional[str]

start_process(process_name, when, process_run_id=None)

Record the start of each Process.

Parameters:

• process_name (str) –

• when (datetime.datetime) –

• process_run_id (Optional[str]) –

Return type:

str

record_process_end(process_name, process_run_id, outputs, when)

Parameters:

• process_name (str) –

• process_run_id (str) –

• outputs (Union[cwltool.utils.CWLObjectType, MutableSequence[cwltool.utils.CWLObjectType], None]) –

• when (datetime.datetime) –

Return type:

None

declare_file(value)

Parameters:

value (cwltool.utils.CWLObjectType) –

Return type:

Tuple[prov.model.ProvEntity, prov.model.ProvEntity, str]

declare_directory(value)

Register any nested files/directories.

Parameters:

value (cwltool.utils.CWLObjectType) –

Return type:

prov.model.ProvEntity

declare_string(value)

Save as string in UTF-8.

Parameters:

value (str) –

Return type:

Tuple[prov.model.ProvEntity, str]

declare_artefact(value)

Create data artefact entities for all file objects.

Parameters:

value (Any) –

Return type:

prov.model.ProvEntity

used_artefacts(job_order, process_run_id, name=None)

Add used() for each data artefact.

Parameters:

• job_order (Union[cwltool.utils.CWLObjectType, List[cwltool.utils.CWLObjectType]]) –

• process_run_id (str) –

• name (Optional[str]) –

Return type:

None

generate_output_prov(final_output, process_run_id, name)

Call wasGeneratedBy() for each output,copy the files into the RO.

Parameters:

• final_output (Union[cwltool.utils.CWLObjectType, MutableSequence[cwltool.utils.CWLObjectType], None]) –

• process_run_id (Optional[str]) –

• name (Optional[str]) –

Return type:

None

prospective_prov(job)

Create prospective prov recording as wfdesc prov:Plan.

Parameters:

job (cwltool.utils.JobsType) –

Return type:

None

activity_has_provenance(activity, prov_ids)

Add http://www.w3.org/TR/prov-aq/ relations to nested PROV files.

Parameters:

• activity (str) –

• prov_ids (Sequence[prov.identifier.Identifier]) –

Return type:

None

finalize_prov_profile(name)

Transfer the provenance related files to the RO.

Parameters:

name (Optional[str]) –

Return type:

List[prov.identifier.QualifiedName]