Add artifact.1/src/artifact_1/core.py

artifact.1/src/artifact_1/core.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List
+
+import pandas as pd
+import matplotlib.pyplot as plt
+from matplotlib.figure import Figure
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SignalData:
+    """Repräsentiert eine Zeile der FM-Signaldaten."""
+
+    timestamp: str
+    frequency_MHz: float
+    signal_dB: float
+    noise_dB: float
+    lat: float
+    lon: float
+
+    @staticmethod
+    def validate(row: pd.Series) -> bool:
+        required_columns = [
+            'timestamp', 'frequency_MHz', 'signal_dB', 'noise_dB', 'lat', 'lon'
+        ]
+        for col in required_columns:
+            if col not in row or pd.isna(row[col]):
+                logger.warning("Fehlender oder ungültiger Wert in Spalte '%s'", col)
+                return False
+        return True
+
+
+def generate_signal_plot(csv_file: str) -> Figure:
+    """Liest ein CSV mit FM-Signaldaten und erzeugt einen Frequenz-vs-Signal-Plot.
+
+    Args:
+        csv_file (str): Pfad zur CSV-Datei mit Spalten 'frequency_MHz' und 'signal_dB'.
+
+    Returns:
+        matplotlib.figure.Figure: Eine Figure mit dem visualisierten Frequenzspektrum.
+    """
+    csv_path = Path(csv_file)
+    if not csv_path.exists() or not csv_path.is_file():
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {csv_file}")
+
+    logger.debug("Lese CSV-Datei: %s", csv_path)
+    df = pd.read_csv(csv_path)
+
+    required_cols = {'frequency_MHz', 'signal_dB'}
+    if not required_cols.issubset(df.columns):
+        raise ValueError(
+            f"CSV-Datei muss Spalten {required_cols} enthalten, gefunden: {set(df.columns)}"
+        )
+
+    # Input validation pro Row, falls Spalten vorhanden sind
+    valid_mask = df.apply(SignalData.validate, axis=1)
+    df_valid = df[valid_mask]
+
+    if df_valid.empty:
+        raise ValueError("Keine gültigen Datenzeilen gefunden nach Validierung.")
+
+    # Sortieren nach Frequenz, um sauberen Plot zu gewährleisten
+    df_sorted = df_valid.sort_values(by='frequency_MHz')
+
+    logger.debug("Erstelle Plot (%d Werte)...", len(df_sorted))
+    fig, ax = plt.subplots(figsize=(10, 5))
+    ax.plot(df_sorted['frequency_MHz'], df_sorted['signal_dB'], label='Signal (dB)', color='tab:blue')
+    ax.set_xlabel('Frequency [MHz]')
+    ax.set_ylabel('Signal Strength [dB]')
+    ax.set_title('FM Spectrum Analysis')
+    ax.grid(True, linestyle='--', alpha=0.7)
+    ax.legend()
+
+    fig.tight_layout()
+    logger.info("Plot erfolgreich erzeugt für Datei: %s", csv_file)
+
+    return fig