diff --git a/cv_last_cycle.py b/cv_last_cycle.py new file mode 100644 index 0000000..42ce8f8 --- /dev/null +++ b/cv_last_cycle.py @@ -0,0 +1,93 @@ +"""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)