94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
|
"""Estrae l'ultimo ciclo completo da un CSV di voltammetria ciclica.
|
||
|
|
|
||
|
|
Un "ciclo" e' definito dal susseguirsi di due inversioni di direzione dello
|
||
|
|
stesso tipo del potenziale (es. massimo -> minimo -> massimo). Si parte
|
||
|
|
dall'ultimo punto del file e si risale fino alla precedente inversione dello
|
||
|
|
stesso tipo (stesso segno di curvatura), cosi' da ottenere un ciclo chiuso
|
||
|
|
senza fermarsi su un punto vicino ma non ancora al vero picco.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import csv
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
SUFFIX = "_ultimo_ciclo"
|
||
|
|
|
||
|
|
|
||
|
|
def read_rows(path: Path) -> list[tuple[str, str]]:
|
||
|
|
with path.open("r", encoding="utf-8", newline="") as f:
|
||
|
|
reader = csv.reader(f, delimiter=";")
|
||
|
|
header = next(reader)
|
||
|
|
rows = [(row[0], row[1]) for row in reader]
|
||
|
|
return header, rows
|
||
|
|
|
||
|
|
|
||
|
|
def find_turning_points(potentials: list[float]) -> list[tuple[int, str]]:
|
||
|
|
n = len(potentials)
|
||
|
|
turning: list[tuple[int, str]] = []
|
||
|
|
for i in range(1, n - 1):
|
||
|
|
if potentials[i - 1] < potentials[i] > potentials[i + 1]:
|
||
|
|
turning.append((i, "max"))
|
||
|
|
elif potentials[i - 1] > potentials[i] < potentials[i + 1]:
|
||
|
|
turning.append((i, "min"))
|
||
|
|
|
||
|
|
if n >= 2:
|
||
|
|
last_diff = potentials[-1] - potentials[-2]
|
||
|
|
if last_diff > 0:
|
||
|
|
turning.append((n - 1, "max"))
|
||
|
|
elif last_diff < 0:
|
||
|
|
turning.append((n - 1, "min"))
|
||
|
|
|
||
|
|
return turning
|
||
|
|
|
||
|
|
|
||
|
|
def last_cycle_bounds(potentials: list[float]) -> tuple[int, int]:
|
||
|
|
turning = find_turning_points(potentials)
|
||
|
|
if len(turning) < 2:
|
||
|
|
raise ValueError("dati insufficienti per identificare un ciclo completo")
|
||
|
|
|
||
|
|
end_idx, end_type = turning[-1]
|
||
|
|
start_idx = next(
|
||
|
|
(idx for idx, typ in reversed(turning[:-1]) if typ == end_type),
|
||
|
|
None,
|
||
|
|
)
|
||
|
|
if start_idx is None:
|
||
|
|
raise ValueError("impossibile trovare un ciclo chiuso (inversione precedente non trovata)")
|
||
|
|
|
||
|
|
return start_idx, end_idx
|
||
|
|
|
||
|
|
|
||
|
|
def extract_last_cycle(input_path: Path, output_path: Path) -> None:
|
||
|
|
header, rows = read_rows(input_path)
|
||
|
|
potentials = [float(p.replace(",", ".")) for p, _ in rows]
|
||
|
|
|
||
|
|
start_idx, end_idx = last_cycle_bounds(potentials)
|
||
|
|
last_cycle_rows = rows[start_idx:end_idx + 1]
|
||
|
|
|
||
|
|
with output_path.open("w", encoding="utf-8", newline="") as f:
|
||
|
|
writer = csv.writer(f, delimiter=";")
|
||
|
|
writer.writerow(header)
|
||
|
|
writer.writerows(last_cycle_rows)
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv: list[str]) -> None:
|
||
|
|
folder = Path(argv[1]) if len(argv) > 1 else Path(__file__).parent
|
||
|
|
csv_files = sorted(
|
||
|
|
p for p in folder.glob("*.csv") if not p.stem.endswith(SUFFIX)
|
||
|
|
)
|
||
|
|
|
||
|
|
if not csv_files:
|
||
|
|
print(f"Nessun file .csv trovato in {folder}")
|
||
|
|
return
|
||
|
|
|
||
|
|
for csv_file in csv_files:
|
||
|
|
output_path = csv_file.with_name(csv_file.stem + SUFFIX + ".csv")
|
||
|
|
try:
|
||
|
|
extract_last_cycle(csv_file, output_path)
|
||
|
|
print(f"OK {csv_file.name} -> {output_path.name}")
|
||
|
|
except ValueError as e:
|
||
|
|
print(f"SKIP {csv_file.name}: {e}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main(sys.argv)
|