264 lines
8.6 KiB
Python
Executable File
264 lines
8.6 KiB
Python
Executable File
import os # 文件系统操作
|
||
import base64 # Base64编码图片
|
||
import paho.mqtt.client as mqtt # MQTT通信
|
||
import pymysql # 数据库操作
|
||
import time # 时间处理
|
||
from datetime import datetime, timedelta # 时间处理
|
||
import serial # 串口通信
|
||
import re # 正则匹配GPS串口数据
|
||
import threading # 多线程上传图片
|
||
|
||
# ======== 配置区 ========
|
||
IMAGE_FOLDER = "/mnt/save/warning" # 图片目录
|
||
MQTT_BROKER = "116.147.36.110"
|
||
MQTT_PORT = 1883
|
||
MQTT_TOPIC = "images/topic"
|
||
|
||
DB_CONFIG = {
|
||
"host": "116.147.36.110",
|
||
"port": 13306,
|
||
"user": "root",
|
||
"password": "Wxit11335577",
|
||
"database": "mqtt"
|
||
}
|
||
|
||
SERIAL_PORT = "/dev/ttyUSB3"
|
||
SERIAL_BAUDRATE = 115200
|
||
|
||
MAX_UPLOADED_SIZE = 100000 # 最大上传记录数
|
||
NOW_SIZE = 0 # 当前记录数
|
||
|
||
LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "upload.log")
|
||
|
||
# ======== 全局变量与线程锁 ========
|
||
last_lon, last_lat = None, None # 上次读取到的GPS
|
||
ser = None # 串口对象
|
||
|
||
serial_lock = threading.Lock()
|
||
uploaded_lock = threading.Lock()
|
||
uploaded = {} # 保存已上传记录的文件及时间
|
||
|
||
# ======== 工具函数 ========
|
||
def write_log(message):
|
||
"""写入日志文件"""
|
||
with open(LOG_FILE, "a") as logf:
|
||
logf.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {message}\n")
|
||
|
||
# def extract_gps_data(serial_line):
|
||
# """
|
||
# 从串口数据中提取经纬度（只处理$GNGLL格式）
|
||
# 示例数据: $GNGLL,3114.72543,N,12130.62735,E,...
|
||
# 只做方向转换，不转换度分格式
|
||
# 返回 (lon, lat)，都为float，带正负号
|
||
# """
|
||
# match = re.search(r'\$GNGLL,(\d+\.\d+),(N|S),(\d+\.\d+),(E|W)', serial_line)
|
||
# if match:
|
||
# lat = float(match.group(1))
|
||
# lat_dir = match.group(2)
|
||
# lon = float(match.group(3))
|
||
# lon_dir = match.group(4)
|
||
|
||
# if lat_dir == 'S':
|
||
# lat = -lat
|
||
# if lon_dir == 'W':
|
||
# lon = -lon
|
||
# return lon, lat
|
||
# return None, None
|
||
|
||
|
||
def extract_gps_data(serial_line):
|
||
"""
|
||
从串口数据中提取经纬度，支持$GNGLL、$GNRMC、$GNGGA格式
|
||
只做方向正负号转换，不转换度分格式
|
||
返回 (lon, lat)，float，带正负号
|
||
"""
|
||
# 先匹配$GNGLL（你之前的）
|
||
match = re.search(r'\$GNGLL,(\d+\.\d+),(N|S),(\d+\.\d+),(E|W)', serial_line)
|
||
if match:
|
||
lat = float(match.group(1))
|
||
lat_dir = match.group(2)
|
||
lon = float(match.group(3))
|
||
lon_dir = match.group(4)
|
||
|
||
if lat_dir == 'S':
|
||
lat = -lat
|
||
if lon_dir == 'W':
|
||
lon = -lon
|
||
|
||
return lon, lat
|
||
|
||
# 尝试匹配$GNRMC
|
||
match = re.search(r'\$GNRMC,[^,]*,[AV],(\d+\.\d+),(N|S),(\d+\.\d+),(E|W)', serial_line)
|
||
if match:
|
||
lat = float(match.group(1))
|
||
lat_dir = match.group(2)
|
||
lon = float(match.group(3))
|
||
lon_dir = match.group(4)
|
||
|
||
if lat_dir == 'S':
|
||
lat = -lat
|
||
if lon_dir == 'W':
|
||
lon = -lon
|
||
|
||
return lon, lat
|
||
|
||
# 尝试匹配$GNGGA
|
||
match = re.search(r'\$GNGGA,[^,]*,(\d+\.\d+),(N|S),(\d+\.\d+),(E|W)', serial_line)
|
||
if match:
|
||
lat = float(match.group(1))
|
||
lat_dir = match.group(2)
|
||
lon = float(match.group(3))
|
||
lon_dir = match.group(4)
|
||
|
||
if lat_dir == 'S':
|
||
lat = -lat
|
||
if lon_dir == 'W':
|
||
lon = -lon
|
||
|
||
return lon, lat
|
||
|
||
return None, None
|
||
|
||
def read_gps_data():
|
||
global last_lon, last_lat, ser
|
||
with serial_lock:
|
||
try:
|
||
if ser is None or not ser.is_open:
|
||
ser = serial.Serial(SERIAL_PORT, SERIAL_BAUDRATE, timeout=1)
|
||
|
||
line = ser.readline().decode('utf-8').strip()
|
||
if line:
|
||
# 只对特定句子类型尝试解析并写日志
|
||
if line.startswith(('$GNGLL', '$GNRMC', '$GNGGA')):
|
||
lon, lat = extract_gps_data(line)
|
||
if lon is not None and lat is not None:
|
||
last_lon, last_lat = lon, lat
|
||
return lon, lat
|
||
else:
|
||
write_log(f"无效的GPS数据: {line}")
|
||
else:
|
||
# 忽略其他NMEA语句
|
||
pass
|
||
except Exception as e:
|
||
write_log(f"串口读取失败: {e}")
|
||
return 1201, 3129
|
||
|
||
|
||
# def read_gps_data():
|
||
# """
|
||
# 从串口读取一行数据并尝试解析GPS经纬度。
|
||
# 不成功时返回 (None, None)
|
||
# """
|
||
# global last_lon, last_lat, ser
|
||
# with serial_lock:
|
||
# try:
|
||
# if ser is None or not ser.is_open:
|
||
# ser = serial.Serial(SERIAL_PORT, SERIAL_BAUDRATE, timeout=1)
|
||
|
||
# line = ser.readline().decode('utf-8').strip()
|
||
# if line:
|
||
# lon, lat = extract_gps_data(line)
|
||
# if lon is not None and lat is not None:
|
||
# last_lon, last_lat = lon, lat # 更新最新值
|
||
# return lon, lat
|
||
# else:
|
||
# write_log(f"无效的GPS数据: {line}")
|
||
# except Exception as e:
|
||
# write_log(f"串口读取失败: {e}")
|
||
# return 1201, 3129
|
||
|
||
def save_to_db(topic, payload, lon, lat):
|
||
"""
|
||
将图片和经纬度信息写入数据库，可接受空值
|
||
"""
|
||
try:
|
||
db = pymysql.connect(**DB_CONFIG)
|
||
cursor = db.cursor()
|
||
sql = "INSERT INTO mqtt_messages (topic, payload, lon, lat) VALUES (%s, %s, %s, %s)"
|
||
cursor.execute(sql, (topic, payload, lon, lat))
|
||
db.commit()
|
||
write_log("数据库插入成功")
|
||
except Exception as e:
|
||
write_log(f"数据库插入失败: {e}")
|
||
finally:
|
||
db.close()
|
||
|
||
def publish_image(image_path):
|
||
"""
|
||
发布图片到MQTT、写入数据库，并附带经纬度信息（如无法获取可为空）
|
||
"""
|
||
with open(image_path, "rb") as f:
|
||
img_bytes = f.read()
|
||
img_b64 = base64.b64encode(img_bytes).decode("utf-8")
|
||
|
||
# MQTT发布
|
||
client = mqtt.Client()
|
||
client.connect(MQTT_BROKER, MQTT_PORT)
|
||
client.publish(MQTT_TOPIC, img_b64)
|
||
client.disconnect()
|
||
|
||
# 获取GPS数据（尝试获取最新，失败用历史，再失败就用None）
|
||
lon, lat = read_gps_data()
|
||
if lon is None or lat is None:
|
||
write_log("GPS读取失败，尝试使用上一次位置")
|
||
lon, lat = last_lon, last_lat
|
||
|
||
if lon is not None and lat is not None:
|
||
write_log(f"使用经纬度上传: lon={lon}, lat={lat}")
|
||
else:
|
||
write_log("经纬度为空，上传空GPS信息")
|
||
|
||
save_to_db(MQTT_TOPIC, img_b64, lon, lat)
|
||
|
||
def upload_worker(img_path):
|
||
"""上传线程任务，处理一张图片上传"""
|
||
try:
|
||
publish_image(img_path)
|
||
write_log("上传线程结束: " + img_path + ":\t成功!")
|
||
except Exception as e:
|
||
write_log(f"上传图片 {img_path} 时出错: {e}")
|
||
|
||
def clean_uploaded():
|
||
"""
|
||
清理7天前的已上传图片记录，避免内存泄漏
|
||
"""
|
||
with uploaded_lock:
|
||
expiration_time = datetime.now() - timedelta(days=7)
|
||
keys_to_remove = [key for key, value in uploaded.items() if value < expiration_time]
|
||
for key in keys_to_remove:
|
||
del uploaded[key]
|
||
if keys_to_remove:
|
||
write_log(f"清理已上传图片记录，移除 {len(keys_to_remove)} 条记录")
|
||
|
||
# ======== 主循环程序入口 ========
|
||
if __name__ == "__main__":
|
||
last_clean_time = time.time()
|
||
try:
|
||
while True:
|
||
now = time.time()
|
||
for filename in os.listdir(IMAGE_FOLDER):
|
||
if filename.lower().endswith((".jpg", ".jpeg", ".png", ".bmp", ".gif")):
|
||
img_path = os.path.join(IMAGE_FOLDER, filename)
|
||
mtime = os.path.getmtime(img_path)
|
||
# 检查文件是否是最近10秒内新创建，且未上传
|
||
if now - mtime <= 10:
|
||
with uploaded_lock:
|
||
if img_path in uploaded:
|
||
continue
|
||
write_log("开始上传图片: " + img_path)
|
||
t = threading.Thread(target=upload_worker, args=(img_path,))
|
||
t.start()
|
||
with uploaded_lock:
|
||
uploaded[img_path] = datetime.now()
|
||
# 每小时清理一次上传记录
|
||
if time.time() - last_clean_time > 3600:
|
||
clean_uploaded()
|
||
last_clean_time = time.time()
|
||
|
||
time.sleep(2) # 每2秒轮询一次
|
||
except Exception as e:
|
||
write_log(f"主程序异常: {e}")
|
||
finally:
|
||
if ser is not None and ser.is_open:
|
||
ser.close()
|