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
|
from typing import (
Any,
Dict,
Iterator,
List,
Optional,
Tuple,
Union,
)
class InputsBuilder:
""" """
def __init__(self) -> None:
self._input_dict: Dict[str, Any] = {}
def set(self, name: str, input: Any) -> "InputsBuilder":
self._input_dict[name] = input
return self
def set_param(self, name: str, value: Any) -> "InputsBuilder":
return self.set(name, param(value=value))
def set_dataset_param(self, name: str, value: str, src: str = "hda") -> "InputsBuilder":
return self.set(name, dataset(value, src=src))
def to_dict(self) -> Dict[str, Any]:
values = {}
for key, value in self.flat_iter():
if hasattr(value, "value"):
value = value.value
values[key] = value
return values
def flat_iter(self, prefix: Optional[str] = None) -> Iterator[Tuple[str, Any]]:
for key, value in self._input_dict.items():
effective_key = key if prefix is None else f"{prefix}|{key}"
if hasattr(value, "flat_iter"):
yield from value.flat_iter(effective_key)
else:
yield effective_key, value
class RepeatBuilder:
def __init__(self) -> None:
self._instances: List[InputsBuilder] = []
def instance(self, inputs: InputsBuilder) -> "RepeatBuilder":
self._instances.append(inputs)
return self
def flat_iter(self, prefix: str) -> Iterator[Tuple[str, Any]]:
for index, instance in enumerate(self._instances):
index_prefix = f"{prefix}_{index}"
yield from instance.flat_iter(index_prefix)
class Param:
def __init__(self, value: Any) -> None:
self.value = value
class DatasetParam(Param):
def __init__(self, value: Union[Dict[str, str], str], src: str = "hda") -> None:
if not isinstance(value, dict):
value = dict(src=src, id=value)
super().__init__(value)
inputs = InputsBuilder
repeat = RepeatBuilder
conditional = InputsBuilder
param = Param
dataset = DatasetParam
__all__ = ("inputs", "repeat", "conditional", "param")
|
