wfexs_backend.ro_crate

Module Contents

Classes

FormalParameter
PropertyValue
Intangible	Although an intangible is a more general concept than PropertyValue keep them isolated for now.
ContactPoint
Action
CreateAction
OrganizeAction
ControlAction
Collection
FixedMixin
FixedFile
ContainerImage
MaterializedContainerImage
WorkflowDiagram
SourceCodeFile
FixedDataset
FixedWorkflow
FixedROCrate	This subclass fixes the limitations from original ROCrate class
WorkflowRunROCrate	This class rules the generation of an RO-Crate

API

exception wfexs_backend.ro_crate.ROCrateGenerationException

Bases: wfexs_backend.common.AbstractWfExSException

class wfexs_backend.ro_crate.FormalParameter(crate: rocrate.rocrate.ROCrate, name: str, additional_type: str | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

class wfexs_backend.ro_crate.PropertyValue(crate: rocrate.rocrate.ROCrate, name: str, value: bool | str | int | float, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

class wfexs_backend.ro_crate.Intangible(crate: rocrate.rocrate.ROCrate, name: str, additionalType: str | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

Although an intangible is a more general concept than PropertyValue keep them isolated for now.

Initialization

class wfexs_backend.ro_crate.ContactPoint(crate: rocrate.rocrate.ROCrate, contactType: str, identifier: str, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

class wfexs_backend.ro_crate.Action(crate: rocrate.rocrate.ROCrate, name: str, startTime: datetime.datetime | None = None, endTime: datetime.datetime | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

class wfexs_backend.ro_crate.CreateAction(crate: rocrate.rocrate.ROCrate, name: str, startTime: datetime.datetime | None = None, endTime: datetime.datetime | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.Action

class wfexs_backend.ro_crate.OrganizeAction(crate: rocrate.rocrate.ROCrate, name: str, startTime: datetime.datetime | None = None, endTime: datetime.datetime | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.Action

class wfexs_backend.ro_crate.ControlAction(crate: rocrate.rocrate.ROCrate, name: str, startTime: datetime.datetime | None = None, endTime: datetime.datetime | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.Action

class wfexs_backend.ro_crate.Collection(crate: rocrate.rocrate.ROCrate, main_entity: FixedFile | FixedDataset | Collection | None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.creativework.CreativeWork

class wfexs_backend.ro_crate.FixedMixin(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.file_or_dir.FileOrDir

class wfexs_backend.ro_crate.FixedFile(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.FixedMixin, rocrate.model.file.File

write(base_path: wfexs_backend.common.PathLikePath) → None

class wfexs_backend.ro_crate.ContainerImage(crate: rocrate.rocrate.ROCrate, name: str, container_type: wfexs_backend.common.ContainerType, registry: str | None = None, tag: str | None = None, identifier: str | None = None, properties: Mapping[str, Any] | None = None)

Bases: rocrate.model.entity.Entity

TYPES = ['ContainerImage', 'SoftwareApplication']

_empty() → Mapping[str, Any]

static _prepare_properties(name: str, container_type: wfexs_backend.common.ContainerType, registry: str | None = None, tag: str | None = None, properties: Mapping[str, Any] | None = None) → Mapping[str, Any]

class wfexs_backend.ro_crate.MaterializedContainerImage(crate: rocrate.rocrate.ROCrate, container_type: wfexs_backend.common.ContainerType, registry: str | None = None, name: str = '', tag: str | None = None, identifier: str | None = None, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.ContainerImage, wfexs_backend.ro_crate.FixedFile

TYPES = ['File', 'ContainerImage', 'SoftwareApplication']

class wfexs_backend.ro_crate.WorkflowDiagram(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.FixedFile

TYPES = ['File', 'ImageObject']

_empty() → Mapping[str, Any]

class wfexs_backend.ro_crate.SourceCodeFile(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.FixedFile

TYPES = ['File', 'SoftwareSourceCode']

_empty() → Mapping[str, Any]

class wfexs_backend.ro_crate.FixedDataset(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.FixedMixin, rocrate.model.dataset.Dataset

write(base_path: str | os.PathLike[str]) → None

class wfexs_backend.ro_crate.FixedWorkflow(crate: rocrate.rocrate.ROCrate, source: str | pathlib.Path | None = None, dest_path: str | pathlib.Path | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None)

Bases: wfexs_backend.ro_crate.FixedMixin, rocrate.model.computationalworkflow.ComputationalWorkflow

TYPES = ['File', 'SoftwareSourceCode', 'ComputationalWorkflow', 'SoftwareApplication']

class wfexs_backend.ro_crate.FixedROCrate(source=None, gen_preview=False, init=False, exclude=None)

Bases: rocrate.rocrate.ROCrate

This subclass fixes the limitations from original ROCrate class

Initialization

add_file(source: str | pathlib.Path | None = None, dest_path: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None) → wfexs_backend.ro_crate.FixedFile

add_file_ext(source: str | pathlib.Path | None = None, dest_path: str | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None, clazz: Type[FixedFile] = FixedFile) → wfexs_backend.ro_crate.FixedFile

source: The absolute path to the local copy of the file, if exists. dest_path: The relative path inside the RO-Crate for the file copy. identifier: The forced value for the @id of the File declaration.

add_dataset(source: str | pathlib.Path | None = None, dest_path: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None) → wfexs_backend.ro_crate.FixedDataset

add_dataset_ext(source: str | pathlib.Path | None = None, dest_path: str | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None) → wfexs_backend.ro_crate.FixedDataset

source: The absolute path to the local copy of the file, if exists. dest_path: The relative path inside the RO-Crate for the file copy. identifier: The forced value for the @id of the File declaration.

add_directory = None

add_workflow(source: str | pathlib.Path | None = None, dest_path: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None, main: bool = False, lang: str | rocrate.model.computerlanguage.ComputerLanguage = 'cwl', lang_version: str | None = None, gen_cwl: bool = False, cls: Type[rocrate.model.computationalworkflow.ComputationalWorkflow] = FixedWorkflow) → wfexs_backend.ro_crate.FixedWorkflow

add_workflow_ext(source: str | pathlib.Path | None = None, dest_path: str | None = None, identifier: str | None = None, fetch_remote: bool = False, validate_url: bool = False, properties: Mapping[str, Any] | None = None, main: bool = False, lang: str | rocrate.model.computerlanguage.ComputerLanguage = 'cwl', lang_version: str | None = None, gen_cwl: bool = False, cls: Type[rocrate.model.computationalworkflow.ComputationalWorkflow] = FixedWorkflow) → wfexs_backend.ro_crate.FixedWorkflow

class wfexs_backend.ro_crate.WorkflowRunROCrate(remote_repo: wfexs_backend.fetchers.RemoteRepo, workflow_pid: str | None, localWorkflow: wfexs_backend.common.LocalWorkflow, materializedEngine: wfexs_backend.workflow_engines.MaterializedWorkflowEngine, workflowEngineVersion: WorkflowEngineVersionStr | None, containerEngineVersion: ContainerEngineVersionStr | None, containerEngineOs: ContainerOperatingSystem | None, arch: ProcessorArchitecture | None, staged_setup: wfexs_backend.common.StagedSetup, payloads: wfexs_backend.common.CratableItem = NoCratableItem, licences: Sequence[LicenceDescription] = [], orcids: Sequence[str | ResolvedORCID] = [], progs: wfexs_backend.common.ProgsMapping = {}, tempdir: pathlib.Path | None = None, scheme_desc: Sequence[Tuple[str, str, int]] = [], crate_pid: str | None = None, licence_matcher: LicenceMatcher | None = None)

This class rules the generation of an RO-Crate

Initialization

_init_empty_crate_and_ComputerLanguage(langVersion: EngineVersion | WFLangVersion | None, licences: Sequence[LicenceDescription], crate_pid: str | None) → None

_process_licences(licdescs: Sequence[LicenceDescription]) → Sequence[str | rocrate.model.creativework.CreativeWork]

_process_licence(licdesc: wfexs_backend.common.LicenceDescription) → str | rocrate.model.creativework.CreativeWork

_add_wfexs_to_crate(scheme_desc: Sequence[Tuple[str, str, int]]) → rocrate.model.softwareapplication.SoftwareApplication

_add_containers(containers: Sequence[Container], sa_crate: rocrate.model.computationalworkflow.ComputationalWorkflow | rocrate.model.softwareapplication.SoftwareApplication, the_workflow_crate: FixedWorkflow | None = None) → MutableSequence[ContainerImage]

addWorkflowInputs(inputs: Sequence[MaterializedInput], are_envvars: bool = False) → Sequence[rocrate.model.entity.Entity]

Add the input’s or environment variables provenance data to a Research Object.

Parameters:

inputs (Sequence[MaterializedInput]) – List of inputs to add

_add_file_to_crate(the_path: pathlib.Path, the_uri: URIType | None, the_id: str | None = None, the_name: RelPath | None = None, the_alternate_name: RelPath | None = None, the_size: int | None = None, the_signature: Fingerprint | None = None, the_licences: Sequence[LicenceDescription] | None = None, the_mime: str | None = None, is_soft_source: bool = False, do_attach: bool = True) → wfexs_backend.ro_crate.FixedFile

_add_collection_to_crate(main_entity: FixedFile | FixedDataset | Collection | None = None) → wfexs_backend.ro_crate.Collection

_add_directory_as_dataset(the_path: pathlib.Path, the_uri: wfexs_backend.common.URIType, the_id: str | None = None, the_name: RelPath | None = None, the_alternate_name: RelPath | None = None, do_attach: bool = True) → Tuple[FixedDataset, Sequence[FixedFile]] | Tuple[None, None]

_add_workflow_to_crate(the_workflow: wfexs_backend.common.LocalWorkflow, lang: rocrate.model.computerlanguage.ComputerLanguage, the_description: str | None, the_weng_crate: rocrate.model.softwareapplication.SoftwareApplication, materialized_engine: wfexs_backend.workflow_engines.MaterializedWorkflowEngine, the_uri: str | None = None, remote_repo: RemoteRepo | None = None, main: bool = False, gen_cwl: bool = False, do_attach: bool = True, was_workflow_run: bool = True) → wfexs_backend.ro_crate.FixedWorkflow

addWorkflowExpectedOutputs(outputs: Sequence[ExpectedOutput]) → None

writeWRROC(filename: pathlib.Path) → None

addStagedWorkflowDetails(inputs: Sequence[MaterializedInput], environment: Sequence[MaterializedInput], outputs: Sequence[ExpectedOutput] | None, profiles: Sequence[str] | None = None) → None

This method is used for WRROCs with only prospective provenance

addWorkflowExecution(stagedExec: wfexs_backend.workflow_engines.StagedExecution, expected_outputs: Sequence[ExpectedOutput] | None = None) → None

_add_workflow_execution_outputs(outputs: Sequence[MaterializedOutput], job_work_dir: pathlib.Path) → Sequence[rocrate.model.entity.Entity]

Add the output’s provenance data to a Research Object.

Parameters:

outputs (Sequence[MaterializedOutput]) – List of outputs to add

_add_GeneratedContent_to_crate(the_content: wfexs_backend.common.GeneratedContent, job_work_dir: pathlib.Path, do_attach: bool = True) → FixedFile | Collection

_add_GeneratedDirectoryContent_as_dataset(the_content: wfexs_backend.common.GeneratedDirectoryContent, job_work_dir: pathlib.Path, do_attach: bool = True) → Tuple[FixedDataset | Collection, Sequence[FixedFile | Collection]] | Tuple[None, None]