Add data_analysis/src/data_analysis/cli.py

data_analysis/src/data_analysis/cli.py

@@ -0,0 +1,88 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any, List
+import numpy as np
+
+# Lokale Imports
+def _import_core():
+    try:
+        from data_analysis.core import analyze_patterns, PatternReport  # type: ignore
+        return analyze_patterns, PatternReport
+    except ImportError as e:
+        raise RuntimeError(f"Fehler beim Import von core-Modul: {e}")
+
+
+def _parse_args(argv: List[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Analysiert Audiodaten und extrahiert Klangmuster."
+    )
+    parser.add_argument(
+        "--input",
+        required=True,
+        help="Pfad zur Audiodatei, die analysiert werden soll.",
+    )
+    parser.add_argument(
+        "--output",
+        required=False,
+        default="output/pattern_report.json",
+        help="Pfad zur JSON-Datei, in die der Musterreport geschrieben wird.",
+    )
+    return parser.parse_args(argv)
+
+
+def _load_audio_data(audio_path: Path) -> np.ndarray:
+    import librosa
+
+    if not audio_path.exists() or not audio_path.is_file():
+        raise FileNotFoundError(f"Audiodatei nicht gefunden: {audio_path}")
+
+    # Librosa lädt Audiodateien als float32-Array
+    data, _ = librosa.load(audio_path, sr=None, mono=True)
+    return data
+
+
+def _serialize_reports(reports: List[Any]) -> list[dict[str, Any]]:
+    serialized = []
+    for r in reports:
+        if hasattr(r, "__dict__"):
+            serialized.append({
+                "pattern_name": getattr(r, "pattern_name", None),
+                "frequency": getattr(r, "frequency", None),
+                "correlation": getattr(r, "correlation", None),
+                "comments": getattr(r, "comments", None),
+            })
+        elif isinstance(r, dict):
+            serialized.append(r)
+        else:
+            raise TypeError(f"Unbekannter Reporttyp: {type(r)}")
+    return serialized
+
+
+def main(argv: List[str] | None = None) -> None:
+    args = _parse_args(argv)
+    analyze_patterns, PatternReport = _import_core()
+
+    audio_path = Path(args.input)
+    output_path = Path(args.output)
+
+    audio_data = _load_audio_data(audio_path)
+
+    try:
+        reports = analyze_patterns(audio_data)
+    except Exception as e:
+        print(f"Fehler bei der Analyse: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    serialized = _serialize_reports(reports)
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(serialized, f, ensure_ascii=False, indent=2)
+
+    print(f"Analyse abgeschlossen. Ergebnisse in {output_path} gespeichert.")
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])