API Reference¶

The public API surface is small; most users interact via the CLI. Importers/exporters live under wf2wf.importers and wf2wf.exporters.

Core Modules¶

.. py:module:: wf2wf.core

wf2wf.core – Intermediate Representation (IR) classes and helpers.

This module defines the canonical, engine-agnostic data structures that all importers must emit and all exporters must consume. Validation utilities and JSON/TOML (de)serialisers will be added in later iterations.

.. py:class:: BCOSpec(object_id: str | None = None, spec_version: str = ‘https://w3id.org/ieee/ieee-2791-schema/2791object.json’, etag: str | None = None, provenance_domain: ~typing.Dict[str, ~typing.Any] = , usability_domain: ~typing.List[str] = , extension_domain: ~typing.List[~typing.Dict[str, ~typing.Any]] = , description_domain: ~typing.Dict[str, ~typing.Any] = , execution_domain: ~typing.Dict[str, ~typing.Any] = , parametric_domain: ~typing.List[~typing.Dict[str, ~typing.Any]] = , io_domain: ~typing.Dict[str, ~typing.Any] = , error_domain: ~typing.Dict[str, ~typing.Any] = ) :module: wf2wf.core

BioCompute Object specification for regulatory compliance.

.. py:attribute:: BCOSpec.description_domain :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: BCOSpec.error_domain :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: BCOSpec.etag :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: BCOSpec.execution_domain :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: BCOSpec.extension_domain :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, ~typing.Any]]

.. py:attribute:: BCOSpec.io_domain :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: BCOSpec.object_id :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: BCOSpec.parametric_domain :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, ~typing.Any]]

.. py:attribute:: BCOSpec.provenance_domain :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: BCOSpec.spec_version :module: wf2wf.core :type: str :value: ‘https://w3id.org/ieee/ieee-2791-schema/2791object.json’

.. py:attribute:: BCOSpec.usability_domain :module: wf2wf.core :type: ~typing.List[str]

.. py:class:: CheckpointSpec(strategy: ‘Optional[str]’ = None, interval: ‘Optional[int]’ = None, storage_location: ‘Optional[str]’ = None, enabled: ‘Optional[bool]’ = None, notes: ‘Optional[str]’ = None) :module: wf2wf.core

.. py:attribute:: CheckpointSpec.enabled :module: wf2wf.core :type: bool | None :value: None

.. py:attribute:: CheckpointSpec.interval :module: wf2wf.core :type: int | None :value: None

.. py:attribute:: CheckpointSpec.notes :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: CheckpointSpec.storage_location :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: CheckpointSpec.strategy :module: wf2wf.core :type: str | None :value: None

.. py:class:: DocumentationSpec(description: str | None = None, label: str | None = None, doc: str | None = None, intent: ~typing.List[str] = , usage_notes: str | None = None, examples: ~typing.List[~typing.Dict[str, ~typing.Any]] = ) :module: wf2wf.core

Rich documentation for workflows and tasks.

.. py:attribute:: DocumentationSpec.description :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: DocumentationSpec.doc :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: DocumentationSpec.examples :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, ~typing.Any]]

.. py:attribute:: DocumentationSpec.intent :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: DocumentationSpec.label :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: DocumentationSpec.usage_notes :module: wf2wf.core :type: str | None :value: None

.. py:class:: Edge(parent: str, child: str) :module: wf2wf.core

Directed edge relating parent → child task.

.. py:attribute:: Edge.child :module: wf2wf.core :type: str

.. py:attribute:: Edge.parent :module: wf2wf.core :type: str

.. py:class:: EnvironmentAdapter(workflow: ~wf2wf.core.Workflow) :module: wf2wf.core

Adapt workflows for specific execution environments.

.. py:method:: EnvironmentAdapter.adapt_for_environment(target_env: str) -> ~wf2wf.core.Workflow :module: wf2wf.core

  Adapt workflow for specific execution environment.

.. py:class:: EnvironmentSpecificValue(value: ~typing.Any = None, environments: ~typing.List[str] | None = None) :module: wf2wf.core

A value that can have different values for different execution environments.

.. py:method:: EnvironmentSpecificValue.add_environment(environment: str) :module: wf2wf.core

  Add an environment to the most recent value's applicable environments.

.. py:method:: EnvironmentSpecificValue.all_environments() -> ~typing.Set[str] :module: wf2wf.core

  Get all environments that have values set.

.. py:method:: EnvironmentSpecificValue.copy() -> ~wf2wf.core.EnvironmentSpecificValue :module: wf2wf.core

  Create a deep copy of this EnvironmentSpecificValue.

.. py:attribute:: EnvironmentSpecificValue.default_value :module: wf2wf.core :type: ~typing.Any | None :value: None

.. py:method:: EnvironmentSpecificValue.get_default_value() -> ~typing.Any | None :module: wf2wf.core

  Get the default value.

.. py:method:: EnvironmentSpecificValue.get_value_for(environment: str) -> ~typing.Any | None :module: wf2wf.core

  Get the value for the given environment, or None if not set.

.. py:method:: EnvironmentSpecificValue.get_value_with_default(environment: str) -> ~typing.Any | None :module: wf2wf.core

  Get the value for the given environment, with fallback to default value.

.. py:method:: EnvironmentSpecificValue.has_default_value() -> bool :module: wf2wf.core

  Check if there's a default value set.

.. py:method:: EnvironmentSpecificValue.has_environment_specific_value(environment: str) -> bool :module: wf2wf.core

  Check if there's an environment-specific value for the given environment.

.. py:method:: EnvironmentSpecificValue.is_applicable_to(environment: str) -> bool :module: wf2wf.core

  Check if any value is applicable to the given environment.

.. py:method:: EnvironmentSpecificValue.remove_environment(environment: str) :module: wf2wf.core

  Remove an environment from all values.

.. py:method:: EnvironmentSpecificValue.set_default_value(value: ~typing.Any) :module: wf2wf.core

  Set the default value explicitly.

.. py:method:: EnvironmentSpecificValue.set_for_environment(value: ~typing.Any, environment: str | None) :module: wf2wf.core

  Set a value for a specific environment (replaces if already present).

.. py:attribute:: EnvironmentSpecificValue.values :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, ~typing.Any]]

.. py:class:: ExecutionEnvironment(name: str, display_name: str, description: str, filesystem_type: str, resource_management: str, environment_isolation: str, file_transfer_mode: str, default_file_transfer_mode: str = ‘auto’, default_resource_specification: bool = False, default_environment_isolation: bool = False, default_error_handling: bool = False, supports_gpu: bool = True, supports_checkpointing: bool = False, supports_partial_results: bool = False, supports_cloud_storage: bool = False, priority: int = 0, deprecated: bool = False, experimental: bool = False) :module: wf2wf.core

Definition of an execution environment with its characteristics.

.. py:attribute:: ExecutionEnvironment.default_environment_isolation :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.default_error_handling :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.default_file_transfer_mode :module: wf2wf.core :type: str :value: ‘auto’

.. py:attribute:: ExecutionEnvironment.default_resource_specification :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.deprecated :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.description :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.display_name :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.environment_isolation :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.experimental :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.file_transfer_mode :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.filesystem_type :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.name :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.priority :module: wf2wf.core :type: int :value: 0

.. py:attribute:: ExecutionEnvironment.resource_management :module: wf2wf.core :type: str

.. py:attribute:: ExecutionEnvironment.supports_checkpointing :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.supports_cloud_storage :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionEnvironment.supports_gpu :module: wf2wf.core :type: bool :value: True

.. py:attribute:: ExecutionEnvironment.supports_partial_results :module: wf2wf.core :type: bool :value: False

.. py:class:: ExecutionModelSpec(model: str, source_format: str, detection_method: str, filesystem_type: str = ‘unknown’, resource_management: str = ‘unknown’, environment_isolation: str = ‘unknown’, file_transfer_mode: str = ‘unknown’, requires_file_transfer: bool = False, requires_resource_specification: bool = False, requires_environment_isolation: bool = False, requires_error_handling: bool = False, detection_indicators: ~typing.List[str] = , transition_notes: ~typing.List[str] = , created_at: str | None = None, modified_at: str | None = None) :module: wf2wf.core

Detailed specification of an execution model with transition analysis capabilities.

.. py:attribute:: ExecutionModelSpec.created_at :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ExecutionModelSpec.detection_indicators :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: ExecutionModelSpec.detection_method :module: wf2wf.core :type: str

.. py:attribute:: ExecutionModelSpec.environment_isolation :module: wf2wf.core :type: str :value: ‘unknown’

.. py:attribute:: ExecutionModelSpec.file_transfer_mode :module: wf2wf.core :type: str :value: ‘unknown’

.. py:attribute:: ExecutionModelSpec.filesystem_type :module: wf2wf.core :type: str :value: ‘unknown’

.. py:method:: ExecutionModelSpec.get_environment_characteristics() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Get environment characteristics as a dictionary.

.. py:method:: ExecutionModelSpec.get_transition_requirements(target_model: str) -> ~typing.List[str] :module: wf2wf.core

  Get list of requirements for transitioning to target model.

.. py:method:: ExecutionModelSpec.is_compatible_with(target_model: str) -> bool :module: wf2wf.core

  Check if this model is compatible with a target model.

.. py:attribute:: ExecutionModelSpec.model :module: wf2wf.core :type: str

.. py:attribute:: ExecutionModelSpec.modified_at :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ExecutionModelSpec.requires_environment_isolation :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionModelSpec.requires_error_handling :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionModelSpec.requires_file_transfer :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionModelSpec.requires_resource_specification :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ExecutionModelSpec.resource_management :module: wf2wf.core :type: str :value: ‘unknown’

.. py:attribute:: ExecutionModelSpec.source_format :module: wf2wf.core :type: str

.. py:attribute:: ExecutionModelSpec.transition_notes :module: wf2wf.core :type: ~typing.List[str]

.. py:method:: ExecutionModelSpec.update_modified() :module: wf2wf.core

  Update the modified timestamp.

Enhanced file specification with CWL features.

.. py:attribute:: FileSpec.basename :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.checksum :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.class_type :module: wf2wf.core :type: str :value: ‘File’

.. py:method:: FileSpec.compute_stats(*, read_contents: bool = False) -> None :module: wf2wf.core

  Populate `checksum`, `size` and optionally `contents` if the path exists.

.. py:attribute:: FileSpec.contents :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.dirname :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.format :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.listing :module: wf2wf.core :type: ~typing.List[~wf2wf.core.FileSpec]

.. py:attribute:: FileSpec.nameext :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.nameroot :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: FileSpec.path :module: wf2wf.core :type: str

.. py:attribute:: FileSpec.secondary_files :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: FileSpec.size :module: wf2wf.core :type: int | None :value: None

.. py:method:: FileSpec.validate() -> None :module: wf2wf.core

.. py:class:: LoggingSpec(log_level: ‘Optional[str]’ = None, log_format: ‘Optional[str]’ = None, log_destination: ‘Optional[str]’ = None, aggregation: ‘Optional[str]’ = None, notes: ‘Optional[str]’ = None) :module: wf2wf.core

.. py:attribute:: LoggingSpec.aggregation :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: LoggingSpec.log_destination :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: LoggingSpec.log_format :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: LoggingSpec.log_level :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: LoggingSpec.notes :module: wf2wf.core :type: str | None :value: None

.. py:class:: MetadataSpec(source_format: str | None = None, source_file: str | None = None, source_version: str | None = None, original_execution_environment: str | None = None, original_source_format: str | None = None, parsing_notes: ~typing.List[str] = , conversion_warnings: ~typing.List[str] = , format_specific: ~typing.Dict[str, ~typing.Any] = , uninterpreted: ~typing.Dict[str, ~typing.Any] = , annotations: ~typing.Dict[str, ~typing.Any] = , environment_metadata: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = , validation_errors: ~typing.List[str] = , quality_metrics: ~typing.Dict[str, ~typing.Any] = ) :module: wf2wf.core

Comprehensive metadata storage for preserving uninterpreted data and format-specific information.

.. py:method:: MetadataSpec.add_environment_metadata(environment: str, key: str, value: ~typing.Any) :module: wf2wf.core

  Add environment-specific metadata.

.. py:method:: MetadataSpec.add_format_specific(key: str, value: ~typing.Any) :module: wf2wf.core

  Add format-specific data.

.. py:method:: MetadataSpec.add_parsing_note(note: str) :module: wf2wf.core

  Add a parsing note or warning.

.. py:method:: MetadataSpec.add_uninterpreted(key: str, value: ~typing.Any) :module: wf2wf.core

  Add uninterpreted data from source.

.. py:attribute:: MetadataSpec.annotations :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: MetadataSpec.conversion_warnings :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: MetadataSpec.environment_metadata :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]]

.. py:attribute:: MetadataSpec.format_specific :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:method:: MetadataSpec.get_environment_metadata(environment: str) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Get metadata for a specific environment.

.. py:attribute:: MetadataSpec.original_execution_environment :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: MetadataSpec.original_source_format :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: MetadataSpec.parsing_notes :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: MetadataSpec.quality_metrics :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: MetadataSpec.source_file :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: MetadataSpec.source_format :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: MetadataSpec.source_version :module: wf2wf.core :type: str | None :value: None

.. py:method:: MetadataSpec.to_dict() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Convert metadata to dictionary for JSON serialization.

.. py:attribute:: MetadataSpec.uninterpreted :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: MetadataSpec.validation_errors :module: wf2wf.core :type: ~typing.List[str]

.. py:class:: NetworkingSpec(network_mode: ‘Optional[str]’ = None, allowed_ports: ‘Optional[List[int]]’ = , egress_rules: ‘Optional[List[str]]’ = , ingress_rules: ‘Optional[List[str]]’ = , notes: ‘Optional[str]’ = None) :module: wf2wf.core

.. py:attribute:: NetworkingSpec.allowed_ports :module: wf2wf.core :type: ~typing.List[int] | None

.. py:attribute:: NetworkingSpec.egress_rules :module: wf2wf.core :type: ~typing.List[str] | None

.. py:attribute:: NetworkingSpec.ingress_rules :module: wf2wf.core :type: ~typing.List[str] | None

.. py:attribute:: NetworkingSpec.network_mode :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: NetworkingSpec.notes :module: wf2wf.core :type: str | None :value: None

.. py:class:: ParameterSpec(id: str, type: Union[str, TypeSpec], label: Optional[str] = None, doc: Optional[str] = None, default: Any = None, format: Optional[str] = None, secondary_files: List[str] = , streamable: bool = False, load_contents: bool = False, load_listing: Optional[str] = None, input_binding: Optional[Dict[str, Any]] = None, output_binding: Optional[Dict[str, Any]] = None, value_from: Optional[str] = None, wildcard_pattern: Optional[str] = None, transfer_mode: EnvironmentSpecificValue[str] = , staging_required: EnvironmentSpecificValue[bool] = , cleanup_after: EnvironmentSpecificValue[bool] = ) :module: wf2wf.core

CWL v1.2.1 parameter specification for inputs and outputs with environment awareness.

.. py:attribute:: ParameterSpec.cleanup_after :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: ParameterSpec.default :module: wf2wf.core :type: Any :value: None

.. py:attribute:: ParameterSpec.doc :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: ParameterSpec.format :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: ParameterSpec.id :module: wf2wf.core :type: str

.. py:attribute:: ParameterSpec.input_binding :module: wf2wf.core :type: Optional[Dict[str, Any]] :value: None

.. py:attribute:: ParameterSpec.label :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: ParameterSpec.load_contents :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ParameterSpec.load_listing :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: ParameterSpec.output_binding :module: wf2wf.core :type: Optional[Dict[str, Any]] :value: None

.. py:attribute:: ParameterSpec.secondary_files :module: wf2wf.core :type: List[str]

.. py:attribute:: ParameterSpec.staging_required :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: ParameterSpec.streamable :module: wf2wf.core :type: bool :value: False

.. py:attribute:: ParameterSpec.transfer_mode :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: ParameterSpec.type :module: wf2wf.core :type: Union[str, TypeSpec]

.. py:method:: ParameterSpec.validate() -> None :module: wf2wf.core

.. py:attribute:: ParameterSpec.value_from :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: ParameterSpec.wildcard_pattern :module: wf2wf.core :type: Optional[str] :value: None

.. py:class:: ProvenanceSpec(authors: ~typing.List[~typing.Dict[str, str]] = , contributors: ~typing.List[~typing.Dict[str, str]] = , created: str | None = None, modified: str | None = None, version: str | None = None, license: str | None = None, doi: str | None = None, citations: ~typing.List[str] = , keywords: ~typing.List[str] = , derived_from: str | None = None, extras: ~typing.Dict[str, ~typing.Any] = ) :module: wf2wf.core

Provenance and authorship information for workflows and tasks.

.. py:attribute:: ProvenanceSpec.authors :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, str]]

.. py:attribute:: ProvenanceSpec.citations :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: ProvenanceSpec.contributors :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, str]]

.. py:attribute:: ProvenanceSpec.created :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ProvenanceSpec.derived_from :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ProvenanceSpec.doi :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ProvenanceSpec.extras :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:attribute:: ProvenanceSpec.keywords :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: ProvenanceSpec.license :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ProvenanceSpec.modified :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: ProvenanceSpec.version :module: wf2wf.core :type: str | None :value: None

.. py:class:: RequirementSpec(class_name: str, data: ~typing.Dict[str, ~typing.Any] = ) :module: wf2wf.core

CWL requirement or hint specification.

.. py:attribute:: RequirementSpec.class_name :module: wf2wf.core :type: str

.. py:attribute:: RequirementSpec.data :module: wf2wf.core :type: ~typing.Dict[str, ~typing.Any]

.. py:method:: RequirementSpec.validate() -> None :module: wf2wf.core

.. py:class:: ScatterSpec(scatter: ~typing.List[str], scatter_method: str = ‘dotproduct’, wildcard_instances: ~typing.List[~typing.Dict[str, str]] = ) :module: wf2wf.core

Scatter operation specification for parallel execution.

.. py:attribute:: ScatterSpec.scatter :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: ScatterSpec.scatter_method :module: wf2wf.core :type: str :value: ‘dotproduct’

.. py:attribute:: ScatterSpec.wildcard_instances :module: wf2wf.core :type: ~typing.List[~typing.Dict[str, str]]

.. py:class:: SecuritySpec(encryption: ‘Optional[str]’ = None, access_policies: ‘Optional[str]’ = None, secrets: ‘Optional[Dict[str, str]]’ = , authentication: ‘Optional[str]’ = None, notes: ‘Optional[str]’ = None) :module: wf2wf.core

.. py:attribute:: SecuritySpec.access_policies :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: SecuritySpec.authentication :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: SecuritySpec.encryption :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: SecuritySpec.notes :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: SecuritySpec.secrets :module: wf2wf.core :type: ~typing.Dict[str, str] | None

.. py:class:: Task(id: str, label: Optional[str] = None, doc: Optional[str] = None, command: EnvironmentSpecificValue[str] = , script: EnvironmentSpecificValue[str] = , inputs: List[ParameterSpec] = , outputs: List[ParameterSpec] = , when: EnvironmentSpecificValue[str] = , scatter: EnvironmentSpecificValue[ScatterSpec] = , cpu: EnvironmentSpecificValue[int] = , mem_mb: EnvironmentSpecificValue[int] = , disk_mb: EnvironmentSpecificValue[int] = , gpu: EnvironmentSpecificValue[int] = , gpu_mem_mb: EnvironmentSpecificValue[int] = , time_s: EnvironmentSpecificValue[int] = , threads: EnvironmentSpecificValue[int] = , conda: EnvironmentSpecificValue[str] = , container: EnvironmentSpecificValue[str] = , workdir: EnvironmentSpecificValue[str] = , env_vars: EnvironmentSpecificValue[Dict[str, str]] = , modules: EnvironmentSpecificValue[List[str]] = , retry_count: EnvironmentSpecificValue[int] = , retry_delay: EnvironmentSpecificValue[int] = , retry_backoff: EnvironmentSpecificValue[str] = , max_runtime: EnvironmentSpecificValue[int] = , checkpoint_interval: EnvironmentSpecificValue[int] = , on_failure: EnvironmentSpecificValue[str] = , failure_notification: EnvironmentSpecificValue[str] = , cleanup_on_failure: EnvironmentSpecificValue[bool] = , restart_from_checkpoint: EnvironmentSpecificValue[bool] = , partial_results: EnvironmentSpecificValue[bool] = , priority: EnvironmentSpecificValue[int] = , file_transfer_mode: EnvironmentSpecificValue[str] = , staging_required: EnvironmentSpecificValue[bool] = , cleanup_after: EnvironmentSpecificValue[bool] = , cloud_provider: EnvironmentSpecificValue[str] = , cloud_storage_class: EnvironmentSpecificValue[str] = , cloud_encryption: EnvironmentSpecificValue[bool] = , parallel_transfers: EnvironmentSpecificValue[int] = , bandwidth_limit: EnvironmentSpecificValue[str] = , requirements: EnvironmentSpecificValue[List[RequirementSpec]] = , hints: EnvironmentSpecificValue[List[RequirementSpec]] = , checkpointing: EnvironmentSpecificValue[CheckpointSpec] = , logging: EnvironmentSpecificValue[LoggingSpec] = , security: EnvironmentSpecificValue[SecuritySpec] = , networking: EnvironmentSpecificValue[NetworkingSpec] = , provenance: Optional[ProvenanceSpec] = None, documentation: Optional[DocumentationSpec] = None, intent: List[str] = , metadata: Optional[MetadataSpec] = None, extra: Dict[str, EnvironmentSpecificValue] = ) :module: wf2wf.core

A single executable node in the workflow DAG with universal environment awareness.

.. py:method:: Task.add_environment_to_field(field_name: str, environment: str) :module: wf2wf.core

  Add an environment to an existing field's applicable environments.

.. py:attribute:: Task.bandwidth_limit :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.checkpoint_interval :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.checkpointing :module: wf2wf.core :type: EnvironmentSpecificValue[CheckpointSpec]

.. py:attribute:: Task.cleanup_after :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.cleanup_on_failure :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.cloud_encryption :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.cloud_provider :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.cloud_storage_class :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.command :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.conda :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.container :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:method:: Task.copy() -> ~wf2wf.core.Task :module: wf2wf.core

  Create a deep copy of this task.

.. py:attribute:: Task.cpu :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.disk_mb :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.doc :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: Task.documentation :module: wf2wf.core :type: Optional[DocumentationSpec] :value: None

.. py:attribute:: Task.env_vars :module: wf2wf.core :type: EnvironmentSpecificValue[Dict[str, str]]

.. py:attribute:: Task.extra :module: wf2wf.core :type: Dict[str, EnvironmentSpecificValue]

.. py:attribute:: Task.failure_notification :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.file_transfer_mode :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:method:: Task.get_for_environment(environment: str) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Get all values applicable to the given environment.

.. py:attribute:: Task.gpu :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.gpu_mem_mb :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:method:: Task.has_explicit_retry_for_environment(environment: str) -> bool :module: wf2wf.core

  Check if retry settings were explicitly set for the given environment.

.. py:attribute:: Task.hints :module: wf2wf.core :type: EnvironmentSpecificValue[List[RequirementSpec]]

.. py:attribute:: Task.id :module: wf2wf.core :type: str

.. py:attribute:: Task.inputs :module: wf2wf.core :type: List[ParameterSpec]

.. py:attribute:: Task.intent :module: wf2wf.core :type: List[str]

.. py:method:: Task.is_active(context: ~typing.Dict[str, ~typing.Any] | None = None, environment: str = ‘shared_filesystem’) -> bool :module: wf2wf.core

  Evaluate the *when* expression (if any) against *context* variables for the given environment.

.. py:attribute:: Task.label :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: Task.logging :module: wf2wf.core :type: EnvironmentSpecificValue[LoggingSpec]

.. py:attribute:: Task.max_runtime :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.mem_mb :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.metadata :module: wf2wf.core :type: Optional[MetadataSpec] :value: None

.. py:attribute:: Task.modules :module: wf2wf.core :type: EnvironmentSpecificValue[List[str]]

.. py:attribute:: Task.networking :module: wf2wf.core :type: EnvironmentSpecificValue[NetworkingSpec]

.. py:attribute:: Task.on_failure :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.outputs :module: wf2wf.core :type: List[ParameterSpec]

.. py:attribute:: Task.parallel_transfers :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.partial_results :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.priority :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.provenance :module: wf2wf.core :type: Optional[ProvenanceSpec] :value: None

.. py:attribute:: Task.requirements :module: wf2wf.core :type: EnvironmentSpecificValue[List[RequirementSpec]]

.. py:attribute:: Task.restart_from_checkpoint :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.retry_backoff :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.retry_count :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.retry_delay :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.scatter :module: wf2wf.core :type: EnvironmentSpecificValue[ScatterSpec]

.. py:method:: Task.scatter_bindings(runtime_inputs: ~typing.Dict[str, ~typing.Any], environment: str = ‘shared_filesystem’) -> ~typing.List[~typing.Dict[str, ~typing.Any]] :module: wf2wf.core

  Return a list of variable bindings for each scatter shard for the given environment.

.. py:attribute:: Task.script :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.security :module: wf2wf.core :type: EnvironmentSpecificValue[SecuritySpec]

.. py:method:: Task.set_for_environment(field_name: str, value: ~typing.Any, environment: str) :module: wf2wf.core

  Set a value for a specific environment.

.. py:method:: Task.set_retry_explicitly(retry_count: int, environment: str) :module: wf2wf.core

  Set retry count explicitly (user-specified, not inferred).

.. py:method:: Task.set_retry_inferred(retry_count: int, environment: str) :module: wf2wf.core

  Set retry count as inferred (system-specified, not user-specified).

.. py:attribute:: Task.staging_required :module: wf2wf.core :type: EnvironmentSpecificValue[bool]

.. py:attribute:: Task.threads :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:attribute:: Task.time_s :module: wf2wf.core :type: EnvironmentSpecificValue[int]

.. py:method:: Task.to_dict() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Convert task to dictionary for JSON serialization, excluding internal fields.

.. py:attribute:: Task.when :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:attribute:: Task.workdir :module: wf2wf.core :type: EnvironmentSpecificValue[str]

.. py:class:: TypeSpec(type: str, items: str | ~wf2wf.core.TypeSpec | None = None, fields: ~typing.Dict[str, ~wf2wf.core.TypeSpec] = , symbols: ~typing.List[str] = , members: ~typing.List[~wf2wf.core.TypeSpec] = , name: str | None = None, nullable: bool = False, default: ~typing.Any = None) :module: wf2wf.core

CWL v1.2.1 type specification with advanced features.

.. py:attribute:: TypeSpec.default :module: wf2wf.core :type: ~typing.Any :value: None

.. py:attribute:: TypeSpec.fields :module: wf2wf.core :type: ~typing.Dict[str, ~wf2wf.core.TypeSpec]

.. py:attribute:: TypeSpec.items :module: wf2wf.core :type: str | ~wf2wf.core.TypeSpec | None :value: None

.. py:attribute:: TypeSpec.members :module: wf2wf.core :type: ~typing.List[~wf2wf.core.TypeSpec]

.. py:attribute:: TypeSpec.name :module: wf2wf.core :type: str | None :value: None

.. py:attribute:: TypeSpec.nullable :module: wf2wf.core :type: bool :value: False

.. py:method:: TypeSpec.parse(obj: str | ~wf2wf.core.TypeSpec | ~typing.Dict[str, ~typing.Any]) -> ~wf2wf.core.TypeSpec :module: wf2wf.core :classmethod:

  Return a :class:`TypeSpec` instance from *obj*.

  Accepts CWL‐style shorthand strings such as ``File``, ``string?`` (nullable),
  ``File[]`` (array of File), or fully fledged mapping objects produced by
  ``cwltool --print-pre``.  If *obj* is already a :class:`TypeSpec*`` it is
  returned unchanged.

.. py:attribute:: TypeSpec.symbols :module: wf2wf.core :type: ~typing.List[str]

.. py:attribute:: TypeSpec.type :module: wf2wf.core :type: str

.. py:method:: TypeSpec.validate() -> None :module: wf2wf.core

  Semantic validation for the CWL type system.

  :raises ValueError: If the type definition is semantically invalid.

.. py:class:: WF2WFJSONDecoder() :module: wf2wf.core

Custom JSON decoder for wf2wf dataclasses and objects. Only supports the new EnvironmentSpecificValue format.

.. py:method:: WF2WFJSONDecoder.decode_environment_specific_value(data: ~typing.Dict[str, ~typing.Any]) -> ~wf2wf.core.EnvironmentSpecificValue :module: wf2wf.core :classmethod:

  Decode EnvironmentSpecificValue from JSON data (new format only).

.. py:method:: WF2WFJSONDecoder.decode_spec(data: ~typing.Dict[str, ~typing.Any], spec_class) -> ~typing.Any :module: wf2wf.core :classmethod:

  Decode spec objects from JSON data.

.. py:class:: WF2WFJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None) :module: wf2wf.core

Custom JSON encoder for wf2wf dataclasses and objects.

.. py:method:: WF2WFJSONEncoder.default(obj: ~typing.Any) -> ~typing.Any :module: wf2wf.core

  Convert wf2wf objects to JSON-serializable format.

.. py:class:: Workflow(name: str, version: str = ‘1.0’, label: Optional[str] = None, doc: Optional[str] = None, tasks: Dict[str, Task] = , edges: List[Edge] = , inputs: List[ParameterSpec] = , outputs: List[ParameterSpec] = , requirements: EnvironmentSpecificValue[List[RequirementSpec]] = , hints: EnvironmentSpecificValue[List[RequirementSpec]] = , provenance: Optional[ProvenanceSpec] = None, documentation: Optional[DocumentationSpec] = None, intent: List[str] = , cwl_version: Optional[str] = None, bco_spec: Optional[BCOSpec] = None, loss_map: List[Dict[str, Any]] = , metadata: Optional[MetadataSpec] = None, extra: Dict[str, EnvironmentSpecificValue] = ) :module: wf2wf.core

A collection of Tasks plus dependency edges and optional metadata with universal environment awareness.

.. py:method:: Workflow.add_edge(parent: str, child: str) :module: wf2wf.core

.. py:method:: Workflow.add_task(task: ~wf2wf.core.Task) :module: wf2wf.core

.. py:attribute:: Workflow.bco_spec :module: wf2wf.core :type: Optional[BCOSpec] :value: None

.. py:method:: Workflow.copy() -> ~wf2wf.core.Workflow :module: wf2wf.core

  Create a deep copy of this workflow.

.. py:attribute:: Workflow.cwl_version :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: Workflow.doc :module: wf2wf.core :type: Optional[str] :value: None

.. py:attribute:: Workflow.documentation :module: wf2wf.core :type: Optional[DocumentationSpec] :value: None

.. py:attribute:: Workflow.edges :module: wf2wf.core :type: List[Edge]

.. py:attribute:: Workflow.extra :module: wf2wf.core :type: Dict[str, EnvironmentSpecificValue]

.. py:method:: Workflow.from_dict(data: ~typing.Dict[str, ~typing.Any]) -> ~wf2wf.core.Workflow :module: wf2wf.core :classmethod:

  Re-hydrate from `json.load(...)` result (best-effort).

.. py:method:: Workflow.from_json(json_str: str) -> ~wf2wf.core.Workflow :module: wf2wf.core :classmethod:

  Re-hydrate from JSON string produced by :py:meth:`to_json`.

.. py:method:: Workflow.get_for_environment(environment: str) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Get workflow configuration for the given environment.

.. py:attribute:: Workflow.hints :module: wf2wf.core :type: EnvironmentSpecificValue[List[RequirementSpec]]

.. py:attribute:: Workflow.inputs :module: wf2wf.core :type: List[ParameterSpec]

.. py:attribute:: Workflow.intent :module: wf2wf.core :type: List[str]

.. py:attribute:: Workflow.label :module: wf2wf.core :type: Optional[str] :value: None

.. py:method:: Workflow.load_json(path: str | ~pathlib.Path) :module: wf2wf.core :classmethod:

  Load Workflow from a JSON file produced by :py:meth:`save_json`.

.. py:attribute:: Workflow.loss_map :module: wf2wf.core :type: List[Dict[str, Any]]

.. py:attribute:: Workflow.metadata :module: wf2wf.core :type: Optional[MetadataSpec] :value: None

.. py:attribute:: Workflow.name :module: wf2wf.core :type: str

.. py:attribute:: Workflow.outputs :module: wf2wf.core :type: List[ParameterSpec]

.. py:attribute:: Workflow.provenance :module: wf2wf.core :type: Optional[ProvenanceSpec] :value: None

.. py:attribute:: Workflow.requirements :module: wf2wf.core :type: EnvironmentSpecificValue[List[RequirementSpec]]

.. py:method:: Workflow.save_json(path: str | ~pathlib.Path, *, indent: int = 2) :module: wf2wf.core

  Write JSON representation to path using custom encoder.

.. py:attribute:: Workflow.tasks :module: wf2wf.core :type: Dict[str, Task]

.. py:method:: Workflow.to_dict() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.core

  Return a plain-Python representation ready for JSON/TOML dump.

.. py:method:: Workflow.to_json(*, indent: int = 2) -> str :module: wf2wf.core

  Return JSON representation using custom encoder.

.. py:method:: Workflow.validate() -> None :module: wf2wf.core

  Run JSON-Schema plus semantic validation checks.

  :raises ValueError or jsonschema.ValidationError if the workflow is invalid.:

.. py:attribute:: Workflow.version :module: wf2wf.core :type: str :value: ‘1.0’

.. py:module:: wf2wf.environ

wf2wf.environ – environment-build helpers (Phase 2)

This initial slice implements §9.2.1-9.2.2 of the design draft: • Generate a deterministic lock hash from a Conda YAML file. • Create a relocatable tarball (stand-in for conda-pack) so downstream exporters can reference a stable artefact even where Conda tooling is unavailable in the test environment.

Real micromamba/conda-pack execution will be wired in later; for now we simulate the build while preserving the critical interface and metadata.

.. py:class:: BuildahBuilder(*, tool: str | None = None, dry_run: bool = True, interactive: bool = False) :module: wf2wf.environ

Wrapper around buildah / podman build for sites that prefer it.

.. py:method:: BuildahBuilder.build(tarball: ~pathlib.Path, tag: str, labels: ~typing.Dict[str, str] | None = None, *, push: bool = False, build_cache: str | None = None, platform: str = ‘linux/amd64’) -> str :module: wf2wf.environ

  Build image, optionally push, and return digest (sha256:...).

.. py:class:: DockerBuildxBuilder(*, dry_run: bool = True, interactive: bool = False) :module: wf2wf.environ

Tiny wrapper around docker buildx build (dry-run by default).

.. py:method:: DockerBuildxBuilder.build(tarball: ~pathlib.Path, tag: str, labels: ~typing.Dict[str, str] | None = None, *, push: bool = False, build_cache: str | None = None, platform: str = ‘linux/amd64’) -> str :module: wf2wf.environ

  Build image, optionally push, and return digest (sha256:...).

.. py:class:: EnvironmentManager(interactive: bool = False, verbose: bool = False) :module: wf2wf.environ

Comprehensive environment and container management for workflow importers.

This class provides unified environment handling across all importers, including detection, parsing, validation, and interactive prompting.

.. py:method:: EnvironmentManager.adapt_environments_for_target(workflow: ~wf2wf.core.Workflow, target_format: str) -> None :module: wf2wf.environ

  Adapt environment specifications for target format.

  :param workflow: Workflow to adapt
  :param target_format: Target format name

.. py:method:: EnvironmentManager.build_environment_images(workflow: ~wf2wf.core.Workflow, registry: str | None = None, push: bool = False, dry_run: bool = True) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.environ

  Build environment images for workflow tasks.

  :param workflow: Workflow to process
  :param registry: Docker registry for pushing images
  :param push: Whether to push images to registry
  :param dry_run: Whether to perform actual builds

  :returns: Dictionary containing build results

.. py:method:: EnvironmentManager.detect_and_parse_environments(workflow: ~wf2wf.core.Workflow, source_format: str, source_path: ~pathlib.Path | None = None) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.environ

  Detect and parse environment specifications from workflow data.

  :param workflow: Workflow object to analyze
  :param source_format: Source format name
  :param source_path: Path to source file (for relative path resolution)

  :returns: Dictionary containing detected environment information

.. py:method:: EnvironmentManager.infer_missing_environments(workflow: ~wf2wf.core.Workflow, source_format: str) -> None :module: wf2wf.environ

  Infer missing environment specifications based on workflow content.

  :param workflow: Workflow to analyze
  :param source_format: Source format name

.. py:method:: EnvironmentManager.prompt_for_missing_environments(workflow: ~wf2wf.core.Workflow, source_format: str) -> None :module: wf2wf.environ

  Prompt user for missing environment specifications.

  :param workflow: Workflow to process
  :param source_format: Source format name

.. py:class:: OCIBuilder() :module: wf2wf.environ

Protocol-like base class for OCI builders.

.. py:method:: OCIBuilder.build(tarball: ~pathlib.Path, tag: str, labels: ~typing.Dict[str, str] | None = None, *, push: bool = False, build_cache: str | None = None) -> str :module: wf2wf.environ

  Build image, optionally push, and return digest (sha256:...).

.. py:function:: adapt_environments_for_target(workflow: ~wf2wf.core.Workflow, target_format: str, **kwargs) -> None :module: wf2wf.environ

Convenience function for adapting environments for target format.

.. py:function:: build_and_push_conda_env_images(conda_envs: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]], docker_registry: str, workflow_name: str = ‘workflow’, *, verbose: bool = False, debug: bool = False, dry_run: bool = False) -> bool :module: wf2wf.environ

Build and push Docker images for multiple conda environments.

:param conda_envs: Dictionary mapping env paths to env info (with ‘hash’ key) :param docker_registry: Docker registry URL :param workflow_name: Name to use for the repository :param verbose: Enable verbose output :param debug: Enable debug output :param dry_run: If True, only show what would be done without actually building/pushing

:returns: True if all operations were successful, False otherwise

.. py:function:: build_docker_image_from_conda_env(env_yaml_path: ~pathlib.Path, image_url: str, *, verbose: bool = False, debug: bool = False, dry_run: bool = False) -> bool :module: wf2wf.environ

Build a Docker image from a conda environment YAML file.

:param env_yaml_path: Path to the conda environment YAML file :param image_url: Full Docker image URL (registry/repo:tag) :param verbose: Enable verbose output :param debug: Enable debug output :param dry_run: If True, only show what would be done without actually building

:returns: True if build was successful, False otherwise

.. py:function:: build_environment_images(workflow: ~wf2wf.core.Workflow, **kwargs) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.environ

Convenience function for building environment images.

.. py:function:: build_oci_image(tarball: ~pathlib.Path, *, tag_prefix: str = ‘wf2wf/env’, backend: str = ‘buildx’, push: bool = False, platform: str = ‘linux/amd64’, build_cache: str | None = None, dry_run: bool = True, interactive: bool = False) -> tuple[str, str] :module: wf2wf.environ

High-level helper that picks a builder backend and returns (tag, digest).

.. py:function:: build_or_reuse_env_image(env_yaml: str | ~pathlib.Path, *, registry: str | None = None, push: bool = False, backend: str = ‘buildx’, dry_run: bool = True, build_cache: str | None = None, cache_dir: ~pathlib.Path | None = None, interactive: bool = False) -> ~typing.Dict[str, str] :module: wf2wf.environ

High-level helper: build image for env_yaml unless identical hash already indexed.

Returns dict with keys tag and digest.

.. py:function:: check_docker_image_exists(image_url: str, *, verbose: bool = False) -> bool :module: wf2wf.environ

Check if a Docker image exists in a remote registry using docker manifest inspect.

.. py:function:: convert_docker_images_to_apptainer(conda_envs: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]], sif_dir: str | ~pathlib.Path, *, verbose: bool = False, debug: bool = False, dry_run: bool = False) -> bool :module: wf2wf.environ

Convert Docker images to Apptainer SIF files for multiple conda environments.

:param conda_envs: Dictionary mapping env paths to env info (with ‘hash’ and ‘docker_image_url’ keys) :param sif_dir: Directory to store SIF files :param verbose: Enable verbose output :param debug: Enable debug output :param dry_run: If True, only show what would be done without actually converting

:returns: True if all conversions were successful, False otherwise

.. py:function:: convert_to_sif(image_ref: str, *, sif_dir: ~pathlib.Path | None = None, dry_run: bool = True) -> ~pathlib.Path :module: wf2wf.environ

Convert OCI image_ref to Apptainer SIF file.

Uses spython if available; otherwise simulates by touching a file.

.. py:function:: detect_and_parse_environments(workflow: ~wf2wf.core.Workflow, source_format: str, **kwargs) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.environ

Convenience function for detecting and parsing environments.

.. py:function:: extract_sbom_digest(env_vars: ~typing.Dict[str, str] | None) -> str | None :module: wf2wf.environ

Extract SBOM digest from environment variables.

:param env_vars: Environment variables dictionary

:returns: SBOM digest if present, None otherwise

.. py:function:: extract_sbom_path(env_vars: ~typing.Dict[str, str] | None) -> str | None :module: wf2wf.environ

Extract SBOM path from environment variables.

:param env_vars: Environment variables dictionary

:returns: SBOM path if present, None otherwise

.. py:function:: extract_sif_path(env_vars: ~typing.Dict[str, str] | None) -> str | None :module: wf2wf.environ

Extract SIF path from environment variables.

:param env_vars: Environment variables dictionary

:returns: SIF path if present, None otherwise

.. py:function:: format_container_for_target_format(container_spec: str, target_format: str) -> str :module: wf2wf.environ

Format container specification for a specific target format.

:param container_spec: Container specification :param target_format: Target format (e.g., “docker”, “singularity”, “cwl”, “wdl”, etc.)

:returns: Formatted container specification for the target format

.. py:function:: generate_conda_activation_script(env_name: str, command: str, *, script_path: str | ~pathlib.Path | None = None) -> ~pathlib.Path :module: wf2wf.environ

Generate a bash script that unpacks and activates a conda environment for HTCondor jobs.

This function creates a script that follows the CHTC best practices for conda environment activation in HTCondor jobs.

:param env_name: Name of the conda environment (without .tar.gz extension) :param command: The command to run after activating the environment :param script_path: Path to save the script (defaults to {env_name}_script.sh)

:returns: Path to the generated script

.. py:function:: generate_lock_hash(env_yaml: ~pathlib.Path) -> str :module: wf2wf.environ

Return sha256 digest hex string of the Conda YAML env_yaml.

The digest is calculated over the normalised file contents (strip CRLF, remove comment lines), ensuring platform-independent hashes.

.. py:function:: generate_sbom(image_ref: str, out_dir: ~pathlib.Path | None = None, *, dry_run: bool = True) -> ~wf2wf.environ.SBOMInfo :module: wf2wf.environ

Generate an SPDX SBOM for image_ref and return :class:SBOMInfo.

In dry-run mode (the default during unit tests) this creates a minimal JSON file containing the image reference and a fake package list.

.. py:function:: get_environment_metadata(env_vars: ~typing.Dict[str, str] | None) -> ~typing.Dict[str, str | None] :module: wf2wf.environ

Extract all environment-related metadata from environment variables.

:param env_vars: Environment variables dictionary

:returns: Dictionary containing SBOM path, SIF path, and SBOM digest

.. py:function:: infer_missing_environments(workflow: ~wf2wf.core.Workflow, source_format: str, **kwargs) -> None :module: wf2wf.environ

Convenience function for inferring missing environments.

.. py:function:: is_docker_available() -> bool :module: wf2wf.environ

Check if Docker is both installed and the daemon is running.

.. py:function:: normalize_container_spec(container_spec: str) -> str :module: wf2wf.environ

Normalize container specification by removing docker:// prefix.

:param container_spec: Container specification (e.g., “docker://image:tag” or “image:tag”)

:returns: // prefix :rtype: Normalized container specification without docker

.. py:function:: pack_conda_environment(env_yaml_path: str | ~pathlib.Path, output_dir: str | ~pathlib.Path | None = None, *, verbose: bool = False, dry_run: bool = False) -> ~pathlib.Path | None :module: wf2wf.environ

Package a conda environment using conda-pack for portable deployment.

This function creates a portable tarball of a conda environment that can be transferred to HTCondor execution nodes and unpacked/activated during job execution.

:param env_yaml_path: Path to the conda environment YAML file :param output_dir: Directory to save the packaged environment (defaults to same dir as YAML) :param verbose: Enable verbose output :param dry_run: If True, only show what would be done without actually packaging

:returns: Path to the created .tar.gz file, or None if packaging failed

:raises RuntimeError: If conda-pack is not available or packaging fails

.. py:function:: prepare_env(env_yaml: str | ~pathlib.Path, *, cache_dir: ~pathlib.Path | None = None, verbose: bool = False, dry_run: bool | None = None) -> ~wf2wf.environ.EnvBuildResult :module: wf2wf.environ

Simulate environment build pipeline and return artefact locations.

Compute lock hash from env_yaml.
Copy YAML to a content-addressed location <hash>.yaml inside cache_dir.
Create a tar.gz containing the YAML as a placeholder for a conda-pack archive and place it next to the lock file.

The function is idempotent: repeated calls with the same YAML content return the same paths without rebuilding.

.. py:function:: prompt_for_missing_environments(workflow: ~wf2wf.core.Workflow, source_format: str, **kwargs) -> None :module: wf2wf.environ

Convenience function for prompting for missing environments.

.. py:function:: prune_cache(*, days: int = 60, min_free_gb: int = 5, verbose: bool = False) :module: wf2wf.environ

Remove cache entries older than days if disk free space below threshold.

Very lightweight implementation; only checks tarballs & SIF files.

.. py:function:: push_docker_image(image_url: str, *, verbose: bool = False, debug: bool = False, dry_run: bool = False) -> bool :module: wf2wf.environ

Push a Docker image to a registry.

:param image_url: Full Docker image URL (registry/repo:tag) :param verbose: Enable verbose output :param debug: Enable debug output :param dry_run: If True, only show what would be done without actually pushing

:returns: True if push was successful, False otherwise

.. py:module:: wf2wf.validate

wf2wf.validate – JSON-Schema validation helper for Workflow IR.

.. py:function:: get_validation_summary(obj: ~typing.Any) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.validate

Get a comprehensive validation summary for a workflow.

:param obj: Workflow object or dictionary to validate

:returns:

         - valid: Boolean indicating if validation passed
         - issues: List of validation issues
         - warnings: List of validation warnings
         - stats: Dictionary with workflow statistics

:rtype: Dictionary with validation summary including

.. py:function:: validate_bco(bco_doc: ~typing.Dict[str, ~typing.Any]) -> None :module: wf2wf.validate

Validate bco_doc against the official IEEE 2791 JSON-Schema.

Downloads the schema (cached per session) and raises :class:jsonschema.ValidationError on failure.

.. py:function:: validate_environment_name(environment: str) -> bool :module: wf2wf.validate

Validate that an environment name is from the predefined list.

:param environment: Environment name to validate

:returns: True if valid, False otherwise

.. py:function:: validate_environment_specific_value(env_value: ~typing.Dict[str, ~typing.Any]) -> ~typing.List[str] :module: wf2wf.validate

Validate an EnvironmentSpecificValue object and return any issues.

:param env_value: EnvironmentSpecificValue dictionary

:returns: List of validation issues (empty if valid)

.. py:function:: validate_file_path(path: str, path_type: str = ‘unix_path’) -> bool :module: wf2wf.validate

Validate a file path against defined patterns.

:param path: Path to validate :param path_type: Type of path (‘unix_path’, ‘windows_path’, ‘url’, ‘docker_image’, ‘conda_env’)

:returns: True if valid, False otherwise

.. py:function:: validate_loss(loss_doc: ~typing.Dict[str, ~typing.Any]) -> None :module: wf2wf.validate

Validate loss_doc against the loss.json schema.

.. py:function:: validate_resource_value(resource_name: str, value: ~typing.Any) -> bool :module: wf2wf.validate

Validate a resource value against defined rules.

:param resource_name: Name of the resource (e.g., ‘cpu’, ‘mem_mb’) :param value: Value to validate

:returns: True if valid, False otherwise

.. py:function:: validate_workflow(obj: ~typing.Any) -> None :module: wf2wf.validate

Validate obj (Workflow or raw dict) against the v0.1 JSON schema.

:raises jsonschema.ValidationError: If the object does not conform to the schema.

.. py:function:: validate_workflow_enhanced(obj: ~typing.Any) -> ~typing.List[str] :module: wf2wf.validate

Enhanced workflow validation with additional checks.

:param obj: Workflow object or dictionary to validate

:returns: List of validation issues (empty if valid)

.. py:function:: validate_workflow_with_enhanced_checks(obj: ~typing.Any) -> None :module: wf2wf.validate

Validate workflow with both JSON Schema and enhanced checks.

This function provides more comprehensive validation than the basic validate_workflow() function.

:param obj: Workflow object or dictionary to validate

:raises jsonschema.ValidationError: If the object does not conform to the JSON schema. :raises ValueError: If enhanced validation checks fail.

Loss Tracking System¶

The loss tracking system provides comprehensive support for tracking, recording, validating, and reinjecting information loss during workflow format conversions.

.. py:module:: wf2wf.loss

Loss tracking and management system for wf2wf.

This package provides comprehensive loss tracking during workflow format conversions, including detection, recording, validation, and restoration of lost information.

.. py:class:: EnvironmentLossRecorder(source_format: str, target_format: str, target_environment: str) :module: wf2wf.loss :canonical: wf2wf.loss.context_detection.EnvironmentLossRecorder

Records environment-specific losses with detailed context.

.. py:method:: EnvironmentLossRecorder.record_environment_specific_value_loss(json_pointer: str, field: str, env_value: EnvironmentSpecificValue, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

  Record loss of environment-specific value with detailed context.

.. py:attribute:: EnvironmentLossRecorder.source_format :module: wf2wf.loss :type: str

.. py:attribute:: EnvironmentLossRecorder.target_environment :module: wf2wf.loss :type: str

.. py:attribute:: EnvironmentLossRecorder.target_format :module: wf2wf.loss :type: str

.. py:class:: FormatLossDetector(source_format: str, target_format: str) :module: wf2wf.loss :canonical: wf2wf.loss.context_detection.FormatLossDetector

Detects format-specific losses during import/export cycles.

.. py:method:: FormatLossDetector.detect_environment_specific_losses(workflow: Workflow) -> List[Dict[str, Any]] :module: wf2wf.loss

  Detect environment-specific value losses between formats.

.. py:attribute:: FormatLossDetector.source_format :module: wf2wf.loss :type: str

.. py:attribute:: FormatLossDetector.target_format :module: wf2wf.loss :type: str

.. py:class:: LossEntry :module: wf2wf.loss :canonical: wf2wf.loss.core.LossEntry

Typed dict wrapper for a loss mapping entry with comprehensive IR support.

.. py:function:: apply(workflow: Workflow, entries: List[LossEntry]) -> int :module: wf2wf.loss

Apply loss entries back to a workflow (reinjection).

This function attempts to reinject lost information back into the workflow IR, marking entries as ‘reapplied’ if successful.

:returns: Number of successfully applied entries

.. py:function:: as_list() -> ~typing.List[~wf2wf.loss.core.LossEntry] :module: wf2wf.loss

Return the current loss entries as a list.

.. py:function:: compute_checksum(workflow: Workflow) -> str :module: wf2wf.loss

Compute SHA-256 checksum of workflow IR for loss tracking.

.. py:function:: create_loss_document(target_engine: str, source_checksum: str, environment_adaptation: ~typing.Dict[str, ~typing.Any] | None = None, **kwargs) -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.loss

Create a comprehensive loss document with summary statistics.

.. py:function:: create_loss_sidecar_summary(workflow: Workflow, source_path: Path) -> Dict[str, Any] :module: wf2wf.loss

Create a summary of loss side-car information for a workflow.

:param workflow: Workflow object :param source_path: Path to the source workflow file

:returns: Dictionary containing loss side-car summary information

.. py:function:: detect_and_apply_loss_sidecar(workflow: Workflow, source_path: Path, verbose: bool = False) -> bool :module: wf2wf.loss

Detect and apply loss side-car during import.

This function looks for a loss side-car file next to the source file and applies any loss information to the workflow.

:param workflow: Workflow object to apply loss information to :param source_path: Path to the source workflow file :param verbose: Enable verbose logging

:returns: True if a loss side-car was found and applied, False otherwise

.. py:function:: detect_and_record_export_losses(workflow: ~wf2wf.core.Workflow, target_format: str, target_environment: str = ‘shared_filesystem’, verbose: bool = False) -> None :module: wf2wf.loss

Detect and record losses when converting to target format for specific environment.

.. py:function:: detect_and_record_import_losses(workflow: Workflow, source_format: str, target_environment: str = ‘shared_filesystem’, verbose: bool = False) -> None :module: wf2wf.loss

Detect and record losses when importing from source format.

.. py:function:: detect_format_specific_losses(workflow: Workflow, source_format: str, target_format: str) -> List[Dict[str, Any]] :module: wf2wf.loss

Detect format-specific losses in a workflow.

.. py:function:: generate_summary() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.loss

Generate summary statistics for the current loss entries.

.. py:function:: prepare(prev_entries: ~typing.List[~wf2wf.loss.core.LossEntry]) -> None :module: wf2wf.loss

Prepare for a new export cycle by remembering previously reapplied entries.

.. py:function:: record(json_pointer: str, field: str, lost_value: ~typing.Any, reason: str, origin: str = ‘user’, *, severity: str = ‘warn’, category: str = ‘advanced_features’, environment_context: ~typing.Dict[str, ~typing.Any] | None = None, adaptation_details: ~typing.Dict[str, ~typing.Any] | None = None, recovery_suggestions: ~typing.List[str] | None = None) -> None :module: wf2wf.loss

Append a comprehensive loss entry describing that field at json_pointer was lost.

:param json_pointer: JSON pointer to the field in the IR :type json_pointer: str :param field: Name of the field that was lost :type field: str :param lost_value: The value that could not be represented in the target format :type lost_value: Any :param reason: Human-readable reason for the loss :type reason: str :param origin: Whether the loss originated from user data or wf2wf processing :type origin: str :param severity: Severity level: info, warn, error :type severity: str :param category: Category of the lost information :type category: str :param environment_context: Environment-specific context for the loss :type environment_context: Optional[Dict[str, Any]] :param adaptation_details: Details about how the value was adapted :type adaptation_details: Optional[Dict[str, Any]] :param recovery_suggestions: Suggestions for recovering or working around the loss :type recovery_suggestions: Optional[List[str]]

.. py:function:: record_cwl_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to CWL format.

.. py:function:: record_dagman_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to DAGMan format.

.. py:function:: record_environment_adaptation(source_env: str, target_env: str, adaptation_type: str, details: ~typing.Dict[str, ~typing.Any], *, severity: str = ‘info’) -> None :module: wf2wf.loss

Record environment adaptation information for loss tracking.

:param source_env: The original execution environment :type source_env: str :param target_env: The target execution environment :type target_env: str :param adaptation_type: Type of adaptation: ‘filesystem_to_distributed’, ‘distributed_to_filesystem’, ‘cloud_migration’, ‘hybrid_conversion’, ‘edge_adaptation’ :type adaptation_type: str :param details: Detailed information about what changed during the adaptation :type details: Dict[str, Any] :param severity: Severity level: info, warn, error :type severity: str

.. py:function:: record_environment_specific_loss(json_pointer: str, field: str, env_value: EnvironmentSpecificValue, target_environment: str, reason: str, *, severity: str = ‘warn’, category: str = ‘environment_specific’) -> None :module: wf2wf.loss

Record loss of environment-specific values.

:param json_pointer: JSON pointer to the field in the IR :type json_pointer: str :param field: Name of the field that was lost :type field: str :param env_value: The environment-specific value that was lost :type env_value: EnvironmentSpecificValue :param target_environment: The target environment where the loss occurred :type target_environment: str :param reason: Reason for the loss :type reason: str :param severity: Severity level: info, warn, error :type severity: str :param category: Category of the lost information :type category: str

.. py:function:: record_environment_specific_value_loss(json_pointer: str, field: str, env_value: EnvironmentSpecificValue, source_format: str, target_format: str, target_environment: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

Record loss of environment-specific value with format context.

.. py:function:: record_error_handling_loss(task_id: str, error_field: str, original_value: ~typing.Any, target_environment: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

Record loss of error handling specifications.

:param task_id: ID of the task :type task_id: str :param error_field: Name of the error handling field (retry_count, retry_delay, etc.) :type error_field: str :param original_value: Original error handling value :type original_value: Any :param target_environment: Target environment where the loss occurred :type target_environment: str :param reason: Reason for the loss :type reason: str :param severity: Severity level: info, warn, error :type severity: str

.. py:function:: record_file_transfer_loss(task_id: str, transfer_field: str, original_value: ~typing.Any, target_environment: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

Record loss of file transfer specifications.

:param task_id: ID of the task :type task_id: str :param transfer_field: Name of the transfer field (file_transfer_mode, staging_required, etc.) :type transfer_field: str :param original_value: Original transfer value :type original_value: Any :param target_environment: Target environment where the loss occurred :type target_environment: str :param reason: Reason for the loss :type reason: str :param severity: Severity level: info, warn, error :type severity: str

.. py:function:: record_galaxy_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to Galaxy format.

.. py:function:: record_nextflow_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to Nextflow format.

.. py:function:: record_resource_specification_loss(task_id: str, resource_field: str, original_value: ~typing.Any, target_environment: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

Record loss of resource specifications.

:param task_id: ID of the task :type task_id: str :param resource_field: Name of the resource field (cpu, mem_mb, disk_mb, gpu, etc.) :type resource_field: str :param original_value: Original resource value :type original_value: Any :param target_environment: Target environment where the loss occurred :type target_environment: str :param reason: Reason for the loss :type reason: str :param severity: Severity level: info, warn, error :type severity: str

.. py:function:: record_snakemake_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to Snakemake format.

.. py:function:: record_spec_class_loss(json_pointer: str, field: str, spec_object: ~typing.Any, spec_type: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss

Record loss of specification class objects.

:param json_pointer: JSON pointer to the field in the IR :type json_pointer: str :param field: Name of the field that was lost :type field: str :param spec_object: The specification object that was lost :type spec_object: Any :param spec_type: Type of specification (LoggingSpec, SecuritySpec, etc.) :type spec_type: str :param reason: Reason for the loss :type reason: str :param severity: Severity level: info, warn, error :type severity: str

.. py:function:: record_wdl_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss

Record losses when converting to WDL format.

.. py:function:: reset() -> None :module: wf2wf.loss

Clear the in-memory loss buffer.

.. py:function:: restore_environment_specific_value(lost_value: Any, field_name: str, expected_type: type, target_environment: str) -> EnvironmentSpecificValue :module: wf2wf.loss

Restore an EnvironmentSpecificValue from lost data.

.. py:function:: validate_environment_specific_value(value: ~typing.Any, field_name: str, expected_type: type) -> bool :module: wf2wf.loss

Validate that a value is a proper EnvironmentSpecificValue.

.. py:function:: validate_loss_entry(entry: ~typing.Dict[str, ~typing.Any], verbose: bool = False) -> bool :module: wf2wf.loss

Validate a single loss entry.

:param entry: Dictionary containing loss entry data :param verbose: Enable verbose logging

:returns: True if the loss entry is valid, False otherwise

.. py:function:: validate_loss_sidecar(loss_data: ~typing.Dict[str, ~typing.Any], workflow, verbose: bool = False) -> bool :module: wf2wf.loss

Validate loss side-car data.

:param loss_data: Dictionary containing loss data :param workflow: Workflow IR object (for checksum validation) :param verbose: Enable verbose logging

:returns: True if the loss side-car is valid, False otherwise

.. py:function:: write(doc: ~typing.Dict[str, ~typing.Any], path: str | ~pathlib.Path, **kwargs) -> None :module: wf2wf.loss

Write loss document to file.

.. py:function:: write_loss_document(path: str | ~pathlib.Path, target_engine: str, source_checksum: str, environment_adaptation: ~typing.Dict[str, ~typing.Any] | None = None, **kwargs) -> None :module: wf2wf.loss

Write a loss document to file.

Loss System Submodules¶

.. py:module:: wf2wf.loss.core

Core loss tracking and management functionality.

.. py:class:: LossEntry :module: wf2wf.loss.core

Typed dict wrapper for a loss mapping entry with comprehensive IR support.

.. py:function:: apply(workflow: Workflow, entries: List[LossEntry]) -> int :module: wf2wf.loss.core

Apply loss entries back to a workflow (reinjection).

This function attempts to reinject lost information back into the workflow IR, marking entries as ‘reapplied’ if successful.

:returns: Number of successfully applied entries

.. py:function:: as_list() -> ~typing.List[~wf2wf.loss.core.LossEntry] :module: wf2wf.loss.core

Return the current loss entries as a list.

.. py:function:: compute_checksum(workflow: Workflow) -> str :module: wf2wf.loss.core

Compute SHA-256 checksum of workflow IR for loss tracking.

Create a comprehensive loss document with summary statistics.

.. py:function:: create_loss_sidecar_summary(workflow: Workflow, source_path: Path) -> Dict[str, Any] :module: wf2wf.loss.core

Create a summary of loss side-car information for a workflow.

:param workflow: Workflow object :param source_path: Path to the source workflow file

:returns: Dictionary containing loss side-car summary information

.. py:function:: detect_and_apply_loss_sidecar(workflow: Workflow, source_path: Path, verbose: bool = False) -> bool :module: wf2wf.loss.core

Detect and apply loss side-car during import.

This function looks for a loss side-car file next to the source file and applies any loss information to the workflow.

:param workflow: Workflow object to apply loss information to :param source_path: Path to the source workflow file :param verbose: Enable verbose logging

:returns: True if a loss side-car was found and applied, False otherwise

.. py:function:: generate_summary() -> ~typing.Dict[str, ~typing.Any] :module: wf2wf.loss.core

Generate summary statistics for the current loss entries.

.. py:function:: prepare(prev_entries: ~typing.List[~wf2wf.loss.core.LossEntry]) -> None :module: wf2wf.loss.core

Prepare for a new export cycle by remembering previously reapplied entries.

Append a comprehensive loss entry describing that field at json_pointer was lost.

Record environment adaptation information for loss tracking.

Record loss of environment-specific values.

Record loss of error handling specifications.

Record loss of file transfer specifications.

Record loss of resource specifications.

.. py:function:: record_spec_class_loss(json_pointer: str, field: str, spec_object: ~typing.Any, spec_type: str, reason: str, *, severity: str = ‘warn’) -> None :module: wf2wf.loss.core

Record loss of specification class objects.

.. py:function:: reset() -> None :module: wf2wf.loss.core

Clear the in-memory loss buffer.

.. py:function:: write(doc: ~typing.Dict[str, ~typing.Any], path: str | ~pathlib.Path, **kwargs) -> None :module: wf2wf.loss.core

Write loss document to file.

Write a loss document to file.

.. py:module:: wf2wf.loss.context_detection

Format-specific loss detection and environment-specific value handling.

.. py:class:: EnvironmentLossRecorder(source_format: str, target_format: str, target_environment: str) :module: wf2wf.loss.context_detection

Records environment-specific losses with detailed context.

  Record loss of environment-specific value with detailed context.

.. py:attribute:: EnvironmentLossRecorder.source_format :module: wf2wf.loss.context_detection :type: str

.. py:attribute:: EnvironmentLossRecorder.target_environment :module: wf2wf.loss.context_detection :type: str

.. py:attribute:: EnvironmentLossRecorder.target_format :module: wf2wf.loss.context_detection :type: str

.. py:class:: FormatLossDetector(source_format: str, target_format: str) :module: wf2wf.loss.context_detection

Detects format-specific losses during import/export cycles.

.. py:method:: FormatLossDetector.detect_environment_specific_losses(workflow: Workflow) -> List[Dict[str, Any]] :module: wf2wf.loss.context_detection

  Detect environment-specific value losses between formats.

.. py:attribute:: FormatLossDetector.source_format :module: wf2wf.loss.context_detection :type: str

.. py:attribute:: FormatLossDetector.target_format :module: wf2wf.loss.context_detection :type: str

.. py:function:: detect_format_specific_losses(workflow: Workflow, source_format: str, target_format: str) -> List[Dict[str, Any]] :module: wf2wf.loss.context_detection

Detect format-specific losses in a workflow.

Record loss of environment-specific value with format context.

.. py:function:: restore_environment_specific_value(lost_value: Any, field_name: str, expected_type: type, target_environment: str) -> EnvironmentSpecificValue :module: wf2wf.loss.context_detection

Restore an EnvironmentSpecificValue from lost data.

.. py:function:: validate_environment_specific_value(value: ~typing.Any, field_name: str, expected_type: type) -> bool :module: wf2wf.loss.context_detection

Validate that a value is a proper EnvironmentSpecificValue.

.. py:module:: wf2wf.loss.export

Export loss detection and recording for different workflow formats.

Detect and record losses when converting to target format for specific environment.

.. py:function:: record_cwl_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to CWL format.

.. py:function:: record_dagman_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to DAGMan format.

.. py:function:: record_galaxy_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to Galaxy format.

.. py:function:: record_nextflow_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to Nextflow format.

.. py:function:: record_snakemake_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to Snakemake format.

.. py:function:: record_wdl_losses(workflow: ~wf2wf.core.Workflow, target_environment: str, verbose: bool = False) -> None :module: wf2wf.loss.export

Record losses when converting to WDL format.

.. py:module:: wf2wf.loss.import_

Import loss detection and validation for loss sidecars.

.. py:function:: detect_and_record_import_losses(workflow: Workflow, source_format: str, target_environment: str = ‘shared_filesystem’, verbose: bool = False) -> None :module: wf2wf.loss.import_

Detect and record losses when importing from source format.

.. py:function:: validate_loss_entry(entry: ~typing.Dict[str, ~typing.Any], verbose: bool = False) -> bool :module: wf2wf.loss.import_

Validate a single loss entry.

:param entry: Dictionary containing loss entry data :param verbose: Enable verbose logging

:returns: True if the loss entry is valid, False otherwise

.. py:function:: validate_loss_sidecar(loss_data: ~typing.Dict[str, ~typing.Any], workflow, verbose: bool = False) -> bool :module: wf2wf.loss.import_

Validate loss side-car data.

:param loss_data: Dictionary containing loss data :param workflow: Workflow IR object (for checksum validation) :param verbose: Enable verbose logging

:returns: True if the loss side-car is valid, False otherwise

Additional API Documentation¶

API Reference