32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
"""Lettura del CSV: tempo in secondi (crescente) e serie come array."""
|
|
|
|
import numpy as np
|
|
|
|
|
|
class ErroreDati(Exception):
|
|
pass
|
|
|
|
|
|
def leggi_csv(percorso, cfg):
|
|
d = cfg["dati"]
|
|
try:
|
|
with open(percorso, encoding="utf-8") as f:
|
|
intestazione = f.readline().strip().split(d["separatore"])
|
|
righe = np.genfromtxt(f, delimiter=d["separatore"])
|
|
except OSError as e:
|
|
raise ErroreDati(f"impossibile leggere '{percorso}': {e.strerror}")
|
|
colonne = [d["colonna_tempo"], *d["serie"]]
|
|
for c in colonne:
|
|
if c not in intestazione:
|
|
raise ErroreDati(f"colonna '{c}' assente in '{percorso}' (trovate: {', '.join(intestazione)})")
|
|
righe = np.atleast_2d(righe)
|
|
if righe.size == 0 or np.isnan(righe).any():
|
|
raise ErroreDati(f"'{percorso}': dati vuoti o non numerici")
|
|
t = righe[:, intestazione.index(d["colonna_tempo"])]
|
|
if d["unita_tempo"] == "ms":
|
|
t = t / 1000.0
|
|
if np.any(np.diff(t) <= 0):
|
|
raise ErroreDati("il tempo non è strettamente crescente (ipotesi H1)")
|
|
serie = {s: righe[:, intestazione.index(s)] for s in d["serie"]}
|
|
return t, serie
|