115 lines
3.4 KiB
Python
Executable File
115 lines
3.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
invoke with the path to a meas.csv file for the stacked_core 51-xx or later demos
|
|
to extract higher-level info from them.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import re
|
|
|
|
from natsort import natsorted
|
|
|
|
from stacked_cores import load_csv, labeled_rows, last_row_before_t, extract_m
|
|
from stacked_cores_39xx import extract_polarity
|
|
|
|
class MeasRow:
|
|
def __init__(self, t_sec: float, m: list):
|
|
self.t_sec = t_sec
|
|
self.m = m
|
|
|
|
def __repr__(self) -> str:
|
|
m = ", ".join(f"{v:6}" for v in self.m)
|
|
return f"MeasRow({self.t_sec}, [{m}])"
|
|
|
|
@staticmethod
|
|
def from_dict(row_data: dict) -> 'MeasRow':
|
|
t_sec = row_data["time"]
|
|
m = [int(m + 0.5) for m in extract_m(row_data)]
|
|
return MeasRow(t_sec, m)
|
|
|
|
def format_float_tuple(t: tuple) -> str:
|
|
formatted_elems = [f"{e:= 05.3f}," for e in t]
|
|
return f"({' '.join(formatted_elems)})"
|
|
|
|
def format_list(l: list) -> str:
|
|
if len(l) == 0: return "[]"
|
|
if len(l) == 1: return f"{l}"
|
|
formatted_elems = [f" {e}," for e in l]
|
|
return "\n".join(["["] + formatted_elems + ["]"])
|
|
|
|
def indented(s: str) -> str:
|
|
return s.replace('\n', '\n ')
|
|
|
|
class ParameterizedMeas:
|
|
def __init__(self, meas = None):
|
|
self.meas = meas or {}
|
|
|
|
def add_meas(self, params: tuple, meas_rows: list):
|
|
self.meas[tuple(params)] = meas_rows
|
|
|
|
def all_rows(self) -> list:
|
|
# this is just `sum(self.meas.values())` but python is a idiot
|
|
rows = []
|
|
for mrows in self.meas.values():
|
|
rows.extend(mrows)
|
|
return rows
|
|
|
|
def runs(self) -> list:
|
|
return self.meas.values()
|
|
|
|
def __repr__(self) -> str:
|
|
meas_entries = "\n".join(
|
|
f" {format_float_tuple(k)}: {indented(format_list(v))}," for (k, v) in natsorted(self.meas.items())
|
|
)
|
|
return f"ParameterizedMeas({{\n{meas_entries}\n}})"
|
|
|
|
def extract_rows(path: str, times: list) -> list:
|
|
header, raw_rows = load_csv(path)
|
|
rows = labeled_rows(header, raw_rows)
|
|
|
|
meas_rows = []
|
|
for t in times:
|
|
row = last_row_before_t(rows, t)
|
|
if not row: return None
|
|
meas_rows.append(MeasRow.from_dict(row))
|
|
|
|
# validate the sim has run to completion
|
|
if meas_rows[-1].t_sec < 0.95 * t: return None
|
|
meas_rows[-1].t_sec = t # make pretty
|
|
|
|
return meas_rows
|
|
|
|
def parse_param(s: str) -> float:
|
|
""" parse a parameter in the form of 'p050' or 'n0015' or '000' """
|
|
if s == "000":
|
|
return 0.0
|
|
|
|
sign = {'n': -1, 'p': 1}[s[0]]
|
|
mag = int(s[1:])
|
|
max_mag = 10**(len(s[1:]) - 1)
|
|
return sign * mag / max_mag
|
|
|
|
def extract_params(pstr: str) -> list:
|
|
""" extract parameters from a string like -n100-000 """
|
|
pieces = [p for p in pstr.split("-") if p]
|
|
return [parse_param(p) for p in pieces]
|
|
|
|
def extract_parameterized_meas(stem: str, times: list) -> ParameterizedMeas:
|
|
""" given some stem, parse all parameterized measurements associated with that stem """
|
|
base_dir, prefix = os.path.split(stem)
|
|
|
|
built = ParameterizedMeas()
|
|
for entry in os.listdir(base_dir):
|
|
if entry.startswith(prefix):
|
|
meas_rows = extract_rows(os.path.join(base_dir, entry, "meas.csv"), times)
|
|
if not meas_rows: continue
|
|
params = extract_params(entry[len(prefix):])
|
|
built.add_meas(params, meas_rows)
|
|
|
|
return built
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(extract_parameterized_meas(sys.argv[1], [float(f) for f in sys.argv[2:]]))
|