/
parser.py
executable file
·89 lines (73 loc) · 2.9 KB
/
parser.py
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
from collections import namedtuple, defaultdict
import time
from webapp.apps.comp import actions
from webapp.apps.comp.compute import Compute
from webapp.apps.comp.displayer import Displayer
from webapp.apps.comp.exceptions import AppError
from webapp.apps.comp.models import Inputs
ParamData = namedtuple("ParamData", ["name", "data"])
class ParameterLookUpException(Exception):
pass
class BaseParser:
def __init__(
self, project, displayer, clean_inputs, compute=None, **valid_meta_params
):
self.project = project
self.clean_inputs = clean_inputs
self.compute = compute or Compute()
self.valid_meta_params = valid_meta_params
for param, value in valid_meta_params.items():
setattr(self, param, value)
defaults = displayer.package_defaults()
self.grouped_defaults = defaults["model_parameters"]
self.flat_defaults = {
k: v for _, sect in self.grouped_defaults.items() for k, v in sect.items()
}
@staticmethod
def append_errors_warnings(errors_warnings, append_func, defaults=None):
"""
Appends warning/error messages to some object, append_obj, according to
the provided function, append_func
"""
for action in ["warnings", "errors"]:
for param in errors_warnings[action]:
msg = errors_warnings[action][param]
append_func(param, msg, defaults)
def parse_parameters(self):
errors_warnings = {
sect: {"errors": {}, "warnings": {}}
for sect in list(self.grouped_defaults) + ["GUI", "API"]
}
adjustment = {sect: {} for sect in self.grouped_defaults}
return errors_warnings, adjustment
def post(self, errors_warnings, params):
data = {
"meta_param_dict": self.valid_meta_params,
"adjustment": params,
"errors_warnings": errors_warnings,
}
job_id, queue_length = self.compute.submit_job(
data, self.project.worker_ext(action=actions.PARSE)
)
return job_id, queue_length
class Parser:
pass
class APIParser(BaseParser):
def parse_parameters(self):
errors_warnings, adjustment = super().parse_parameters()
extra_keys = set(self.clean_inputs.keys() - self.grouped_defaults.keys())
if extra_keys:
errors_warnings["API"]["errors"] = {
"extra_keys": [f"Has extra sections: {' ,'.join(extra_keys)}"]
}
for sect in adjustment:
adjustment[sect].update(self.clean_inputs.get(sect, {}))
# kick off async parsing
job_id, queue_length = self.post(errors_warnings, adjustment)
return {
"job_id": job_id,
"queue_length": queue_length,
"adjustment": adjustment,
"errors_warnings": errors_warnings,
"custom_adjustment": None,
}