wfexs_backend.utils.rocrate

`wfexs_backend.utils.rocrate`#

Module Contents#

Classes#

`ReproducibilityLevel`
`ContainerTypeMetadata`
`ROCratePayload`
`ContainerImageAdditionalType`
`ROCrateToolbox`

Functions#

ReadROCrateMetadata

Data#

`ContainerTypeMetadataDetails`
`ApplicationCategory2ContainerType`
`WORKFLOW_RUN_BASE`
`WORKFLOW_RUN_CONTEXT`
`WORKFLOW_RUN_NAMESPACE`
`WFEXS_TERMS_BASE`
`WFEXS_TERMS_CONTEXT`
`WFEXS_TERMS_NAMESPACE`
`CONTAINER_DOCKERIMAGE_SHORT`
`CONTAINER_SIFIMAGE_SHORT`
`ContentWithURIsMIMEs`
`RevContentWithURIsMIMEs`
`StrContainerAdditionalType2ContainerImageAdditionalType`
`ContainerType2AdditionalType`
`AdditionalType2ContainerType`
`ROCRATE_JSONLD_FILENAME`
`LEGACY_ROCRATE_JSONLD_FILENAME`

API#

class wfexs_backend.utils.rocrate.ReproducibilityLevel#

Bases: enum.IntEnum

Minimal = 'auto(...)'#

Metadata = 'auto(...)'#

Full = 'auto(...)'#

class wfexs_backend.utils.rocrate.ContainerTypeMetadata#

Bases: typing.NamedTuple

sa_id: str = None#

applicationCategory: str = None#

ct_applicationCategory: str = None#

class wfexs_backend.utils.rocrate.ROCratePayload#

Bases: typing.NamedTuple

rel_path: str = None#

path: pathlib.Path = None#

signature: Fingerprint | None = None#

wfexs_backend.utils.rocrate.ContainerTypeMetadataDetails: Final[Mapping[ContainerType, ContainerTypeMetadata]] = None#

wfexs_backend.utils.rocrate.ApplicationCategory2ContainerType: Final[Mapping[str, ContainerType]] = None#

wfexs_backend.utils.rocrate.WORKFLOW_RUN_BASE: Final[str] = 'https://w3id.org/ro/terms/workflow-run'#

wfexs_backend.utils.rocrate.WORKFLOW_RUN_CONTEXT: Final[str] = None#

wfexs_backend.utils.rocrate.WORKFLOW_RUN_NAMESPACE: Final[str] = None#

wfexs_backend.utils.rocrate.WFEXS_TERMS_BASE: Final[str] = 'https://w3id.org/ro/terms/wfexs'#

wfexs_backend.utils.rocrate.WFEXS_TERMS_CONTEXT: Final[str] = None#

wfexs_backend.utils.rocrate.WFEXS_TERMS_NAMESPACE: Final[str] = None#

wfexs_backend.utils.rocrate.CONTAINER_DOCKERIMAGE_SHORT: Final[str] = 'DockerImage'#

wfexs_backend.utils.rocrate.CONTAINER_SIFIMAGE_SHORT: Final[str] = 'SIFImage'#

wfexs_backend.utils.rocrate.ContentWithURIsMIMEs = None#

wfexs_backend.utils.rocrate.RevContentWithURIsMIMEs = 'dict(...)'#

class wfexs_backend.utils.rocrate.ContainerImageAdditionalType(*args, **kwds)#

Bases: enum.Enum

Docker = None#

Singularity = None#

wfexs_backend.utils.rocrate.StrContainerAdditionalType2ContainerImageAdditionalType: Final[Mapping[str, ContainerImageAdditionalType]] = None#

wfexs_backend.utils.rocrate.ContainerType2AdditionalType: Final[Mapping[ContainerType, ContainerImageAdditionalType]] = None#

wfexs_backend.utils.rocrate.AdditionalType2ContainerType: Final[Mapping[ContainerImageAdditionalType, ContainerType]] = None#

exception wfexs_backend.utils.rocrate.ROCrateToolboxException#: Bases: Exception

wfexs_backend.utils.rocrate.ROCRATE_JSONLD_FILENAME: Final[str] = 'ro-crate-metadata.json'#

wfexs_backend.utils.rocrate.LEGACY_ROCRATE_JSONLD_FILENAME: Final[str] = 'ro-crate-metadata.jsonld'#

wfexs_backend.utils.rocrate.ReadROCrateMetadata(workflowROCrateFilename: pathlib.Path, public_name: str) → Tuple[Any, pathlib.Path | None]#

class wfexs_backend.utils.rocrate.ROCrateToolbox(wfexs: wfexs_backend.wfexs_backend.WfExSBackend)#

Bases: abc.ABC

RELATIVE_ROCRATE_SCHEME: Final[str] = 'rel-crate'#

RELATIVE_ROCRATE_SPARQL_BASE: Final[str] = None#

RELATIVE_ROCRATE_NS: Final[str] = 'crate'#

SCHEMA_ORG_PREFIX: Final[str] = 'http://schema.org/'#

SPARQL_NS = None#

LEAF_TYPE_2_ADDITIONAL_TYPE: Final[Mapping[str, str]] = None#

LEAF_TYPE_2_OUTPUT_ADDITIONAL_TYPE: Final[Mapping[str, str]] = None#

IS_ROCRATE_SPARQL: Final[str] = <Multiline-String>#

GET_LICENCES_SPARQL: Final[str] = <Multiline-String>#

identifyROCrate(jsonld: Mapping[str, Any], public_name: str) → Tuple[rdflib.query.ResultRow | None, rdflib.graph.Graph]#

This method is used to identify where the input JSON is a JSON-LD related to RO-Crate.

The returned value is a tuple, where the first element is the result row giving the QName of the root dataset, and the different profiles being matched: RO-Crate, Workflow RO-Crate, WRROC process and WRROC workflow. The second element of the returned tuple is the rdflib RDF graph from the read JSON-LD, which should allow exploring it.

OBTAIN_WORKFLOW_PID_SPARQL: Final[str] = <Multiline-String>#

LIST_PARTS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_RUNS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_RUN_CONTAINERS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_WORKFLOW_INPUTS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_WORKFLOW_ENV_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_WORKFLOW_OUTPUTS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_EXECUTION_INPUTS_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_EXECUTION_ENV_SPARQL: Final[str] = <Multiline-String>#

OBTAIN_EXECUTION_OUTPUTS_SPARQL: Final[str] = <Multiline-String>#

_parseContainersFromWorkflow(g: rdflib.graph.Graph, main_entity: rdflib.term.Identifier, payload_dir: pathlib.Path | None = None) → Tuple[ContainerType, Sequence[Container]] | None#

_parseContainersFromExecution(g: rdflib.graph.Graph, execution: rdflib.term.Identifier, main_entity: rdflib.term.Identifier, payload_dir: pathlib.Path | None = None) → Tuple[ContainerType, Sequence[Container]] | None#

__parseContainersResults(qcontainersres: rdflib.query.Result, main_entity: rdflib.term.Identifier, payload_dir: pathlib.Path | None = None) → Tuple[ContainerType, Sequence[Container]] | None#

_parseOutputsFromExecution(g: rdflib.graph.Graph, execution: rdflib.term.Identifier, main_entity: rdflib.term.Identifier, public_name: str) → wfexs_backend.workflow.OutputsBlock#

_parseOutputsFromMainEntity(g: rdflib.graph.Graph, main_entity: rdflib.term.Identifier, public_name: str) → wfexs_backend.workflow.OutputsBlock#

__parseOutputsResults(qoutputsres: rdflib.query.Result, g: rdflib.graph.Graph, public_name: str) → wfexs_backend.workflow.OutputsBlock#

_parseInputsFromExecution(g: rdflib.graph.Graph, execution: rdflib.term.Identifier, main_entity: rdflib.term.Identifier, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[ParamsBlock, Sequence[MaterializedInput] | None]#

_parseInputsFromMainEntity(g: rdflib.graph.Graph, main_entity: rdflib.term.Identifier, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[ParamsBlock, Sequence[MaterializedInput] | None]#

__processPayloadInput(inputrow: rdflib.query.ResultRow, payload_dir: pathlib.Path, the_uri: str, licences: Sequence[str], input_type: str, kindobj: wfexs_backend.common.ContentKind, cached_inputs_hash: MutableMapping[str, MaterializedInput]) → MutableMapping[str, MaterializedInput]#

__parseInputsResults(qinputsres: rdflib.query.Result, g: rdflib.graph.Graph, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[ParamsBlock, Sequence[MaterializedInput] | None]#

_parseEnvFromExecution(g: rdflib.graph.Graph, execution: rdflib.term.Identifier, main_entity: rdflib.term.Identifier, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[EnvironmentBlock, Sequence[MaterializedInput] | None]#

_parseEnvFromMainEntity(g: rdflib.graph.Graph, main_entity: rdflib.term.Identifier, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[EnvironmentBlock, Sequence[MaterializedInput] | None]#

__parseEnvResults(qenvres: rdflib.query.Result, g: rdflib.graph.Graph, default_licences: Sequence[str], public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[EnvironmentBlock, Sequence[MaterializedInput] | None]#: This method is (almost) identical to __parseInputsResults

_getLicences(g: rdflib.graph.Graph, entity: rdflib.term.Identifier, public_name: str) → Sequence[str]#

__processPayloadEntity(the_entity: rdflib.term.Identifier, payload_dir: pathlib.Path, kindobj: wfexs_backend.common.ContentKind, entity_type: str, entity_name: str, the_file_size: rdflib.term.Node, the_file_sha256: rdflib.term.Node) → ROCratePayload | None#

__list_entity_parts(g: rdflib.graph.Graph, entity: rdflib.term.Identifier, public_name: str) → rdflib.query.Result#

__list_payload_entity_parts(g: rdflib.graph.Graph, entity: rdflib.term.Identifier, public_name: str, payload_dir: pathlib.Path) → Sequence[str | ROCratePayload]#

extractWorkflowMetadata(g: rdflib.graph.Graph, main_entity: rdflib.term.Identifier, default_repo: str | None, public_name: str, payload_dir: pathlib.Path | None = None) → Tuple[RemoteRepo, WorkflowType, LocalWorkflow | None]#

generateWorkflowMetaFromJSONLD(jsonld_obj: Mapping[str, Any], public_name: str, retrospective_first: bool = True, reproducibility_level: wfexs_backend.utils.rocrate.ReproducibilityLevel = ReproducibilityLevel.Metadata, strict_reproducibility_level: bool = False, payload_dir: pathlib.Path | None = None) → Tuple[RemoteRepo, WorkflowType, ContainerType, ParamsBlock, Sequence[str] | None, EnvironmentBlock, OutputsBlock, LocalWorkflow | None, Sequence[Container], Sequence[MaterializedInput] | None, Sequence[MaterializedInput] | None]#