From 818bda26cb68b16b586d2193d1959b851dc32188 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Thu, 18 Jun 2026 01:48:55 +0300 Subject: [PATCH] rename: sort files into races/// automatically rename_unprocessed now takes races_dir; when set, creates races//Practice|Qualifying|Race/ and moves the file there. GUI and CLI both pass _RACES_DIR so renaming from any folder always ends up in the right place. --- telemetry/__main__.py | 4 +++- telemetry/gui.py | 2 +- telemetry/rename.py | 52 +++++++++++++++++++++++++++++++++++++------ tests/test_rename.py | 29 ++++++++++++++++++++++-- 4 files changed, 76 insertions(+), 11 deletions(-) diff --git a/telemetry/__main__.py b/telemetry/__main__.py index dd6b13d..c4f43df 100644 --- a/telemetry/__main__.py +++ b/telemetry/__main__.py @@ -7,6 +7,8 @@ from .report_common import report_path from . import report_technique, report_setup, report_compare, report_race from .rename import rename_unprocessed +_RACES_DIR = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'races')) + def _print_result(abs_path: str, tokens: int) -> None: print(abs_path) @@ -28,7 +30,7 @@ def _run_rename(targets: list[str]) -> None: print('Error: no CSV targets resolved', file=sys.stderr) sys.exit(2) - results = rename_unprocessed(targets) + results = rename_unprocessed(targets, races_dir=_RACES_DIR) for r in results: if r.status == 'renamed': print(f'renamed: {r.old} -> {r.new}') diff --git a/telemetry/gui.py b/telemetry/gui.py index ed339da..44eb545 100644 --- a/telemetry/gui.py +++ b/telemetry/gui.py @@ -204,7 +204,7 @@ class _App: def _worker_rename(self, d: str) -> None: try: - results = rename_unprocessed([d]) + results = rename_unprocessed([d], races_dir=_RACES_DIR) if not results: self._push('No CSV files found') for r in results: diff --git a/telemetry/rename.py b/telemetry/rename.py index 0f7c28d..4b9075b 100644 --- a/telemetry/rename.py +++ b/telemetry/rename.py @@ -7,6 +7,23 @@ from .parser import _parse_header_row, _parse_value_row _SESSION_PAT = re.compile(r'^(FP[1-3]|P[1-3]|Q[1-3]|R|Race|Sprint)$', re.IGNORECASE) +def _session_folder(event: str) -> str: + e = event.upper() + if re.match(r'^(FP[1-3]|P[1-3])$', e): + return 'Practice' + if re.match(r'^(SQ[1-3]|Q[1-3])$', e): + return 'Qualifying' + if re.match(r'^SPRINT$', e): + return 'Sprint' + if re.match(r'^(R|RACE)$', e): + return 'Race' + return 'Other' + + +def _safe_dirname(name: str) -> str: + return re.sub(r'[<>:"/\\|?*]', '_', name).strip(' .') + + def lap_token_present(stem: str) -> bool: return any(re.fullmatch(r'L\d+', t) for t in stem.split('_')) @@ -68,6 +85,23 @@ def read_session_type(path: str) -> str: return session.get('event', '').strip() +def read_metadata(path: str) -> tuple[int, str, str]: + """Returns (lap_number, event_type, track_name).""" + with open(path, encoding='utf-8') as f: + lines = [f.readline() for _ in range(5)] + sess_keys = _parse_header_row(lines[1]) + sess_vals = _parse_value_row(lines[2]) + session = dict(zip(sess_keys, sess_vals)) + track_keys = _parse_header_row(lines[3]) + track_vals = _parse_value_row(lines[4]) + track = dict(zip(track_keys, track_vals)) + return ( + int(float(track['Lap'])), + session.get('event', '').strip(), + session.get('track', '').strip(), + ) + + @dataclass(frozen=True) class RenameResult: old: str @@ -76,7 +110,7 @@ class RenameResult: detail: str -def rename_unprocessed(targets: list[str]) -> list[RenameResult]: +def rename_unprocessed(targets: list[str], races_dir: str | None = None) -> list[RenameResult]: paths: list[str] = [] for t in targets: t = os.path.abspath(t) @@ -101,8 +135,7 @@ def rename_unprocessed(targets: list[str]) -> list[RenameResult]: results.append(RenameResult(name, None, 'skipped', 'already processed')) continue try: - lap = read_lap_number(p) - session_type = read_session_type(p) + lap, session_type, track_name = read_metadata(p) except Exception as e: results.append(RenameResult(name, None, 'error', str(e))) continue @@ -110,14 +143,19 @@ def rename_unprocessed(targets: list[str]) -> list[RenameResult]: results.append(RenameResult(name, None, 'error', 'session type not found in CSV')) continue new_name = insert_tokens(name, session_type, lap) - dest = os.path.join(os.path.dirname(p), new_name) + if races_dir is not None: + dest_dir = os.path.join(races_dir, _safe_dirname(track_name), _session_folder(session_type)) + os.makedirs(dest_dir, exist_ok=True) + else: + dest_dir = os.path.dirname(p) + dest = os.path.join(dest_dir, new_name) if os.path.exists(dest): - results.append(RenameResult(name, new_name, 'error', 'target exists')) + results.append(RenameResult(name, dest, 'error', 'target exists')) continue try: os.rename(p, dest) except OSError as e: - results.append(RenameResult(name, new_name, 'error', str(e))) + results.append(RenameResult(name, dest, 'error', str(e))) continue - results.append(RenameResult(name, new_name, 'renamed', '')) + results.append(RenameResult(name, dest, 'renamed', '')) return results diff --git a/tests/test_rename.py b/tests/test_rename.py index 97c750f..a8e0cdd 100644 --- a/tests/test_rename.py +++ b/tests/test_rename.py @@ -3,7 +3,8 @@ import shutil import pytest from telemetry.rename import ( lap_token_present, insert_lap_token, session_token_present, - insert_tokens, read_lap_number, read_session_type, rename_unprocessed, + insert_tokens, read_lap_number, read_session_type, read_metadata, + rename_unprocessed, ) @@ -77,6 +78,15 @@ def test_read_session_type_sample(sample_path): assert read_session_type(sample_path) == 'P1' +# --- read_metadata --- + +def test_read_metadata_sample(sample_path): + lap, event, track = read_metadata(sample_path) + assert lap == 7 + assert event == 'P1' + assert track != '' + + # --- rename_unprocessed --- def test_rename_unprocessed_renames(tmp_path, sample_path): @@ -87,7 +97,7 @@ def test_rename_unprocessed_renames(tmp_path, sample_path): r = results[0] assert r.status == "renamed" assert r.new is not None and "L7" in r.new and "P1" in r.new - assert os.path.exists(tmp_path / r.new) + assert os.path.exists(r.new) def test_rename_unprocessed_skips_already_tagged(tmp_path, sample_path): @@ -114,3 +124,18 @@ def test_rename_unprocessed_dir_target(tmp_path, sample_path): results = rename_unprocessed([str(tmp_path)]) assert len(results) == 1 assert results[0].status == "renamed" + + +def test_rename_unprocessed_moves_to_races_dir(tmp_path, sample_path): + src = str(tmp_path / "zandvoort_74.074_ferrari.csv") + shutil.copy(sample_path, src) + races_dir = str(tmp_path / 'races') + results = rename_unprocessed([src], races_dir=races_dir) + assert len(results) == 1 + r = results[0] + assert r.status == 'renamed' + assert r.new is not None and 'P1' in r.new and 'L7' in r.new + assert os.path.exists(r.new) + assert r.new.startswith(races_dir) + assert 'Practice' in r.new + assert not os.path.exists(src)