Python-Integration

Diese Seite bietet praktische Python-Beispiele für die Arbeit mit dem FGA Logger — Live-Seriendaten lesen, in CSV speichern und Daten zur Analyse laden.

Abhängigkeiten:

• pyserial — serielle Portkommunikation
• pandas — Datenanalyse (optional, für Analysebeispiele)

Installieren mit:

pip install pyserial pandas

---

Live-Seriendaten lesen

Dieses Beispiel verbindet sich mit dem FGA Logger über USB und gibt jede eingehende CSV-Zeile aus:

import serial

# Auf Ihren Port ändern:
#   Windows:   'COM3', 'COM4', usw.
#   Linux:     '/dev/ttyUSB0'
#   macOS:     '/dev/tty.usbserial-XXXX'
PORT = 'COM3'
BAUD = 115200  # Anpassen an Ihre FGA Logger Serieneinstellungen

def read_logger(port, baud):
    with serial.Serial(port, baud, timeout=2) as ser:
        print(f"Verbunden mit {port} bei {baud} Baud")
        print("Warte auf Daten...\n")

        while True:
            line = ser.readline().decode('utf-8', errors='replace').strip()
            if line:
                print(line)

if __name__ == '__main__':
    read_logger(PORT, BAUD)

---

Seriendaten in CSV speichern

Dieses Beispiel liest vom FGA Logger und speichert alle eingehenden Zeilen in einer lokalen CSV-Datei. Die erste empfangene Kopfzeile wird einmal geschrieben; alle nachfolgenden Datenzeilen folgen.

import serial
import csv
from datetime import datetime

PORT    = 'COM3'
BAUD    = 115200
OUTFILE = f"fga_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"

# Erwarteter CSV-Header vom FGA Logger
EXPECTED_HEADER = [
    'Timestamp_ms', 'B1x_nT', 'B1y_nT', 'B1z_nT', 'B1v_nT',
    'B2x_nT', 'B2y_nT', 'B2z_nT', 'B2v_nT',
    'Lat_deg', 'Lon_deg', 'Alt_m', 'SIV', 'Fix', 'HDOP_m'
]

def save_to_csv(port, baud, outfile):
    header_written = False

    with serial.Serial(port, baud, timeout=5) as ser, \
         open(outfile, 'w', newline='') as f:

        writer = csv.writer(f)
        print(f"Protokollierung in {outfile}")

        try:
            while True:
                line = ser.readline().decode('utf-8', errors='replace').strip()
                if not line:
                    continue

                fields = line.split(',')

                # Header einmal schreiben (entweder vom Gerät oder unser erwarteter Header)
                if not header_written:
                    if fields[0] == 'Timestamp_ms':
                        writer.writerow(fields)   # Geräte-Header verwenden
                    else:
                        writer.writerow(EXPECTED_HEADER)  # Erwarteten Header verwenden
                        writer.writerow(fields)   # Diese Zeile als Daten schreiben
                    header_written = True
                    f.flush()
                    continue

                # Datenzeilen schreiben
                writer.writerow(fields)
                f.flush()
                print(f"  {line[:80]}")  # Erste 80 Zeichen zur Überwachung ausgeben

        except KeyboardInterrupt:
            print(f"\nProtokollierung gestoppt. Datei gespeichert: {outfile}")

if __name__ == '__main__':
    save_to_csv(PORT, BAUD, OUTFILE)

---

CSV-Datei mit pandas laden

Sobald Sie eine CSV-Datei haben — entweder von der SD-Karte oder über Seriell gespeichert — laden Sie sie mit pandas zur Analyse:

import pandas as pd

CSV_FILE = 'fga_log_20240315_143022.csv'

# CSV laden
df = pd.read_csv(CSV_FILE)

print("Form:", df.shape)
print("\nErste Zeilen:")
print(df.head())

print("\nSpaltentypen:")
print(df.dtypes)

print("\nGrundlegende Statistiken:")
print(df[['B1x_nT', 'B1y_nT', 'B1z_nT', 'B1v_nT']].describe())

---

Gradient berechnen (Gradiometer-Modus)

In Gradiometer-Konfigurationen mit zwei Sensorbaugruppen den Gradienten pro Achse berechnen:

import pandas as pd

df = pd.read_csv('fga_log.csv')

# Gradient berechnen (Sensor 1 minus Sensor 2) pro Achse
df['Grad_x_nT'] = df['B1x_nT'] - df['B2x_nT']
df['Grad_y_nT'] = df['B1y_nT'] - df['B2y_nT']
df['Grad_z_nT'] = df['B1z_nT'] - df['B2z_nT']

# Gesamter Gradientenbetrag
df['Grad_v_nT'] = (
    df['Grad_x_nT']**2 +
    df['Grad_y_nT']**2 +
    df['Grad_z_nT']**2
) ** 0.5

print(df[['Timestamp_ms', 'Grad_x_nT', 'Grad_y_nT', 'Grad_z_nT', 'Grad_v_nT']].head(10))

---

Nach GPS-Qualität filtern

Zeilen ohne GPS-Fix oder mit schlechter Genauigkeit vor der Analyse herausfiltern:

import pandas as pd

df = pd.read_csv('fga_log.csv')

# Nur Zeilen mit 3D-GPS-Fix und HDOP unter 2.0 behalten
df_clean = df[
    (df['Fix'] == 3) &
    (df['HDOP_m'] < 2.0) &
    (df['SIV'] >= 4)
].copy()

print(f"Gesamtzeilen:    {len(df)}")
print(f"Bereinigte Zeilen: {len(df_clean)}")
print(f"Entfernt:        {len(df) - len(df_clean)}")

---

Gesamtfeld darstellen

import pandas as pd
import matplotlib.pyplot as plt

df = pd.read_csv('fga_log.csv')

plt.figure(figsize=(12, 4))
plt.plot(df['Timestamp_ms'] / 1000, df['B1v_nT'], linewidth=0.8)
plt.xlabel('Zeit (s)')
plt.ylabel('Gesamtfeld B1 (nT)')
plt.title('FGA Logger — Gesamtes Magnetfeld')
plt.tight_layout()
plt.savefig('field_plot.png', dpi=150)
plt.show()

---

GPS-Track als GeoJSON exportieren

GPS-Track für die Verwendung in QGIS oder anderen GIS-Tools exportieren:

import pandas as pd
import json

df = pd.read_csv('fga_log.csv')

# Nur Zeilen mit gültigem GPS-Fix behalten
df_gps = df[df['Fix'] >= 2].dropna(subset=['Lat_deg', 'Lon_deg'])

# GeoJSON FeatureCollection erstellen
features = []
for _, row in df_gps.iterrows():
    feature = {
        "type": "Feature",
        "geometry": {
            "type": "Point",
            "coordinates": [row['Lon_deg'], row['Lat_deg'], row['Alt_m']]
        },
        "properties": {
            "timestamp_ms": row['Timestamp_ms'],
            "B1v_nT": row['B1v_nT'],
            "B2v_nT": row['B2v_nT'],
        }
    }
    features.append(feature)

geojson = {"type": "FeatureCollection", "features": features}

with open('track.geojson', 'w') as f:
    json.dump(geojson, f, indent=2)

print(f"{len(features)} Punkte nach track.geojson exportiert")