Add audio_signal_analysis/src/audio_signal_analysis/cli.py

audio_signal_analysis/src/audio_signal_analysis/cli.py

@@ -0,0 +1,64 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+from audio_signal_analysis.core import analyze_audio_signal
+
+
+def _load_audio_data(audio_path: Path) -> dict[str, Any]:
+    if not audio_path.exists() or not audio_path.is_file():
+        raise FileNotFoundError(f"Audio file not found: {audio_path}")
+    with open(audio_path, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    if not isinstance(data, dict) or "timestamp" not in data or "audio_level" not in data:
+        raise ValueError("Invalid audio data format. Expecting dict with 'timestamp' and 'audio_level'.")
+    return data
+
+
+def _load_temperature_data(temp_path: Path) -> pd.DataFrame:
+    if not temp_path.exists() or not temp_path.is_file():
+        raise FileNotFoundError(f"Temperature file not found: {temp_path}")
+    df = pd.read_csv(temp_path)
+    expected_cols = {"timestamp", "temperature_c"}
+    if not expected_cols.issubset(df.columns):
+        raise ValueError(f"Temperature CSV must contain columns {expected_cols}.")
+    return df
+
+
+def _write_output(output_path: Path, result: float) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump({"correlation_coefficient": result}, f, indent=2)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Analyse von Audiodaten und Temperaturkorrelation.")
+    parser.add_argument("--audio", required=True, help="Pfad zur JSON-Datei mit Audiodaten.")
+    parser.add_argument("--temp", required=True, help="Pfad zur CSV-Datei mit Temperaturdaten.")
+    parser.add_argument("--out", required=False, default="output/correlation_result.json", help="Pfad zur Ausgabedatei.")
+
+    args = parser.parse_args()
+
+    try:
+        audio_path = Path(args.audio)
+        temp_path = Path(args.temp)
+        output_path = Path(args.out)
+
+        audio_data = _load_audio_data(audio_path)
+        _ = _load_temperature_data(temp_path)  # loaded for validation / presence
+
+        result = analyze_audio_signal(audio_data)
+
+        _write_output(output_path, result)
+        print(f"Analysis complete. Correlation written to: {output_path}")
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()