cwltool.cwlprov.provenance_profile

Classes

ProvenanceProfile

Provenance profile.

Functions

copy_job_order(job, job_order_object)

Create copy of job object for provenance.

Module Contents

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])

fsaccess

orcid

research_object

folder

document

host_provenance

user_provenance

engine_uuid

add_to_manifest

full_name

workflow_run_uuid

workflow_run_uri = 'urn: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, collections.abc.MutableSequence[cwltool.utils.CWLObjectType], None])
when (datetime.datetime)

Return type:

None

declare_file(value)

Construct a FileEntity for the given CWL File object.

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, collections.abc.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 (collections.abc.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]