Python集成
本页提供使用FGA Logger的实用Python示例——读取实时串行数据、保存到CSV以及加载数据进行分析。
依赖项：
安装命令：
pip install pyserial pandas
读取实时串行数据
此示例通过USB连接到FGA Logger并打印每个传入的CSV行：
import serial
# 更改为您的端口：
# Windows: 'COM3', 'COM4', 等
# Linux: '/dev/ttyUSB0'
# macOS: '/dev/tty.usbserial-XXXX'
PORT = 'COM3'
BAUD = 115200 # 调整以匹配您的FGA Logger串行设置
def read_logger(port, baud):
with serial.Serial(port, baud, timeout=2) as ser:
print(f"已连接到 {port}，波特率 {baud}")
print("等待数据...\n")
while True:
line = ser.readline().decode('utf-8', errors='replace').strip()
if line:
print(line)
if __name__ == '__main__':
read_logger(PORT, BAUD)
将串行数据保存为CSV
此示例从FGA Logger读取并将所有传入行保存到本地CSV文件。第一个接收到的标题行写入一次；所有后续数据行跟随其后。
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"
# FGA Logger的预期CSV标题
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"记录到 {outfile}")
try:
while True:
line = ser.readline().decode('utf-8', errors='replace').strip()
if not line:
continue
fields = line.split(',')
# 写入标题一次（来自设备或我们的预期标题）
if not header_written:
if fields[0] == 'Timestamp_ms':
writer.writerow(fields) # 使用设备标题
else:
writer.writerow(EXPECTED_HEADER) # 使用预期标题
writer.writerow(fields) # 将此行写为数据
header_written = True
f.flush()
continue
# 写入数据行
writer.writerow(fields)
f.flush()
print(f" {line[:80]}") # 打印前80个字符进行监控
except KeyboardInterrupt:
print(f"\n记录停止。文件已保存：{outfile}")
if __name__ == '__main__':
save_to_csv(PORT, BAUD, OUTFILE)
使用pandas加载CSV文件
获得CSV文件后（来自SD卡或通过串行保存），使用pandas加载进行分析：
import pandas as pd
CSV_FILE = 'fga_log_20240315_143022.csv'
# 加载CSV
df = pd.read_csv(CSV_FILE)
print("形状：", df.shape)
print("\n前几行：")
print(df.head())
print("\n列类型：")
print(df.dtypes)
print("\n基本统计：")
print(df[['B1x_nT', 'B1y_nT', 'B1z_nT', 'B1v_nT']].describe())
计算梯度（梯度仪模式）
在具有两个传感器组件的梯度仪配置中，按轴计算梯度：
import pandas as pd
df = pd.read_csv('fga_log.csv')
# 按轴计算梯度（传感器1减传感器2）
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']
# 总梯度幅度
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))
按GPS质量过滤
分析前过滤掉没有GPS定位或精度差的行：
import pandas as pd
df = pd.read_csv('fga_log.csv')
# 仅保留3D GPS定位且HDOP低于2.0的行
df_clean = df[
(df['Fix'] == 3) &
(df['HDOP_m'] < 2.0) &
(df['SIV'] >= 4)
].copy()
print(f"总行数： {len(df)}")
print(f"清洁行数：{len(df_clean)}")
print(f"已删除： {len(df) - len(df_clean)}")
绘制总场
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('时间 (s)')
plt.ylabel('总场B1 (nT)')
plt.title('FGA Logger — 总磁场')
plt.tight_layout()
plt.savefig('field_plot.png', dpi=150)
plt.show()
将GPS轨迹导出为GeoJSON
导出GPS轨迹以在QGIS或其他GIS工具中使用：
import pandas as pd
import json
df = pd.read_csv('fga_log.csv')
# 仅保留有效GPS定位的行
df_gps = df[df['Fix'] >= 2].dropna(subset=['Lat_deg', 'Lon_deg'])
# 构建GeoJSON FeatureCollection
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)} 个点导出到 track.geojson")