summaryrefslogtreecommitdiff
path: root/lib/utils.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/utils.py')
-rw-r--r--lib/utils.py93
1 files changed, 0 insertions, 93 deletions
diff --git a/lib/utils.py b/lib/utils.py
index 61cb6f1..0cc9a7b 100644
--- a/lib/utils.py
+++ b/lib/utils.py
@@ -33,99 +33,6 @@ class NpEncoder(json.JSONEncoder):
return super(NpEncoder, self).default(obj)
-class CSVfile:
- def __init__(self):
- pass
-
- def load(self, f):
- observations = list()
- for lineno, line in enumerate(f):
- if lineno == 0:
- param_names = line.split(",")[1:-1]
- attr_names = line.removesuffix("\n").split(",")[-1:]
- else:
- param_values = list(map(soft_cast_int, line.split(",")[1:-1]))
- attr_values = list(
- map(soft_cast_float, line.removesuffix("\n").split(",")[-1:])
- )
- observations.append(
- {
- "name": "CSVFile",
- "param": dict(zip(param_names, param_values)),
- "attribute": dict(zip(attr_names, attr_values)),
- }
- )
- return observations
-
-
-class Logfile:
- def __init__(self):
- pass
-
- def kv_to_param(self, kv_str, cast):
- try:
- key, value = kv_str.split("=")
- value = cast(value)
- return key, value
- except ValueError:
- logger.warning(f"Invalid key-value pair: {kv_str}")
- raise
-
- def kv_to_param_f(self, kv_str):
- return self.kv_to_param(kv_str, soft_cast_float)
-
- def kv_to_param_i(self, kv_str):
- return self.kv_to_param(kv_str, soft_cast_int)
-
- def load(self, f):
- observations = list()
- for lineno, line in enumerate(f):
- m = re.search(r"\[::\] *([^|]*?) *[|] *([^|]*?) *[|] *(.*)", line)
- if m:
- name_str = m.group(1)
- param_str = m.group(2)
- attr_str = m.group(3)
- try:
- param = dict(map(self.kv_to_param_i, param_str.split()))
- attr = dict(map(self.kv_to_param_f, attr_str.split()))
- observations.append(
- {
- "name": name_str,
- "param": param,
- "attribute": attr,
- }
- )
- except ValueError:
- logger.warning(
- f"Error parsing {f}: invalid key-value pair in line {lineno+1}"
- )
- logger.warning(f"Offending entry:\n{line}")
- raise
-
- return observations
-
- def dump(self, observations, f):
- for observation in observations:
- name = observation["name"]
- param = observation["param"]
- attr = observation["attribute"]
-
- param_str = " ".join(
- map(
- lambda kv: f"{kv[0]}={kv[1]}",
- sorted(param.items(), key=lambda kv: kv[0]),
- )
- )
- attr_str = " ".join(
- map(
- lambda kv: f"{kv[0]}={kv[1]}",
- sorted(attr.items(), key=lambda kv: kv[0]),
- )
- )
-
- print(f"[::] {name} | {param_str} | {attr_str}", file=f)
-
-
def running_mean(x: np.ndarray, N: int) -> np.ndarray:
"""
Compute `N` elements wide running average over `x`.