1
|
|
|
import enum |
|
|
|
|
2
|
|
|
import os |
3
|
|
|
from inspect import cleandoc |
4
|
|
|
from typing import Optional, Iterable, Any, Mapping, Union, Callable, Sequence, Type |
5
|
|
|
|
6
|
|
|
import typer |
|
|
|
|
7
|
|
|
from typeddfs import TypedDf |
|
|
|
|
8
|
|
|
|
9
|
|
|
from mandos.entry._common_args import T |
10
|
|
|
|
11
|
|
|
|
12
|
|
|
class _Args: |
13
|
|
|
@staticmethod |
14
|
|
|
def _arg(doc: str, *names, default: Optional[T] = None, req: bool = False, **kwargs): |
15
|
|
|
kwargs = dict( |
16
|
|
|
help=cleandoc(doc), |
17
|
|
|
**kwargs, |
18
|
|
|
allow_dash=True, |
19
|
|
|
) |
20
|
|
|
if req: |
|
|
|
|
21
|
|
|
return typer.Argument(default, **kwargs) |
22
|
|
|
else: |
23
|
|
|
return typer.Option(default, *names, **kwargs) |
24
|
|
|
|
25
|
|
|
@staticmethod |
26
|
|
|
def _path( |
27
|
|
|
doc: str, *names, default: Optional[str], f: bool, d: bool, out: bool, req: bool, **kwargs |
|
|
|
|
28
|
|
|
): |
29
|
|
|
# if it's None, we're going to have a special default set afterward, so we'll explain it in the doc |
|
|
|
|
30
|
|
|
if out and default is None: |
31
|
|
|
kwargs = dict(show_default=False, **kwargs) |
32
|
|
|
kwargs = { |
33
|
|
|
**dict( |
34
|
|
|
exists=not out, |
35
|
|
|
dir_okay=d, |
36
|
|
|
file_okay=f, |
37
|
|
|
readable=out, |
38
|
|
|
writable=not out, |
39
|
|
|
), |
40
|
|
|
**kwargs, |
41
|
|
|
} |
42
|
|
|
return _Args._arg(doc, *names, default=default, req=req, **kwargs) |
43
|
|
|
|
44
|
|
|
|
45
|
|
|
class Arg(_Args): |
|
|
|
|
46
|
|
|
@staticmethod |
47
|
|
|
def out_file(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
48
|
|
|
return _Args._path( |
49
|
|
|
doc, *names, default=default, f=True, d=False, out=True, req=True, **kwargs |
50
|
|
|
) |
51
|
|
|
|
52
|
|
|
@staticmethod |
53
|
|
|
def out_dir(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
54
|
|
|
return _Args._path( |
55
|
|
|
doc, *names, default=default, f=True, d=True, out=True, req=True, **kwargs |
56
|
|
|
) |
57
|
|
|
|
58
|
|
|
@staticmethod |
59
|
|
|
def out_path(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
60
|
|
|
return _Args._path( |
61
|
|
|
doc, *names, default=default, f=True, d=True, out=False, req=True, **kwargs |
62
|
|
|
) |
63
|
|
|
|
64
|
|
|
@staticmethod |
65
|
|
|
def in_file(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
66
|
|
|
return _Args._path( |
67
|
|
|
doc, *names, default=default, f=True, d=False, out=False, req=True, **kwargs |
68
|
|
|
) |
69
|
|
|
|
70
|
|
|
@staticmethod |
71
|
|
|
def in_dir(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
72
|
|
|
return _Args._path( |
73
|
|
|
doc, *names, default=default, f=False, d=True, out=False, req=True, **kwargs |
74
|
|
|
) |
75
|
|
|
|
76
|
|
|
@staticmethod |
77
|
|
|
def in_path(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
78
|
|
|
return _Args._path( |
79
|
|
|
doc, *names, default=default, f=True, d=True, out=False, req=True, **kwargs |
80
|
|
|
) |
81
|
|
|
|
82
|
|
|
@staticmethod |
83
|
|
|
def val(doc: str, *names, default: Optional[T] = None, **kwargs): |
|
|
|
|
84
|
|
|
return _Args._arg(doc, *names, default=default, req=True, **kwargs) |
85
|
|
|
|
86
|
|
|
|
87
|
|
|
class Opt(_Args): |
|
|
|
|
88
|
|
|
@staticmethod |
89
|
|
|
def out_file(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
90
|
|
|
return _Args._path( |
91
|
|
|
doc, *names, default=default, f=True, d=False, out=True, req=False, **kwargs |
92
|
|
|
) |
93
|
|
|
|
94
|
|
|
@staticmethod |
95
|
|
|
def out_dir(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
96
|
|
|
return _Args._path( |
97
|
|
|
doc, *names, default=default, f=True, d=True, out=True, req=False, **kwargs |
98
|
|
|
) |
99
|
|
|
|
100
|
|
|
@staticmethod |
101
|
|
|
def out_path(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
102
|
|
|
return _Args._path( |
103
|
|
|
doc, |
104
|
|
|
*names, |
105
|
|
|
default=default, |
106
|
|
|
f=True, |
107
|
|
|
d=True, |
108
|
|
|
out=False, |
109
|
|
|
req=False, |
110
|
|
|
exists=False, |
111
|
|
|
**kwargs, |
112
|
|
|
) |
113
|
|
|
|
114
|
|
|
@staticmethod |
115
|
|
|
def in_file(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
116
|
|
|
return _Args._path( |
117
|
|
|
doc, *names, default=default, f=True, d=False, out=False, req=False, **kwargs |
118
|
|
|
) |
119
|
|
|
|
120
|
|
|
@staticmethod |
121
|
|
|
def in_dir(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
122
|
|
|
return _Args._path( |
123
|
|
|
doc, *names, default=default, f=False, d=True, out=False, req=False, **kwargs |
124
|
|
|
) |
125
|
|
|
|
126
|
|
|
@staticmethod |
127
|
|
|
def in_path(doc: str, *names, default: Optional[str] = None, **kwargs): |
|
|
|
|
128
|
|
|
return _Args._path( |
129
|
|
|
doc, |
130
|
|
|
*names, |
131
|
|
|
default=default, |
132
|
|
|
f=True, |
133
|
|
|
d=True, |
134
|
|
|
out=False, |
135
|
|
|
req=False, |
136
|
|
|
exists=False, |
137
|
|
|
**kwargs, |
138
|
|
|
) |
139
|
|
|
|
140
|
|
|
@staticmethod |
141
|
|
|
def val(doc: str, *names, default: Optional[T] = None, **kwargs): |
|
|
|
|
142
|
|
|
return _Args._arg(doc, *names, default=default, req=False, **kwargs) |
143
|
|
|
|
144
|
|
|
@staticmethod |
145
|
|
|
def flag(doc: str, *names, **kwargs): |
|
|
|
|
146
|
|
|
return _Args._arg(doc, *names, default=False, req=False, **kwargs) |
147
|
|
|
|
148
|
|
|
|
149
|
|
|
class ArgUtils: |
|
|
|
|
150
|
|
|
@classmethod |
151
|
|
|
def definition_bullets(cls, dct: Mapping[Any, Any], colon: str = ": ", indent: int = 12) -> str: |
|
|
|
|
152
|
|
|
joiner = os.linesep * 2 + " " * indent |
153
|
|
|
jesus = [f" - {k}{colon}{v}" for k, v in dct.items()] |
154
|
|
|
return joiner.join(jesus) |
155
|
|
|
|
156
|
|
|
@classmethod |
157
|
|
|
def definition_list(cls, dct: Mapping[Any, Any], colon: str = ": ", sep: str = "; ") -> str: |
|
|
|
|
158
|
|
|
jesus = [f"{k}{colon}{v}" for k, v in dct.items()] |
159
|
|
|
return sep.join(jesus) |
160
|
|
|
|
161
|
|
|
@classmethod |
162
|
|
|
def df_description(cls, tdf: Type[TypedDf]) -> str: |
|
|
|
|
163
|
|
|
req = tdf.get_typing().required_columns |
164
|
|
|
res = [*tdf.get_typing().reserved_index_names, *tdf.get_typing().reserved_columns] |
165
|
|
|
line1 = "Required columns: " + ", ".join(req) |
166
|
|
|
line2 = "Optional columns: " + ", ".join(res) |
167
|
|
|
return line1 + "\n\n" + line2 |
168
|
|
|
|
169
|
|
|
@classmethod |
170
|
|
|
def list( |
|
|
|
|
171
|
|
|
cls, |
|
|
|
|
172
|
|
|
lst: Iterable[Any], |
|
|
|
|
173
|
|
|
attr: Union[None, str, Callable[[Any], Any]] = None, |
|
|
|
|
174
|
|
|
sep: str = "; ", |
|
|
|
|
175
|
|
|
) -> str: |
176
|
|
|
x = [] |
|
|
|
|
177
|
|
|
for v in lst: |
|
|
|
|
178
|
|
|
if attr is None and isinstance(v, enum.Enum): |
179
|
|
|
x += [v.name] |
|
|
|
|
180
|
|
|
elif attr is None: |
181
|
|
|
x += [str(v)] |
|
|
|
|
182
|
|
|
elif isinstance(attr, str): |
183
|
|
|
x += [str(getattr(v, attr))] |
|
|
|
|
184
|
|
|
else: |
185
|
|
|
x += [str(attr(v))] |
|
|
|
|
186
|
|
|
return sep.join(x) |
187
|
|
|
|
188
|
|
|
@classmethod |
189
|
|
|
def parse_taxon_id_or_name(cls, taxon: Union[int, str]) -> Union[int, str]: |
|
|
|
|
190
|
|
|
if isinstance(taxon, str): |
|
|
|
|
191
|
|
|
return taxon |
192
|
|
|
elif isinstance(taxon, str) and taxon.isdigit(): |
193
|
|
|
return int(taxon) |
194
|
|
|
raise ValueError(f"Taxon {taxon} must be an ID or name") |
195
|
|
|
|
196
|
|
|
@classmethod |
197
|
|
|
def parse_taxon_id(cls, taxon: Union[int, str]) -> int: |
|
|
|
|
198
|
|
|
try: |
199
|
|
|
return int(taxon) |
200
|
|
|
except ValueError: |
201
|
|
|
raise ValueError(f"Taxon {taxon} must be an exact ID") from None |
202
|
|
|
|
203
|
|
|
@classmethod |
204
|
|
|
def parse_taxa(cls, taxa: str) -> Sequence[Union[int, str]]: |
|
|
|
|
205
|
|
|
taxa = [t.strip() for t in taxa.split(",")] |
206
|
|
|
return [ArgUtils.parse_taxon_id_or_name(t) for t in taxa] |
207
|
|
|
|
208
|
|
|
|
209
|
|
|
__all__ = ["Arg", "Opt", "ArgUtils"] |
210
|
|
|
|