树莓派无网络时存储GPS数据，联网后同步至InfluxDB方案咨询

针对你的需求，我们可以通过本地SQLite数据库存储离线数据+捕获网络异常实现离线写入+定时检测网络并同步本地数据的方式来实现。以下是修改后的完整代码：

from datetime import datetime
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
from influxdb_client.exceptions import InfluxDBError
import serial
import sqlite3
import time
import socket

# InfluxDB配置
token = "<mytoken>"
org = "<myorg>"
bucket = "<mybucket>"
influx_url = "<influxurl>"

# 本地SQLite配置
DB_PATH = "/home/pi/gps_offline.db"

def init_local_db():
    """初始化本地SQLite数据库，创建GPS数据存储表"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    # 创建表，存储纬度、经度、UTC时间戳（字符串格式，避免时间处理问题）
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS gps_data (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            latitude REAL NOT NULL,
            longitude REAL NOT NULL,
            timestamp TEXT NOT NULL
        )
    ''')
    conn.commit()
    conn.close()

def is_network_available():
    """检测网络是否可用，通过尝试连接InfluxDB服务器端口实现"""
    try:
        # 解析InfluxDB URL的主机和端口
        url_parts = influx_url.split("//")[-1].split(":")
        host = url_parts[0]
        port = int(url_parts[1]) if len(url_parts) > 1 else 8086
        # 尝试建立TCP连接
        socket.create_connection((host, port), timeout=5)
        return True
    except (socket.error, ValueError):
        return False

def save_to_local(lat, longi, timestamp):
    """将GPS数据保存到本地SQLite"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        INSERT INTO gps_data (latitude, longitude, timestamp)
        VALUES (?, ?, ?)
    ''', (lat, longi, timestamp))
    conn.commit()
    conn.close()

def sync_local_data(write_api):
    """将本地存储的所有数据同步到InfluxDB，成功后删除本地数据"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    # 查询所有未同步的数据
    cursor.execute('SELECT latitude, longitude, timestamp FROM gps_data ORDER BY id')
    rows = cursor.fetchall()
    if not rows:
        conn.close()
        return
    
    points = []
    for row in rows:
        lat, longi, ts_str = row
        # 将字符串时间戳转换为datetime对象
        ts = datetime.fromisoformat(ts_str)
        point = Point("latest GPS") \
            .field("latitude", lat) \
            .field("longitude", longi) \
            .time(ts, WritePrecision.NS)
        points.append(point)
    
    try:
        # 批量写入InfluxDB
        write_api.write(bucket, org, points)
        # 同步成功后删除本地数据
        cursor.execute('DELETE FROM gps_data')
        conn.commit()
        print(f"成功同步{len(rows)}条本地GPS数据到InfluxDB")
    except InfluxDBError as e:
        print(f"同步失败: {str(e)}")
    finally:
        conn.close()

def convert_to_degrees(raw_value):
    decimal_value = raw_value / 100.00
    degrees = int(decimal_value)
    mm_mmmm = (decimal_value - int(decimal_value)) / 0.6
    position = degrees + mm_mmmm
    return float("%.4f" % position)

def main():
    # 初始化本地数据库
    init_local_db()
    
    # 初始化InfluxDB客户端（不使用with语句，避免网络断开时客户端被关闭）
    client = None
    write_api = None
    try:
        client = InfluxDBClient(url=influx_url, token=token, org=org)
        write_api = client.write_api(write_options=SYNCHRONOUS)
    except InfluxDBError:
        print("初始连接InfluxDB失败，将使用本地存储")
    
    # 初始化串口
    ser = serial.Serial("/dev/ttyS0")
    gpgga_info = "$GPGGA,"
    
    # 同步检查间隔（秒）
    sync_check_interval = 60
    last_sync_check = time.time()
    
    while True:
        received_data = str(ser.readline())
        GPGGA_data_available = received_data.find(gpgga_info)
        if GPGGA_data_available > 0:
            GPGGA_buffer = received_data.split("$GPGGA,", 1)[1]
            NMEA_buff = GPGGA_buffer.split(',')
            
            # 确保数据字段完整，避免索引越界
            if len(NMEA_buff) < 4:
                continue
                
            extract_latitude = NMEA_buff[1]
            extract_longitude = NMEA_buff[3]
            
            try:
                lat = float(extract_latitude)
                lat = convert_to_degrees(lat)
                longi = float(extract_longitude)
                longi = convert_to_degrees(longi)
                current_ts = datetime.utcnow().isoformat()
                
                # 尝试写入InfluxDB
                if write_api and is_network_available():
                    point = Point("latest GPS") \
                        .field("latitude", lat) \
                        .field("longitude", longi) \
                        .time(datetime.utcnow(), WritePrecision.NS)
                    write_api.write(bucket, org, point)
                    print("数据已写入InfluxDB")
                else:
                    # 网络不可用，保存到本地
                    save_to_local(lat, longi, current_ts)
                    print("网络不可用，数据已保存到本地")
                    
            except ValueError as e:
                print(f"解析GPS数据失败: {str(e)}")
                continue
        
        # 定时检查网络并同步本地数据
        if time.time() - last_sync_check >= sync_check_interval:
            if is_network_available():
                print("网络已恢复，开始同步本地数据...")
                if not write_api:
                    # 重新初始化InfluxDB客户端
                    try:
                        client = InfluxDBClient(url=influx_url, token=token, org=org)
                        write_api = client.write_api(write_options=SYNCHRONOUS)
                    except InfluxDBError as e:
                        print(f"重新连接InfluxDB失败: {str(e)}")
                        last_sync_check = time.time()
                        continue
                sync_local_data(write_api)
            last_sync_check = time.time()
            
        # 短暂休眠，降低CPU占用
        time.sleep(0.1)

if __name__ == "__main__":
    main()

关键功能说明 #

• 本地存储：使用Python自带的SQLite数据库，无需额外安装依赖，存储GPS数据的纬度、经度和UTC时间戳。
• 网络检测：通过尝试连接InfluxDB服务器的端口来判断网络是否可用，避免误判。
• 异常处理：捕获InfluxDB写入异常，自动切换到本地存储；网络恢复时自动重新初始化客户端。
• 定时同步：每隔60秒检查一次网络状态，恢复后批量同步本地所有数据到InfluxDB，同步成功后清空本地数据。
• 数据完整性：增加GPS字段完整性检查，避免因NMEA数据格式错误导致程序崩溃。

使用注意事项 #

1. 确保树莓派上已安装influxdb_client和pyserial：

   pip install influxdb-client pyserial
   

2. 修改token、org、bucket、influx_url为你的实际配置。
3. SQLite数据库文件默认存储在/home/pi/gps_offline.db，可根据需要修改DB_PATH变量。

内容的提问来源于stack exchange，提问作者Krishnan Manimaran