Add marker_analysis/src/marker_analysis/cli.py

marker_analysis/src/marker_analysis/cli.py

@@ -0,0 +1,68 @@
+import argparse
+import json
+import os
+from typing import Any, List
+from marker_analysis.core import validate_markers, MarkerData
+
+
+def _load_marker_data(file_path: str) -> List[MarkerData]:
+    """Lädt und validiert Marker-Daten aus der angegebenen JSON-Datei."""
+    if not os.path.exists(file_path):
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {file_path}")
+
+    with open(file_path, 'r', encoding='utf-8') as f:
+        try:
+            raw_data = json.load(f)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Ungültiges JSON-Format: {e}") from e
+
+    if not isinstance(raw_data, list):
+        raise ValueError("Eingabedatei muss eine Liste von Objekten enthalten.")
+
+    marker_entries = []
+    for entry in raw_data:
+        try:
+            marker = MarkerData(**entry)
+            marker_entries.append(marker)
+        except Exception as e:
+            raise ValueError(f"Fehler bei Datenvalidierung eines Eintrags: {e}") from e
+
+    assert all(hasattr(m, 'marker_id') and hasattr(m, 'timestamp') and hasattr(m, 'status') for m in marker_entries), (
+        "Ungültige MarkerData-Datenstruktur festgestellt"
+    )
+    return marker_entries
+
+
+def _write_results(output_path: str, results: dict[str, Any]) -> None:
+    os.makedirs(os.path.dirname(output_path), exist_ok=True)
+    with open(output_path, 'w', encoding='utf-8') as f:
+        json.dump(results, f, indent=2)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Validiert Marker-Daten aus eBPF write_pre/write_post Hooks und analysiert retry-freie Reads."
+    )
+    parser.add_argument(
+        '--input',
+        required=True,
+        help='Pfad zur JSON-Datei mit Marker-Daten.'
+    )
+    parser.add_argument(
+        '--output',
+        required=False,
+        default='output/validation_results.json',
+        help='Pfad für die Ausgabe der Validierungsergebnisse.'
+    )
+
+    args = parser.parse_args()
+
+    marker_data = _load_marker_data(args.input)
+    results = validate_markers(marker_data)
+    _write_results(args.output, results)
+
+    print(f"Validierung abgeschlossen. Ergebnisse gespeichert unter: {args.output}")
+
+
+if __name__ == '__main__':
+    main()