(Modified) Add ict to clt conversion

.github/workflows/run_workflows.yml

@@ -181,6 +181,11 @@ jobs:
       # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
       run: cd workflow-inference-compiler/ && pytest tests/test_rest_core.py -k test_rest_core --cwl_runner cwltool
 
+    - name: PyTest Run ICT to CLT conversion Tests
+      if: always()
+      # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
+      run: cd workflow-inference-compiler/ && pytest tests/test_ict_to_clt_conversion.py -k test_ict_to_clt
+
     # NOTE: The steps below are for repository_dispatch only. For all other steps, please insert above
     # this comment.
 

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "typeguard",
     "pydantic>=2.6",
     "pydantic-settings",
+    "pydantic[email]",
     "docker",
     # FYI also need uidmap to run podman rootless
     "podman",

src/sophios/api/utils/converter.py

@@ -1,12 +1,15 @@
+
 import copy
 from pathlib import Path
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Union
 import json
 import yaml
 from jsonschema import Draft202012Validator
 from sophios.utils_yaml import wic_loader
 
 from sophios.wic_types import Json, Cwl
+from sophios.api.utils.ict.ict_spec.model import ICT
+from sophios.api.utils.ict.ict_spec.cast import cast_to_ict
 
 SCHEMA_FILE = Path(__file__).parent / "input_object_schema.json"
 SCHEMA: Json = {}
@@ -141,3 +144,20 @@ def wfb_to_wic(inp: Json) -> Cwl:
         node = inp_restrict["nodes"][0]
         workflow_temp = node["cwlScript"]
     return workflow_temp
+
+
+def ict_to_clt(ict: Union[ICT, Path, str, dict], network_access: bool = False) -> dict:
+    """
+    Convert ICT to CWL CommandLineTool
+
+    Args:
+        ict (Union[ICT, Path, str, dict]): ICT to convert to CLT. ICT can be an ICT object,
+        a path to a yaml file, or a dictionary containing ICT
+
+    Returns:
+        dict: A dictionary containing the CLT
+    """
+
+    ict_local = ict if isinstance(ict, ICT) else cast_to_ict(ict)
+
+    return ict_local.to_clt(network_access=network_access)

src/sophios/api/utils/ict/ict_spec/cast.py

@@ -0,0 +1,27 @@
+import json
+from pathlib import Path
+from typing import Union
+
+from yaml import safe_load
+
+from sophios.api.utils.ict.ict_spec.model import ICT
+
+
+def cast_to_ict(ict: Union[Path, str, dict]) -> ICT:
+
+    if isinstance(ict, str):
+        ict = Path(ict)
+
+    if isinstance(ict, Path):
+        if str(ict).endswith(".yaml") or str(ict).endswith(".yml"):
+            with open(ict, "r", encoding="utf-8") as f_o:
+                data = safe_load(f_o)
+        elif str(ict).endswith(".json"):
+            with open(ict, "r", encoding="utf-8") as f_o:
+                data = json.load(f_o)
+        else:
+            raise ValueError(f"File extension not supported: {ict}")
+
+        return ICT(**data)
+
+    return ICT(**ict)

src/sophios/api/utils/ict/ict_spec/hardware/__init__.py

@@ -0,0 +1,10 @@
+"""Hardware Requirements for ICT."""
+
+from sophios.api.utils.ict.ict_spec.hardware.objects import (
+    CPU,
+    GPU,
+    HardwareRequirements,
+    Memory,
+)
+
+__all__ = ["CPU", "Memory", "GPU", "HardwareRequirements"]

src/sophios/api/utils/ict/ict_spec/hardware/objects.py

@@ -0,0 +1,108 @@
+# pylint: disable=no-member
+"""Hardware Requirements for ICT."""
+from typing import Annotated, Optional, Union, Any
+
+from pydantic import BaseModel, BeforeValidator, Field
+
+
+def validate_str(s_t: Union[int, float, str]) -> Union[str, None]:
+    """Return a string from int, float, or str."""
+    if s_t is None:
+        return None
+    if isinstance(s_t, str):
+        return s_t
+    if not isinstance(s_t, (int, float)) or isinstance(s_t, bool):
+        raise ValueError("must be an int, float, or str")
+    return str(s_t)
+
+
+StrInt = Annotated[str, BeforeValidator(validate_str)]
+
+
+class CPU(BaseModel):
+    """CPU object."""
+
+    cpu_type: Optional[str] = Field(
+        None,
+        alias="type",
+        description="Any non-standard or specific processor limitations.",
+        examples=["arm64"],
+    )
+    cpu_min: Optional[StrInt] = Field(
+        None,
+        alias="min",
+        description="Minimum requirement for CPU allocation where 1 CPU unit is equivalent to 1 physical CPU core or 1 virtual core.",
+        examples=["100m"],
+    )
+    cpu_recommended: Optional[StrInt] = Field(
+        None,
+        alias="recommended",
+        description="Recommended requirement for CPU allocation for optimal performance.",
+        examples=["200m"],
+    )
+
+
+class Memory(BaseModel):
+    """Memory object."""
+
+    memory_min: Optional[StrInt] = Field(
+        None,
+        alias="min",
+        description="Minimum requirement for memory allocation, measured in bytes.",
+        examples=["129Mi"],
+    )
+    memory_recommended: Optional[StrInt] = Field(
+        None,
+        alias="recommended",
+        description="Recommended requirement for memory allocation for optimal performance.",
+        examples=["200Mi"],
+    )
+
+
+class GPU(BaseModel):
+    """GPU object."""
+
+    gpu_enabled: Optional[bool] = Field(
+        None,
+        alias="enabled",
+        description="Boolean value indicating if the plugin is optimized for GPU.",
+        examples=[False],
+    )
+    gpu_required: Optional[bool] = Field(
+        None,
+        alias="required",
+        description="Boolean value indicating if the plugin requires a GPU to run.",
+        examples=[False],
+    )
+    gpu_type: Optional[str] = Field(
+        None,
+        alias="type",
+        description="	Any identifying label for GPU hardware specificity.",
+        examples=["cuda11"],
+    )
+
+
+ATTRIBUTES = [
+    "cpu_type",
+    "cpu_min",
+    "cpu_recommended",
+    "memory_min",
+    "memory_recommended",
+    "gpu_enabled",
+    "gpu_required",
+    "gpu_type",
+]
+
+
+class HardwareRequirements(BaseModel):
+    """HardwareRequirements object."""
+
+    cpu: Optional[CPU] = Field(None, description="CPU requirements.")
+    memory: Optional[Memory] = Field(None, description="Memory requirements.")
+    gpu: Optional[GPU] = Field(None, description="GPU requirements.")
+
+    def __getattribute__(self, name: str) -> Any:
+        """Get attribute."""
+        if name in ATTRIBUTES:
+            return super().__getattribute__(name.split("_")[0]).__getattribute__(name)
+        return super().__getattribute__(name)

src/sophios/api/utils/ict/ict_spec/io/__init__.py

@@ -0,0 +1,5 @@
+"""IO objects."""
+
+from .objects import IO
+
+__all__ = ["IO"]

src/sophios/api/utils/ict/ict_spec/io/objects.py

@@ -0,0 +1,148 @@
+"""IO objects for ICT."""
+
+import enum
+from typing import Optional, Union, Any
+
+from pydantic import BaseModel, Field
+
+
+CWL_IO_DICT: dict[str, str] = {
+    "string": "string",
+    "number": "double",
+    "array": "string",
+    "boolean": "boolean",
+    # TODO: File vs Directory?
+}
+
+
+class TypesEnum(str, enum.Enum):
+    """Types enum for ICT IO."""
+
+    STRING = "string"
+    NUMBER = "number"
+    ARRAY = "array"
+    BOOLEAN = "boolean"
+    PATH = "path"
+
+
+# def _get_cwl_type(io_name: str, io_type: str) -> str:
+def _get_cwl_type(io_type: str) -> str:
+    """Return the CWL type from the ICT IO type."""
+    if io_type == "path":
+        # NOTE: for now, default to directory
+        # this needs to be addressed
+        # path could be File or Directory
+        return "Directory"
+        # if bool(re.search("dir", io_name, re.I)):
+        #     return "Directory"
+        # return "File"
+    return CWL_IO_DICT[io_type]
+
+
+class IO(BaseModel):
+    """IO BaseModel."""
+
+    name: str = Field(
+        description=(
+            "Unique input or output name for this plugin, case-sensitive match to"
+            "corresponding variable expected by tool."
+        ),
+        examples=["thresholdtype"],
+    )
+    io_type: TypesEnum = Field(
+        ...,
+        alias="type",
+        description="Defines the parameter passed to the ICT tool based on broad categories of basic types.",
+        examples=["string"],
+    )
+    description: Optional[str] = Field(
+        None,
+        description="Short text description of expected value for field.",
+        examples=["Algorithm type for thresholding"],
+    )
+    defaultValue: Optional[Any] = Field(
+        None,
+        description="Optional default value.",
+        examples=["42"],
+    )
+    required: bool = Field(
+        description="Boolean (true/false) value indicating whether this "
+        + "field needs an associated value.",
+        examples=["true"],
+    )
+    io_format: Union[list[str], dict] = Field(
+        ...,
+        alias="format",
+        description="Defines the actual value(s) that the input/output parameter"
+        + "represents using an ontology schema.",
+    )  # TODO ontology
+
+    @property
+    def _is_optional(self) -> str:
+        """Return '' if required, '?' if default exists, else '?'."""
+        if self.defaultValue is not None:
+            return "?"
+        if self.required:
+            return ""
+
+        return "?"
+
+    def convert_uri_format(self, uri_format: Any) -> str:
+        """Convert to cwl format
+        Args:
+            format (_type_): _description_
+        """
+        return f"edam:format_{uri_format.split('_')[-1]}"
+
+    def _input_to_cwl(self) -> dict:
+        """Convert inputs to CWL."""
+        cwl_dict_ = {
+            "inputBinding": {"prefix": f"--{self.name}"},
+            # "type": f"{_get_cwl_type(self.name, self.io_type)}{self._is_optional}",
+            "type": f"{_get_cwl_type(self.io_type)}{self._is_optional}",
+        }
+
+        if (
+            isinstance(self.io_format, dict)
+            and self.io_format.get("uri", None) is not None  # pylint: disable=no-member
+        ):
+            # pylint: disable-next=unsubscriptable-object
+            cwl_dict_["format"] = self.convert_uri_format(self.io_format["uri"])
+        if self.defaultValue is not None:
+            cwl_dict_["default"] = self.defaultValue
+        return cwl_dict_
+
+    def _output_to_cwl(self, inputs: Any) -> dict:
+        """Convert outputs to CWL."""
+        if self.io_type == "path":
+            if self.name in inputs:
+                if (
+                    not isinstance(self.io_format, list)
+                    and self.io_format["term"].lower()
+                    == "directory"  # pylint: disable=unsubscriptable-object
+                ):
+                    cwl_type = "Directory"
+                elif (
+                    not isinstance(self.io_format, list)
+                    and self.io_format["term"].lower()
+                    == "file"  # pylint: disable=unsubscriptable-object
+                ):
+                    cwl_type = "File"
+                else:
+                    cwl_type = "File"
+
+                cwl_dict_ = {
+                    "outputBinding": {"glob": f"$(inputs.{self.name}.basename)"},
+                    "type": cwl_type,
+                }
+                if (
+                    not isinstance(self.io_format, list)
+                    and self.io_format.get("uri", None)
+                    is not None  # pylint: disable=no-member
+                ):
+                    # pylint: disable-next=unsubscriptable-object
+                    cwl_dict_["format"] = self.convert_uri_format(self.io_format["uri"])
+                return cwl_dict_
+
+            raise ValueError(f"Output {self.name} not found in inputs")
+        raise NotImplementedError(f"Output not supported {self.name}")

src/sophios/api/utils/ict/ict_spec/metadata/__init__.py

@@ -0,0 +1,5 @@
+"""Metadata objects."""
+
+from .objects import Metadata
+
+__all__ = ["Metadata"]