1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239
|
from enum import Enum
from typing import (
List,
Optional,
Union,
)
from pydantic import (
Field,
)
from typing_extensions import (
Annotated,
Literal,
NotRequired,
TypedDict,
)
from ._base import ToolSourceBaseModel
class Container(ToolSourceBaseModel):
type: Literal["docker", "singularity"]
container_id: str
class Requirement(ToolSourceBaseModel):
type: Literal["package", "set_environment"]
class ContainerRequirement(ToolSourceBaseModel):
type: Literal["container"]
container: Container
class PackageRequirement(Requirement):
type: Literal["package"]
name: str
version: Optional[str]
class SetEnvironmentRequirement(Requirement):
type: Literal["set_environment"]
environment: str
cores_min_description = "Minimum reserved number of CPU cores."
cores_max_description = "Maximum reserved number of CPU cores."
cores_description = """May be a fractional value to indicate to a scheduling algorithm that one core can be allocated to multiple jobs. For example, a value of 0.25 indicates that up to 4 jobs may run in parallel on 1 core. A value of 1.25 means that up to 3 jobs can run on a 4 core system (4/1.25 ≈ 3).
The reported number of CPU cores reserved for the process is a non-zero integer calculated by rounding up the cores request to the next whole number.
"""
ram_min_description = "Minimum reserved RAM in mebibytes (2**20)."
ram_max_description = "Maximum reserved RAM in mebibytes (2**20)."
ram_description = """May be a fractional value. If so, the actual RAM request is rounded up to the next whole number. The reported amount of RAM reserved for the process is a non-zero integer."""
class ResourceRequirement(ToolSourceBaseModel):
type: Literal["resource"]
cores_min: Annotated[
Union[int, float, None], Field(description=f"{cores_min_description}\n{cores_description}")
] = 1
cores_max: Annotated[
Union[int, float, None], Field(description=f"{cores_max_description}\n{cores_description}")
] = None
ram_min: Annotated[Union[int, float, None], Field(description=f"{ram_min_description}\n{ram_description}")] = 256
ram_max: Annotated[Union[int, float, None], Field(description=f"{ram_max_description}\n{ram_description}")] = None
tmpdir_min: Optional[Union[int, float]] = None
tmpdir_max: Optional[Union[int, float]] = None
cuda_version_min: Optional[Union[int, float]] = None
cuda_compute_capability: Optional[Union[int, float]] = None
gpu_memory_min: Optional[Union[int, float]] = None
cuda_device_count_min: Optional[Union[int, float]] = None
cuda_device_count_max: Optional[Union[int, float]] = None
shm_size: Optional[Union[int, float]] = None
class JavascriptRequirement(ToolSourceBaseModel):
type: Literal["javascript"]
expression_lib: Optional[
List[
Annotated[
str,
Field(
title="expression_lib",
description="Provide Javascript/ECMAScript 5.1 code here that will be available for expressions inside the `shell_command` field.",
examples=[
r"""function pickValue() {
if (inputs.conditional_parameter.test_parameter == "a") {
return inputs.conditional_parameter.integer_parameter
} else {
return inputs.conditional_parameter.boolean_parameter
}
}"""
],
),
]
]
]
class XrefDict(TypedDict):
value: str
type: str
class TemplateConfigFile(ToolSourceBaseModel):
content: str
name: Optional[str] = None
filename: Optional[str] = None
class InputConfigFileContent(ToolSourceBaseModel):
format: Literal["json"] = "json"
handle_files: Optional[Literal["paths", "staging_path_and_source_path"]] = None
type: Literal["inputs"] = "inputs"
class InputConfigFile(ToolSourceBaseModel):
name: Optional[str] = None
content: InputConfigFileContent
filename: Optional[str] = None
class FileSourceConfigFileContent(ToolSourceBaseModel):
type: Literal["files"] = "files"
class FileSourceConfigFile(ToolSourceBaseModel):
name: Optional[str]
filename: Optional[str] = None
content: FileSourceConfigFileContent
class XmlTemplateConfigFile(TemplateConfigFile):
eval_engine: Literal["cheetah"] = "cheetah"
class YamlTemplateConfigFile(TemplateConfigFile):
eval_engine: Literal["ecmascript"] = "ecmascript"
class Citation(ToolSourceBaseModel):
type: str
content: str
class HelpContent(ToolSourceBaseModel):
format: Literal["restructuredtext", "plain_text", "markdown"]
content: str
class OutputCompareType(str, Enum):
diff = "diff"
re_match = "re_match"
sim_size = "sim_size"
re_match_multiline = "re_match_multiline"
contains = "contains"
image_diff = "image_diff"
class DrillDownOptionsDict(TypedDict):
name: Optional[str]
value: str
options: List["DrillDownOptionsDict"]
selected: bool
# For fields... just implementing a subset of CWL for Galaxy flavors of these objects
# so far.
CwlType = Literal["File", "null", "boolean", "int", "float", "string"]
FieldType = Union[CwlType, List[CwlType]]
class FieldDict(TypedDict):
name: str
type: FieldType
format: NotRequired[Optional[str]]
JsonTestDatasetDefDict = TypedDict(
"JsonTestDatasetDefDict",
{
"class": Literal["File"],
"path": NotRequired[Optional[str]],
"location": NotRequired[Optional[str]],
"name": NotRequired[Optional[str]],
"dbkey": NotRequired[Optional[str]],
"filetype": NotRequired[Optional[str]],
"composite_data": NotRequired[Optional[List[str]]],
"tags": NotRequired[Optional[List[str]]],
},
)
JsonTestCollectionDefElementDict = Union[
"JsonTestCollectionDefDatasetElementDict", "JsonTestCollectionDefCollectionElementDict"
]
JsonTestCollectionDefDatasetElementDict = TypedDict(
"JsonTestCollectionDefDatasetElementDict",
{
"identifier": str,
"class": Literal["File"],
"path": NotRequired[Optional[str]],
"location": NotRequired[Optional[str]],
"name": NotRequired[Optional[str]],
"dbkey": NotRequired[Optional[str]],
"filetype": NotRequired[Optional[str]],
"composite_data": NotRequired[Optional[List[str]]],
"tags": NotRequired[Optional[List[str]]],
},
)
BaseJsonTestCollectionDefCollectionElementDict = TypedDict(
"BaseJsonTestCollectionDefCollectionElementDict",
{
"class": Literal["Collection"],
"collection_type": Optional[str],
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
},
)
JsonTestCollectionDefCollectionElementDict = TypedDict(
"JsonTestCollectionDefCollectionElementDict",
{
"identifier": str,
"class": Literal["Collection"],
"collection_type": Optional[str],
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
},
)
JsonTestCollectionDefDict = TypedDict(
"JsonTestCollectionDefDict",
{
"class": Literal["Collection"],
"collection_type": Optional[str],
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
"name": NotRequired[Optional[str]],
},
)
|