diff --git a/data_analysis/src/data_analysis/cli.py b/data_analysis/src/data_analysis/cli.py new file mode 100644 index 0000000..a45bea4 --- /dev/null +++ b/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:])