From 7306182a4e024a563011efedcf6465bcc11b9c2f Mon Sep 17 00:00:00 2001 From: Mika Date: Thu, 29 Jan 2026 16:23:30 +0000 Subject: [PATCH] Add drift_data_collection/src/drift_data_collection/cli.py --- .../src/drift_data_collection/cli.py | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 drift_data_collection/src/drift_data_collection/cli.py diff --git a/drift_data_collection/src/drift_data_collection/cli.py b/drift_data_collection/src/drift_data_collection/cli.py new file mode 100644 index 0000000..1600504 --- /dev/null +++ b/drift_data_collection/src/drift_data_collection/cli.py @@ -0,0 +1,49 @@ +import argparse +import json +import sys +from pathlib import Path +from typing import List + +from drift_data_collection.core import collect_frozen_runs + + +def _load_run_ids(input_path: Path) -> List[str]: + if not input_path.exists(): + raise FileNotFoundError(f"Eingabedatei nicht gefunden: {input_path}") + with input_path.open('r', encoding='utf-8') as f: + data = json.load(f) + if not isinstance(data, list) or not all(isinstance(x, str) for x in data): + raise ValueError("Eingabedatei muss eine JSON-Liste von Strings enthalten (Run-IDs).") + assert data, "Eingabedatei darf nicht leer sein." + return data + + +def _parse_arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Sammelt Frozen-Run-Daten für Drift-Analysen.") + parser.add_argument('--input', required=True, help='Pfad zur JSON-Datei mit Run-IDs.') + parser.add_argument('--output', required=True, help='Pfad zur Ausgabe-JSON-Datei der gesammelten Daten.') + return parser.parse_args() + + +def main() -> None: + args = _parse_arguments() + input_path = Path(args.input) + output_path = Path(args.output) + + try: + run_ids = _load_run_ids(input_path) + frozen_runs = collect_frozen_runs(run_ids) + + # Validierung der Rückgabe + assert isinstance(frozen_runs, list), "collect_frozen_runs muss eine Liste liefern." + with output_path.open('w', encoding='utf-8') as f: + json.dump(frozen_runs, f, indent=2, default=str) + + print(f"Gesammelte Daten wurden erfolgreich nach {output_path} geschrieben.") + except Exception as e: + print(f"Fehler: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == '__main__': + main() \ No newline at end of file