1

CMakeLists.txt

@@ -1,63 +1,56 @@
 cmake_minimum_required(VERSION 3.10)
 
-# 工程名称（随便改）
-project(rtsp_server)
+project(rtmp_publisher)
 
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 add_compile_definitions(GLIB_DISABLE_DEPRECATION_WARNINGS)
 
-# 设置可执行文件输出目录
-set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/../bin)
+# 输出到工程 bin/
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin)
 
-# 找 GStreamer
+# ---------- GStreamer ----------
 find_package(PkgConfig REQUIRED)
 pkg_check_modules(GSTREAMER REQUIRED gstreamer-1.0)
-pkg_check_modules(GSTREAMER_RTSP REQUIRED gstreamer-rtsp-server-1.0)
 
-# 添加头文件目录
-include_directories(
+# ---------- 源码 ----------
+file(GLOB SRC_FILES src/*.cpp)
+
+add_executable(camera_to_rtmp ${SRC_FILES})
+
+# ---------- 头文件 ----------
+target_include_directories(camera_to_rtmp PRIVATE
     ${CMAKE_SOURCE_DIR}/include
     ${CMAKE_SOURCE_DIR}/third_party/include
     ${CMAKE_SOURCE_DIR}/third_party/include/paho_mqtt
     ${GSTREAMER_INCLUDE_DIRS}
-    ${GSTREAMER_RTSP_INCLUDE_DIRS}
 )
 
-# 源码目录
-file(GLOB SRC_FILES src/*.cpp)
-
-# 生成可执行文件（名字和 project 不一定要一样）
-add_executable(camera_to_rtsp ${SRC_FILES})
-
-# 链接库目录
-link_directories(
-    ${GSTREAMER_LIBRARY_DIRS}
-    ${GSTREAMER_RTSP_LIBRARY_DIRS}
+# ---------- 链接库 ----------
+target_link_directories(camera_to_rtmp PRIVATE
     ${CMAKE_SOURCE_DIR}/third_party/lib
 )
 
-# 链接库
-target_link_libraries(camera_to_rtsp
+target_link_libraries(camera_to_rtmp
     ${GSTREAMER_LIBRARIES}
-    ${GSTREAMER_RTSP_LIBRARIES}
     pthread
     ${CMAKE_SOURCE_DIR}/third_party/lib/libpaho-mqttpp3.a
     ${CMAKE_SOURCE_DIR}/third_party/lib/libpaho-mqtt3a.a
     ${CMAKE_SOURCE_DIR}/third_party/lib/libpaho-mqtt3c.a
 )
 
-set_target_properties(camera_to_rtsp PROPERTIES
+# ---------- RPATH ----------
+set_target_properties(camera_to_rtmp PROPERTIES
     BUILD_RPATH "\$ORIGIN/lib"
     INSTALL_RPATH "\$ORIGIN/lib"
 )
 
-# 运行时把 config.json 复制到 bin/ 目录
+# ---------- 配置文件 ----------
 add_custom_command(
-    TARGET camera_to_rtsp POST_BUILD
-    COMMAND ${CMAKE_COMMAND} -E make_directory $<TARGET_FILE_DIR:camera_to_rtsp>
+    TARGET camera_to_rtmp POST_BUILD
+    COMMAND ${CMAKE_COMMAND} -E make_directory $<TARGET_FILE_DIR:camera_to_rtmp>
     COMMAND ${CMAKE_COMMAND} -E copy_if_different
         ${CMAKE_SOURCE_DIR}/config.json
-        $<TARGET_FILE_DIR:camera_to_rtsp>/config.json.default
+        $<TARGET_FILE_DIR:camera_to_rtmp>/config.json.default
 )

include/mqtt_client_wrapper.hpp

@@ -1,12 +1,13 @@
 // mqtt_client_wrapper.hppa
 #pragma once
 
+#include <atomic>
+#include <memory>
+
 #include "app_config.hpp"
 #include "logger.hpp"
 #include "mqtt_client.hpp"
-#include "rtsp_manager.hpp"
-#include <memory>
-#include <atomic>
+#include "rtmp_manager.hpp"
 
 // 启动 MQTT 客户端线程（内部自动重连、订阅等）
 void mqtt_client_thread_func();

include/rtmp_manager.hpp

@@ -0,0 +1,67 @@
+// rtmp_manager.hpp
+#pragma once
+
+#include <gst/gst.h>
+
+#include <atomic>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+#include "app_config.hpp"
+#include "logger.hpp"
+
+std::string get_ip_address(const std::string& ifname);
+
+class RTMPManager
+{
+   public:
+    struct StreamStatus
+    {
+        bool running{false};
+        std::string last_error;
+    };
+
+    // 初始化与控制接口
+    static void init();
+    static void start_all();
+    static void stop_all();
+    static bool is_streaming(const std::string& cam_name);
+    static std::string get_stream_url(const std::string& cam_name);
+
+    // 新增：单通道状态结构体（用于心跳）
+    struct ChannelInfo
+    {
+        int loc;             // 摄像头位置索引（0~7）
+        std::string url;     // 正常时的推流地址
+        bool running;        // 是否正常运行
+        std::string reason;  // 错误原因（仅在异常时填）
+    };
+
+    // 新增：获取所有通道详细状态
+    static std::vector<ChannelInfo> get_all_channels_status();
+
+   private:
+    struct StreamContext
+    {
+        std::atomic<bool> thread_running{false};
+        std::thread thread;
+
+        StreamStatus status;
+        std::mutex status_mutex;
+
+        // 新增：记录最后收到帧的时间（毫秒）
+        std::atomic<int64_t> last_frame_ms{0};
+    };
+
+    static void stream_loop(Camera cam, StreamContext* ctx);
+    static GstElement* create_pipeline(const Camera& cam);
+    static std::string make_key(const std::string& name);
+
+    static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
+    static std::mutex streams_mutex;
+
+    static constexpr int RETRY_BASE_DELAY_MS = 3000;
+};

include/rtsp_manager.hpp

@@ -1,50 +0,0 @@
-// rtsp_manager.hpp
-#pragma once
-
-#include <gst/gst.h>
-#include <gst/rtsp-server/rtsp-server.h>
-#include "app_config.hpp"
-#include <unordered_map>
-#include <string>
-#include <mutex>
-#include <vector>
-
-class RTSPManager
-{
-public:
-    static void init();
-    static void start(const std::vector<Camera> &cameras);
-    static void stop();
-
-    static void mount_camera(const Camera &cam);
-    static void unmount_camera(const Camera &cam);
-
-    static bool is_streaming(const std::string &cam_name);
-    static bool is_any_streaming();
-
-private:
-    static GMainLoop *loop;
-    static GMainContext *main_context;
-    static GstRTSPServer *server;
-
-    static std::unordered_map<std::string, bool> streaming_status;
-
-    // 工厂创建
-    static GstRTSPMediaFactory *create_media_factory(const Camera &cam);
-
-    // 挂载/卸载
-    static gboolean mount_camera_in_main(gpointer data);
-    static gboolean unmount_camera_in_main(gpointer data);
-
-    // factory 管理
-    static std::unordered_map<std::string, GstRTSPMediaFactory *> mounted_factories;
-    static std::mutex mounted_factories_mutex;
-
-    // pipeline/media 管理
-    static std::unordered_map<std::string, std::vector<GstRTSPMedia *>> media_map;
-    static std::mutex media_map_mutex;
-
-    // gstreamer 信号
-    static void on_media_created(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data);
-    static void on_media_unprepared(GstRTSPMedia *media, gpointer user_data);
-};

include/serial_AT.hpp

@@ -0,0 +1,9 @@
+#pragma once
+
+#include "serial_port.h"
+
+// 初始化 AT 串口（启动线程）
+void init_serial_at(const std::string& device, int baudrate);
+
+// 停止 AT 串口（停止线程，join）
+void stop_serial_at();

include/serial_port.h

@@ -0,0 +1,54 @@
+#pragma once
+
+#include <atomic>
+#include <functional>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "logger.hpp"
+
+class SerialPort
+{
+   public:
+    using ReceiveCallback = std::function<void(const std::vector<uint8_t>&)>;
+    using ReceiveStringCallback = std::function<void(const std::string&)>;
+
+    SerialPort(const std::string& id, const std::string& device, int baudrate, int retry_interval = 5);
+
+    ~SerialPort();
+
+    void start();  // 启动串口（含自动重连）
+    void stop();   // 停止串口
+
+    bool is_open() const;
+    bool send_data(const std::vector<uint8_t>& data);
+    bool send_data(const std::string& data);
+
+    void set_receive_callback(ReceiveCallback cb);
+    void set_receive_callback(ReceiveStringCallback cb);
+
+   private:
+    bool open_port();
+    void close_port();
+    void reader_loop();
+    void reconnect_loop();
+    bool configure_port(int fd);
+
+   private:
+    std::string id_;
+    std::string device_;
+    int baudrate_;
+    int fd_ = -1;
+
+    std::atomic<bool> running_{false};
+    std::atomic<bool> stop_flag_{false};
+
+    std::thread reader_thread_;
+    std::thread reconnect_thread_;
+    std::mutex send_mutex_;
+
+    ReceiveCallback receive_callback_;
+    int retry_interval_;
+};

src/main.cpp

@@ -1,139 +1,83 @@
 // main.cpp
+#include <unistd.h>
+
+#include <atomic>
+#include <chrono>
+#include <csignal>
+#include <thread>
+
 #include "app_config.hpp"
-#include "rtsp_manager.hpp"
 #include "logger.hpp"
 #include "mqtt_client_wrapper.hpp"
-#include <thread>
-#include <atomic>
-#include <csignal>
-#include <unistd.h>
-#include <chrono>
-
-constexpr bool ENABLE_RTSP_THREAD = true;
-constexpr bool ENABLE_MQTT_THREAD = true;
+#include "rtmp_manager.hpp"
+#include "serial_AT.hpp"
 
 std::atomic<bool> g_running(true);
 
-static void minimal_signal_handler(int signum)
+// ---------- 信号处理 ----------
+static void signal_handler(int)
 {
     g_running.store(false, std::memory_order_relaxed);
-    const char msg[] = "[MAIN] Signal received, initiating shutdown...\n";
+    const char msg[] = "[MAIN] Signal received, shutting down...\n";
     write(STDERR_FILENO, msg, sizeof(msg) - 1);
 }
 
 int main()
 {
-    struct sigaction sigAct{};
-    sigAct.sa_handler = minimal_signal_handler;
-    sigemptyset(&sigAct.sa_mask);
-    sigAct.sa_flags = 0;
-    sigaction(SIGINT, &sigAct, nullptr);
-    sigaction(SIGTERM, &sigAct, nullptr);
+    // ---------- 信号 ----------
+    struct sigaction sa{};
+    sa.sa_handler = signal_handler;
+    sigemptyset(&sa.sa_mask);
+    sa.sa_flags = 0;
+    sigaction(SIGINT, &sa, nullptr);
+    sigaction(SIGTERM, &sa, nullptr);
     signal(SIGPIPE, SIG_IGN);
 
+    // ---------- 日志 ----------
     Logger::set_log_to_file(get_executable_dir_file_path("app.log"));
+    LOG_INFO("[MAIN] ===== Video RTMP Publisher Starting =====");
 
+    // ---------- 配置 ----------
     try
     {
         g_app_config = AppConfig::load_from_file(get_executable_dir_file_path("config.json"));
+        LOG_INFO("[MAIN] Config loaded.");
     }
-    catch (const std::exception &e)
+    catch (const std::exception& e)
     {
-        LOG_ERROR(std::string("Failed to load config: ") + e.what());
+        LOG_ERROR(std::string("[MAIN] Failed to load config: ") + e.what());
         return -1;
     }
 
-    std::atomic<bool> rtsp_thread_exited(false);
-    std::atomic<bool> mqtt_thread_exited(false);
+    init_serial_at("/dev/ttyUSB3", 115200);
+    // ---------- GStreamer ----------
+    RTMPManager::init();
 
-    // 初始化 GStreamer
-    RTSPManager::init();
+    // ---------- 启动 RTMP 推流 ----------
+    LOG_INFO("[MAIN] Starting RTMP pipelines...");
+    RTMPManager::start_all();
 
-    std::thread rtsp_thread;
-    std::thread mqtt_thread;
-
-    if (ENABLE_RTSP_THREAD)
-    {
-        rtsp_thread = std::thread([&]()
-                                  {
-            RTSPManager::start(g_app_config.cameras);
-            rtsp_thread_exited.store(true, std::memory_order_relaxed); });
-
-        LOG_INFO("[MAIN] RTSP thread started");
-    }
-    else
-    {
-        LOG_INFO("[MAIN] RTSP thread disabled by toggle");
-    }
-
-    if (ENABLE_MQTT_THREAD)
-    {
-        mqtt_thread = std::thread([&]()
+    // ---------- MQTT ----------
+    std::thread mqtt_thread(
+        []
         {
+            LOG_INFO("[MAIN] MQTT thread started.");
             mqtt_client_thread_func();
-            mqtt_thread_exited.store(true, std::memory_order_relaxed); });
+            LOG_INFO("[MAIN] MQTT thread exited.");
+        });
 
-        LOG_INFO("[MAIN] MQTT thread started");
-    }
-    else
-    {
-        LOG_INFO("[MAIN] MQTT thread disabled by toggle");
-    }
+    // ---------- 主循环 ----------
+    while (g_running.load(std::memory_order_relaxed)) std::this_thread::sleep_for(std::chrono::milliseconds(200));
 
-    while (g_running.load(std::memory_order_relaxed))
-        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    // ---------- 退出 ----------
+    LOG_INFO("[MAIN] Shutdown requested. Stopping services...");
 
-    LOG_INFO("[MAIN] Shutdown requested, stopping services...");
+    stop_serial_at();
 
-    const auto max_wait = std::chrono::seconds(5);
-    const auto poll_interval = std::chrono::milliseconds(100);
+    RTMPManager::stop_all();
 
-    if (ENABLE_RTSP_THREAD)
-    {
-        RTSPManager::stop();
+    if (mqtt_thread.joinable()) mqtt_thread.join();
 
-        if (rtsp_thread.joinable())
-            rtsp_thread.join();
-
-        LOG_INFO("[MAIN] RTSP thread finished and joined.");
-    }
-
-    auto deadline = std::chrono::steady_clock::now() + max_wait;
-
-    if (ENABLE_MQTT_THREAD)
-    {
-        while (!mqtt_thread_exited.load(std::memory_order_relaxed) &&
-               std::chrono::steady_clock::now() < deadline)
-        {
-            std::this_thread::sleep_for(poll_interval);
-        }
-
-        if (mqtt_thread.joinable())
-        {
-            if (mqtt_thread_exited.load(std::memory_order_relaxed))
-            {
-                mqtt_thread.join();
-                LOG_INFO("[MAIN] MQTT thread finished and joined.");
-            }
-            else
-            {
-                LOG_WARN("[MAIN] MQTT thread did not exit within timeout.");
-            }
-        }
-    }
-
-    bool any_failed = false;
-    if (ENABLE_RTSP_THREAD && rtsp_thread.joinable() && !rtsp_thread_exited.load())
-        any_failed = true;
-    if (ENABLE_MQTT_THREAD && mqtt_thread.joinable() && !mqtt_thread_exited.load())
-        any_failed = true;
-
-    if (any_failed)
-    {
-        LOG_ERROR("[MAIN] Threads did not exit in time. Forcing immediate termination.");
-        _exit(1);
-    }
-
-    LOG_INFO("[MAIN] Program exited cleanly.");
+    LOG_INFO("[MAIN] ===== Video RTMP Publisher Exited Cleanly =====");
     return 0;
 }

src/mqtt_client_wrapper.cpp

@@ -56,7 +56,7 @@ static void on_mqtt_message_received(const std::string& topic, const std::string
             auto j = nlohmann::json::parse(message);
             auto seqNo = j["data"].value("seqNo", "");
 
-            LOG_INFO("[MQTT] video_down received, RTSP always running (no action).");
+            LOG_INFO("[MQTT] video_down received, stream is always available via MediaMTX.");
 
             nlohmann::json reply_data;
             reply_data["time"] = Logger::get_current_time_utc8();
@@ -75,7 +75,7 @@ static void on_mqtt_message_received(const std::string& topic, const std::string
         // ------- substream_down 应答 -------
         if (topic == g_app_config.mqtt.topics.substream_down)
         {
-            LOG_INFO("[MQTT] substream_down received (ignored).");
+            LOG_WARN("[MQTT] substream_down received, substream is not supported in MediaMTX mode.");
             return;
         }
 

src/rtmp_manager.cpp

@@ -0,0 +1,459 @@
+// rtmp_manager.cpp
+#include "rtmp_manager.hpp"
+
+#include <arpa/inet.h>
+#include <ifaddrs.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <sys/stat.h>
+
+#include <chrono>
+#include <cstring>
+#include <thread>
+
+// ========== 工具函数 ==========
+static bool device_exists(const std::string& path)
+{
+    struct stat st;
+    return (stat(path.c_str(), &st) == 0);
+}
+
+// 动态获取指定网卡 IPv4 地址
+std::string get_ip_address(const std::string& ifname)
+{
+    struct ifaddrs *ifaddr, *ifa;
+    char ip[INET_ADDRSTRLEN] = {0};
+
+    if (getifaddrs(&ifaddr) == -1)
+    {
+        perror("getifaddrs");
+        return "";
+    }
+
+    for (ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next)
+    {
+        if (ifa->ifa_addr == nullptr) continue;
+
+        if (ifa->ifa_addr->sa_family == AF_INET && ifname == ifa->ifa_name)
+        {
+            void* addr = &((struct sockaddr_in*)ifa->ifa_addr)->sin_addr;
+            if (inet_ntop(AF_INET, addr, ip, sizeof(ip)))
+            {
+                freeifaddrs(ifaddr);
+                return ip;
+            }
+        }
+    }
+
+    freeifaddrs(ifaddr);
+    return "";
+}
+
+// ========== 静态成员 ==========
+std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
+std::mutex RTMPManager::streams_mutex;
+
+// ========== 初始化 ==========
+void RTMPManager::init()
+{
+    gst_init(nullptr, nullptr);
+    LOG_INFO("[RTMP] GStreamer initialized.");
+}
+
+std::string RTMPManager::make_key(const std::string& name) { return name + "_main"; }
+
+// ========== 创建推流管线 ==========
+// GstElement* RTMPManager::create_pipeline(const Camera& cam)
+// {
+//     const int width = cam.width;
+//     const int height = cam.height;
+//     const int fps = cam.fps;
+//     const int bitrate = cam.bitrate;
+
+//     // MediaMTX 中的 stream key
+//     const std::string stream_name = cam.name;
+
+//     // RTMP 推送到 MediaMTX
+//     // mediamtx.yml 中 paths 会自动创建
+//     const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + stream_name;
+
+//     /*
+//      * Pipeline 说明：
+//      * v4l2src -> mpph264enc -> h264parse -> flvmux -> rtmpsink
+//      *
+//      * - 不使用 tee（降低死锁概率）
+//      * - 不引入音频（MediaMTX 不强制要求）
+//      * - 纯视频、纯 RTMP、纯 TCP
+//      */
+//     std::string pipeline_str = "v4l2src name=src device=" + cam.device +
+//                                " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
+//                                ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
+//                                "/1 "
+//                                " ! mpph264enc bps=" +
+//                                std::to_string(bitrate) + " gop=" + std::to_string(fps) +
+//                                " rc-mode=cbr "
+//                                " ! h264parse name=parse "
+//                                " ! flvmux streamable=true "
+//                                " ! rtmpsink location=\"" +
+//                                rtmp_url +
+//                                "\" "
+//                                "   sync=false async=false";
+
+//     GError* error = nullptr;
+//     GstElement* pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
+//     if (error)
+//     {
+//         LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message);
+//         g_error_free(error);
+//         return nullptr;
+//     }
+
+//     return pipeline;
+// }
+
+GstElement* RTMPManager::create_pipeline(const Camera& cam)
+{
+    const int bitrate = cam.bitrate;
+    const int gop = 30;  // 固定 GOP，不强依赖实时 fps
+
+    const std::string stream_name = cam.name;
+    const std::string rtmp_url = "rtmp://127.0.0.1:1935/" + stream_name;
+
+    std::string pipeline_str =
+        // ===== V4L2 Source =====
+        "v4l2src device=" + cam.device +
+        " io-mode=dmabuf do-timestamp=true "
+
+        // ⚠️ 不再锁死 framerate（这是装车最关键的一步）
+        "! video/x-raw,format=NV12,width=1280,height=960 "
+
+        // ===== 启动/抖动缓冲（吃掉脏帧）=====
+        "! queue max-size-buffers=8 max-size-time=0 max-size-bytes=0 "
+        "leaky=downstream "
+
+        // ===== MPP H.264 Encoder =====
+        "! mpph264enc name=enc "
+        "rc-mode=cbr "
+        "bps=" +
+        std::to_string(bitrate) +
+        " "
+        "gop=" +
+        std::to_string(gop) +
+        " "
+        "profile=main "
+        "header-mode=each-idr "
+
+        // ===== H264 Parse =====
+        "! h264parse config-interval=1 "
+        "! video/x-h264,stream-format=avc,alignment=au "
+
+        // ===== RTMP =====
+        "! flvmux streamable=true "
+        "! rtmpsink location=\"" +
+        rtmp_url +
+        "\" "
+        "sync=false async=false";
+
+    GError* error = nullptr;
+    GstElement* pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
+    if (error)
+    {
+        LOG_ERROR(std::string("[RTMP] Pipeline creation failed: ") + error->message);
+        g_error_free(error);
+        return nullptr;
+    }
+
+    return pipeline;
+}
+
+// ========== 主推流循环 ==========
+void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
+{
+    const std::string key = make_key(cam.name);
+
+    constexpr int64_t START_TIMEOUT_MS = 12000;     // 启动阶段无帧
+    constexpr int64_t NO_FRAME_TIMEOUT_MS = 15000;  // 运行阶段突然无帧 ⇒ pipeline 卡死重启
+
+    while (ctx->thread_running)
+    {
+        // 1) 检查设备节点
+        struct stat st{};
+        if (stat(cam.device.c_str(), &st) != 0)
+        {
+            std::lock_guard<std::mutex> lk(ctx->status_mutex);
+            ctx->status.running = false;
+            ctx->status.last_error = "Device not found: " + cam.device;
+            LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
+            std::this_thread::sleep_for(std::chrono::seconds(3));
+            continue;
+        }
+
+        // 2) 创建 pipeline
+        GstElement* pipeline = create_pipeline(cam);
+        if (!pipeline)
+        {
+            std::lock_guard<std::mutex> lk(ctx->status_mutex);
+            ctx->status.running = false;
+            ctx->status.last_error = "Pipeline creation failed";
+            std::this_thread::sleep_for(std::chrono::seconds(3));
+            continue;
+        }
+        gst_element_set_name(pipeline, key.c_str());
+        GstBus* bus = gst_element_get_bus(pipeline);
+
+        // 3) 帧探测：挂在 encoder 的 src pad（真实“视频已生成”）
+        ctx->last_frame_ms.store(0, std::memory_order_relaxed);
+
+        {
+            GstElement* enc = gst_bin_get_by_name(GST_BIN(pipeline), "enc");
+            if (!enc) { LOG_ERROR("[RTMP] Failed to find encoder element"); }
+            else
+            {
+                GstPad* pad = gst_element_get_static_pad(enc, "src");
+                if (pad)
+                {
+                    gst_pad_add_probe(
+                        pad, GST_PAD_PROBE_TYPE_BUFFER,
+                        [](GstPad*, GstPadProbeInfo*, gpointer data) -> GstPadProbeReturn
+                        {
+                            auto* ts = static_cast<std::atomic<int64_t>*>(data);
+                            auto now = std::chrono::steady_clock::now().time_since_epoch();
+                            auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(now).count();
+                            ts->store(ms, std::memory_order_relaxed);
+                            return GST_PAD_PROBE_OK;
+                        },
+                        &(ctx->last_frame_ms), nullptr);
+                    gst_object_unref(pad);
+                }
+                gst_object_unref(enc);
+            }
+        }
+
+        // 4) 启动 pipeline
+        LOG_INFO("[RTMP] Starting stream: " + key);
+        gst_element_set_state(pipeline, GST_STATE_PLAYING);
+
+        GstState state = GST_STATE_NULL, pending = GST_STATE_NULL;
+        if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) != GST_STATE_CHANGE_SUCCESS ||
+            state != GST_STATE_PLAYING)
+        {
+            std::lock_guard<std::mutex> lk(ctx->status_mutex);
+            ctx->status.running = false;
+            ctx->status.last_error = "Failed to enter PLAYING";
+            LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
+
+            gst_element_set_state(pipeline, GST_STATE_NULL);
+            if (bus) gst_object_unref(bus);
+            gst_object_unref(pipeline);
+
+            std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
+            continue;
+        }
+
+        // pipeline 成功启动
+        {
+            std::lock_guard<std::mutex> lk(ctx->status_mutex);
+            ctx->status.running = true;
+            ctx->status.last_error.clear();
+        }
+        LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
+
+        auto launch_tp = std::chrono::steady_clock::now();
+
+        bool need_restart = false;
+
+        // 5) 主循环：检测停帧或 error
+        while (ctx->thread_running)
+        {
+            auto now = std::chrono::steady_clock::now();
+            int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+            int64_t last_ms = ctx->last_frame_ms.load(std::memory_order_relaxed);
+
+            if (last_ms == 0)
+            {
+                auto startup_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - launch_tp).count();
+                if (startup_ms > START_TIMEOUT_MS)
+                {
+                    std::lock_guard<std::mutex> lk(ctx->status_mutex);
+                    ctx->status.running = false;
+                    ctx->status.last_error = "No frames detected at startup";
+                    LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
+                    need_restart = true;
+                    break;
+                }
+            }
+            else
+            {
+                int64_t idle_ms = now_ms - last_ms;
+                if (idle_ms > NO_FRAME_TIMEOUT_MS)
+                {
+                    std::lock_guard<std::mutex> lk(ctx->status_mutex);
+                    ctx->status.running = false;
+                    ctx->status.last_error = "Frame stalled for " + std::to_string(idle_ms) + " ms";
+                    LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
+                    need_restart = true;
+                    break;
+                }
+                else
+                {
+                    std::lock_guard<std::mutex> lk(ctx->status_mutex);
+                    ctx->status.running = true;
+                    ctx->status.last_error.clear();
+                }
+            }
+
+            // 错误消息检测
+            GstMessage* msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
+                                                         (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
+
+            if (!msg) continue;
+
+            if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
+            {
+                GError* err = nullptr;
+                gst_message_parse_error(msg, &err, nullptr);
+                std::lock_guard<std::mutex> lk(ctx->status_mutex);
+                ctx->status.running = false;
+                ctx->status.last_error = err ? err->message : "GStreamer error";
+                LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
+                if (err) g_error_free(err);
+                need_restart = true;
+            }
+            else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
+            {
+                std::lock_guard<std::mutex> lk(ctx->status_mutex);
+                ctx->status.running = false;
+                ctx->status.last_error = "EOS (End of stream)";
+                LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
+                need_restart = true;
+            }
+
+            gst_message_unref(msg);
+            if (need_restart) break;
+        }
+
+        // 清理 & 重建
+        gst_element_set_state(pipeline, GST_STATE_NULL);
+        if (bus) gst_object_unref(bus);
+        gst_object_unref(pipeline);
+
+        if (ctx->thread_running)
+        {
+            LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
+            std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
+        }
+    }
+
+    {
+        std::lock_guard<std::mutex> lk(ctx->status_mutex);
+        ctx->status.running = false;
+    }
+
+    LOG_INFO("[RTMP] Stream thread exited for " + key);
+}
+
+// ========== 启停与状态 ==========
+void RTMPManager::start_all()
+{
+    LOG_INFO("[RTMP] Starting enabled record streams...");
+    std::lock_guard<std::mutex> lock(streams_mutex);
+
+    int delay_ms = 0;
+    for (const auto& cam : g_app_config.cameras)
+    {
+        // 跳过未启用摄像头
+        if (!cam.enabled)
+        {
+            LOG_INFO("[RTMP] Skip disabled camera: " + cam.name);
+            continue;
+        }
+
+        auto key = make_key(cam.name);
+        if (streams.find(key) != streams.end())
+        {
+            LOG_INFO("[RTMP] Stream already running: " + key);
+            continue;
+        }
+
+        auto ctx = std::make_unique<StreamContext>();
+        ctx->thread_running.store(true);
+
+        ctx->thread = std::thread(
+            [cam, ptr = ctx.get(), delay_ms]()
+            {
+                std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
+                stream_loop(cam, ptr);
+            });
+
+        streams.emplace(key, std::move(ctx));
+        delay_ms += 200;
+    }
+}
+
+void RTMPManager::stop_all()
+{
+    std::lock_guard<std::mutex> lock(streams_mutex);
+    for (auto& kv : streams) kv.second->thread_running.store(false);
+    for (auto& kv : streams)
+        if (kv.second->thread.joinable()) kv.second->thread.join();
+    streams.clear();
+}
+
+bool RTMPManager::is_streaming(const std::string& cam_name)
+{
+    std::lock_guard<std::mutex> lock(streams_mutex);
+    auto it = streams.find(make_key(cam_name));
+    if (it == streams.end()) return false;
+
+    auto& ctx = *(it->second);
+    std::lock_guard<std::mutex> lk(ctx.status_mutex);
+    return ctx.status.running;
+}
+
+std::string RTMPManager::get_stream_url(const std::string& cam_name)
+{
+    std::string ip = get_ip_address("enP2p33s0");
+    if (ip.empty()) ip = "127.0.0.1";
+    return "rtsp://" + ip + ":18554/" + cam_name;
+}
+// ========== 汇总状态 ==========
+std::vector<RTMPManager::ChannelInfo> RTMPManager::get_all_channels_status()
+{
+    std::vector<ChannelInfo> result;
+    std::lock_guard<std::mutex> lock(streams_mutex);
+
+    for (size_t i = 0; i < g_app_config.cameras.size(); ++i)
+    {
+        const auto& cam = g_app_config.cameras[i];
+        auto key = make_key(cam.name);
+
+        ChannelInfo ch;
+        ch.loc = static_cast<int>(i);
+        ch.url.clear();
+        ch.running = false;
+        ch.reason = "Not started";
+
+        auto it = streams.find(key);
+        if (it != streams.end())
+        {
+            auto& ctx = *(it->second);
+
+            std::lock_guard<std::mutex> lk(ctx.status_mutex);
+            auto& status = it->second->status;
+
+            ch.running = status.running;
+            if (status.running)
+            {
+                ch.url = get_stream_url(cam.name);
+                ch.reason.clear();
+            }
+            else
+            {
+                ch.reason = status.last_error.empty() ? "Unknown error" : status.last_error;
+            }
+        }
+
+        result.push_back(ch);
+    }
+    return result;
+}

src/rtsp_manager.cpp

@@ -1,382 +0,0 @@
-// rtsp_manager.cpp
-#include "rtsp_manager.hpp"
-
-#include <fcntl.h>
-#include <linux/videodev2.h>
-#include <sys/ioctl.h>
-
-#include <algorithm>
-#include <chrono>
-#include <iostream>
-#include <thread>
-#include <unordered_map>
-
-#include "logger.hpp"
-
-// 静态变量定义
-GMainLoop* RTSPManager::loop = nullptr;
-GMainContext* RTSPManager::main_context = nullptr;
-GstRTSPServer* RTSPManager::server = nullptr;
-
-std::unordered_map<std::string, bool> RTSPManager::streaming_status;
-std::unordered_map<std::string, GstRTSPMediaFactory*> RTSPManager::mounted_factories;
-std::mutex RTSPManager::mounted_factories_mutex;
-
-// ==============================
-// ✅ 不再保存 GstRTSPMedia* 裸指针
-// 改为：每路 camera 的“活跃 session 计数”
-// ==============================
-static std::unordered_map<std::string, int> g_client_count;
-static std::mutex g_client_count_mutex;
-
-// 日志降噪：media-configure 过于频繁时只提示
-static std::unordered_map<std::string, std::chrono::steady_clock::time_point> last_media_ts;
-static std::mutex last_media_ts_mutex;
-
-bool set_v4l2_format(const std::string& dev, int width, int height)
-{
-    int fd = open(dev.c_str(), O_RDWR);
-    if (fd < 0)
-    {
-        LOG_ERROR("Failed to open " + dev);
-        return false;
-    }
-
-    struct v4l2_format fmt;
-    memset(&fmt, 0, sizeof(fmt));
-    fmt.type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
-
-    // 读格式，若相同则略过
-    if (ioctl(fd, VIDIOC_G_FMT, &fmt) == 0)
-    {
-        bool match = true;
-        if (fmt.fmt.pix_mp.width != (unsigned int)width || fmt.fmt.pix_mp.height != (unsigned int)height ||
-            fmt.fmt.pix_mp.pixelformat != V4L2_PIX_FMT_NV12)
-        {
-            match = false;
-        }
-
-        if (match)
-        {
-            close(fd);
-            LOG_INFO("[RTSP] V4L2 format already NV12 " + std::to_string(width) + "x" + std::to_string(height) +
-                     " for " + dev);
-            return true;
-        }
-    }
-
-    memset(&fmt, 0, sizeof(fmt));
-    fmt.type = V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE;
-    fmt.fmt.pix_mp.width = width;
-    fmt.fmt.pix_mp.height = height;
-    fmt.fmt.pix_mp.pixelformat = V4L2_PIX_FMT_NV12;
-    fmt.fmt.pix_mp.num_planes = 1;
-
-    if (ioctl(fd, VIDIOC_S_FMT, &fmt) < 0)
-    {
-        LOG_ERROR("VIDIOC_S_FMT failed for " + dev);
-        close(fd);
-        return false;
-    }
-
-    LOG_INFO("[RTSP] Set V4L2 format to NV12 " + std::to_string(width) + "x" + std::to_string(height) + " for " + dev);
-    close(fd);
-    return true;
-}
-
-void RTSPManager::init()
-{
-    gst_init(nullptr, nullptr);
-    LOG_INFO("[RTSP] GStreamer initialized.");
-}
-
-GstRTSPMediaFactory* RTSPManager::create_media_factory(const Camera& cam)
-{
-    // 启动前把 v4l2 格式设成我们想要的
-    set_v4l2_format(cam.device, cam.width, cam.height);
-
-    int w = cam.width;
-    int h = cam.height;
-
-    std::string caps =
-        "video/x-raw,format=NV12,"
-        "width=" +
-        std::to_string(w) + ",height=" + std::to_string(h) + ",framerate=" + std::to_string(cam.fps) + "/1";
-
-    // 关键：tee 一路给 RTSP，一路给 fakesink，保持 v4l2src / mpph264enc 永远在跑
-    std::string launch_str = "( v4l2src device=" + cam.device +
-                             " io-mode=2 is-live=true do-timestamp=true"
-                             " ! " +
-                             caps +
-                             " ! tee name=t "
-                             " ! queue leaky=downstream max-size-time=0 max-size-bytes=0 max-size-buffers=0"
-                             " ! mpph264enc name=enc rc-mode=cbr bps=" +
-                             std::to_string(cam.bitrate) + " gop=" + std::to_string(cam.fps) +
-                             " header-mode=1"
-                             " ! h264parse"
-                             " ! rtph264pay name=pay0 pt=96 config-interval=1 "
-                             " t. ! queue leaky=downstream max-size-buffers=3 max-size-bytes=0 max-size-time=0"
-                             " ! fakesink sync=false async=false )";
-
-    LOG_INFO("[RTSP] Launch for " + cam.name + ": " + launch_str);
-
-    GstRTSPMediaFactory* factory = gst_rtsp_media_factory_new();
-    gst_rtsp_media_factory_set_launch(factory, launch_str.c_str());
-
-    // 所有客户端共享同一个 pipeline，避免频繁拉起 v4l2 / encoder
-    gst_rtsp_media_factory_set_shared(factory, TRUE);
-
-    // 客户端断开时不要乱 suspend/reset
-    gst_rtsp_media_factory_set_suspend_mode(factory, GST_RTSP_SUSPEND_MODE_NONE);
-
-    g_signal_connect_data(factory, "media-configure", G_CALLBACK(on_media_created), g_strdup(cam.name.c_str()),
-                          (GClosureNotify)g_free, (GConnectFlags)0);
-
-    return factory;
-}
-
-void RTSPManager::start(const std::vector<Camera>& cams)
-{
-    server = gst_rtsp_server_new();
-    gst_rtsp_server_set_service(server, "8554");
-
-    // ✅ attach 之前设置 backlog
-    gst_rtsp_server_set_backlog(server, 32);
-
-    loop = g_main_loop_new(nullptr, FALSE);
-    main_context = g_main_loop_get_context(loop);
-
-    GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server);
-
-    for (const auto& cam : cams)
-    {
-        if (!cam.enabled) continue;
-
-        GstRTSPMediaFactory* factory = create_media_factory(cam);
-        std::string mount_point = "/" + cam.name;
-
-        gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory);
-
-        {
-            std::lock_guard<std::mutex> lock(mounted_factories_mutex);
-            mounted_factories[cam.name] = factory;
-            streaming_status[cam.name] = true;
-        }
-
-        // 初始化计数
-        {
-            std::lock_guard<std::mutex> lock(g_client_count_mutex);
-            g_client_count.emplace(cam.name, 0);
-        }
-
-        LOG_INFO("[RTSP] Camera '" + cam.name + "' mounted at rtsp://0.0.0.0:8554" + mount_point);
-    }
-
-    g_object_unref(mounts);
-
-    gst_rtsp_server_attach(server, nullptr);
-
-    LOG_INFO("[RTSP] Server running on rtsp://0.0.0.0:8554");
-    g_main_loop_run(loop);
-
-    if (server)
-    {
-        g_object_unref(server);
-        server = nullptr;
-    }
-    if (loop)
-    {
-        g_main_loop_unref(loop);
-        loop = nullptr;
-    }
-
-    LOG_INFO("[RTSP] Server stopped.");
-}
-
-void RTSPManager::on_media_created(GstRTSPMediaFactory*, GstRTSPMedia* media, gpointer user_data)
-{
-    const char* cam_name = static_cast<const char*>(user_data);
-
-    // ✅ 只做日志降噪：别让“抑制”影响生命周期处理
-    bool noisy = false;
-    auto now = std::chrono::steady_clock::now();
-    {
-        std::lock_guard<std::mutex> lock(last_media_ts_mutex);
-        auto& last = last_media_ts[cam_name];
-        if (last.time_since_epoch().count() != 0 && now - last < std::chrono::seconds(2)) noisy = true;
-        last = now;
-    }
-
-    int count_after = 0;
-    {
-        std::lock_guard<std::mutex> lock(g_client_count_mutex);
-        int& c = g_client_count[cam_name];  // 若不存在会创建；这里无所谓
-        c++;
-        count_after = c;
-    }
-
-    if (noisy)
-    {
-        LOG_WARN(std::string("[RTSP] media-configure frequent: ") + cam_name +
-                 " (active_sessions=" + std::to_string(count_after) + ")");
-    }
-    else
-    {
-        LOG_INFO(std::string("[RTSP] media-configure for camera: ") + cam_name +
-                 " (active_sessions=" + std::to_string(count_after) + ")");
-    }
-
-    // ✅ 生命周期绑定：unprepared 一定会对应减计数
-    g_signal_connect_data(media, "unprepared", G_CALLBACK(on_media_unprepared), g_strdup(cam_name),
-                          (GClosureNotify)g_free, (GConnectFlags)0);
-}
-
-void RTSPManager::on_media_unprepared(GstRTSPMedia* /*media*/, gpointer user_data)
-{
-    const char* cam_name = static_cast<const char*>(user_data);
-
-    int count_after = 0;
-    {
-        std::lock_guard<std::mutex> lock(g_client_count_mutex);
-        auto it = g_client_count.find(cam_name);
-        if (it == g_client_count.end())
-        {
-            // 可能发生在 unmount 后 teardown 仍在进行的边界情况
-            LOG_WARN(std::string("[RTSP] media-unprepared but cam not in client_count: ") + cam_name);
-            return;
-        }
-
-        if (it->second > 0) it->second--;
-        count_after = it->second;
-    }
-
-    LOG_INFO(std::string("[RTSP] media-unprepared: ") + cam_name + " (active_sessions=" + std::to_string(count_after) +
-             ")");
-
-    // ✅ 不要 g_object_unref(media) —— gst-rtsp-server 会处理
-}
-
-gboolean RTSPManager::mount_camera_in_main(gpointer data)
-{
-    Camera* cam = static_cast<Camera*>(data);
-    if (!cam || !server)
-    {
-        delete cam;
-        return G_SOURCE_REMOVE;
-    }
-
-    GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server);
-
-    std::string mount_point = "/" + cam->name;
-    GstRTSPMediaFactory* factory = create_media_factory(*cam);
-
-    gst_rtsp_mount_points_add_factory(mounts, mount_point.c_str(), factory);
-    g_object_unref(mounts);
-
-    {
-        std::lock_guard<std::mutex> lock(mounted_factories_mutex);
-        mounted_factories[cam->name] = factory;
-        streaming_status[cam->name] = true;
-    }
-
-    {
-        std::lock_guard<std::mutex> lock(g_client_count_mutex);
-        g_client_count[cam->name] = 0;
-    }
-
-    LOG_INFO("[RTSP] Camera '" + cam->name + "' mounted at rtsp://localhost:8554" + mount_point);
-
-    delete cam;
-    return G_SOURCE_REMOVE;
-}
-
-gboolean RTSPManager::unmount_camera_in_main(gpointer data)
-{
-    Camera* cam = static_cast<Camera*>(data);
-    if (!cam || !server)
-    {
-        delete cam;
-        return G_SOURCE_REMOVE;
-    }
-
-    std::string cam_name = cam->name;
-    std::string mount_point = "/" + cam_name;
-
-    // ✅ 仅移除 mount；media teardown 交给 gst-rtsp-server
-    GstRTSPMountPoints* mounts = gst_rtsp_server_get_mount_points(server);
-    if (mounts)
-    {
-        gst_rtsp_mount_points_remove_factory(mounts, mount_point.c_str());
-        g_object_unref(mounts);
-    }
-
-    {
-        std::lock_guard<std::mutex> lock(mounted_factories_mutex);
-        auto it = mounted_factories.find(cam_name);
-        if (it != mounted_factories.end())
-        {
-            // 为了避免 double-unref：这里不手动 unref factory
-            mounted_factories.erase(it);
-        }
-        streaming_status[cam_name] = false;
-    }
-
-    {
-        std::lock_guard<std::mutex> lock(g_client_count_mutex);
-        g_client_count.erase(cam_name);
-    }
-
-    LOG_INFO("[RTSP] Camera '" + cam_name + "' unmounted.");
-    delete cam;
-    return G_SOURCE_REMOVE;
-}
-
-void RTSPManager::mount_camera(const Camera& cam)
-{
-    Camera* camCopy = new Camera(cam);
-    g_main_context_invoke(
-        main_context, [](gpointer data) -> gboolean { return RTSPManager::mount_camera_in_main(data); }, camCopy);
-}
-
-void RTSPManager::unmount_camera(const Camera& cam)
-{
-    Camera* camCopy = new Camera(cam);
-    g_main_context_invoke(
-        main_context, [](gpointer data) -> gboolean { return RTSPManager::unmount_camera_in_main(data); }, camCopy);
-}
-
-bool RTSPManager::is_streaming(const std::string& cam_name)
-{
-    std::lock_guard<std::mutex> lock(mounted_factories_mutex);
-    auto it = streaming_status.find(cam_name);
-    return it != streaming_status.end() ? it->second : false;
-}
-
-bool RTSPManager::is_any_streaming()
-{
-    std::lock_guard<std::mutex> lock(mounted_factories_mutex);
-    for (auto& kv : streaming_status)
-        if (kv.second) return true;
-    return false;
-}
-
-void RTSPManager::stop()
-{
-    if (!loop) return;
-
-    if (server) gst_rtsp_server_set_backlog(server, 0);  // optional
-
-    if (main_context)
-    {
-        g_main_context_invoke(
-            main_context,
-            [](gpointer data) -> gboolean
-            {
-                g_main_loop_quit(static_cast<GMainLoop*>(data));
-                return G_SOURCE_REMOVE;
-            },
-            loop);
-    }
-
-    g_main_loop_quit(loop);
-}

src/serial_AT.cpp

@@ -0,0 +1,162 @@
+#include "serial_AT.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <mutex>
+#include <sstream>
+#include <thread>
+
+#include "logger.hpp"
+#include "serial_port.h"
+
+// ================== 运行控制 ==================
+static std::atomic<bool> serial_at_running{false};
+
+// ================== AT 任务结构 ==================
+struct AtTask
+{
+    std::string cmd;
+    int interval_sec;  // 周期（秒）
+    int max_retries;   // -1 表示无限
+    int sent_count;
+    std::chrono::steady_clock::time_point last_sent;
+};
+
+static std::unique_ptr<SerialPort> serial_at;
+static std::thread serial_at_sender;
+static std::mutex at_tasks_mutex;
+
+// ================== AT 任务列表 ==================
+//
+// CSQ  : 高频心跳（信号强度）
+// QENG : 低频全量无线质量
+//
+static std::vector<AtTask> at_tasks = {
+    {"AT+CSQ", 5, -1, 0, {}},                   // 每 5 秒
+    {"AT+QENG=\"servingcell\"", 15, -1, 0, {}}  // 每 30 秒
+};
+
+// ================== 发送线程 ==================
+static void serial_at_send_loop()
+{
+    LOG_INFO("[serial_at] Sender thread started");
+
+    while (serial_at_running.load(std::memory_order_relaxed))
+    {
+        if (!serial_at || !serial_at->is_open())
+        {
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        auto now = std::chrono::steady_clock::now();
+
+        {
+            std::lock_guard<std::mutex> lock(at_tasks_mutex);
+
+            for (auto& task : at_tasks)
+            {
+                bool need_send = false;
+
+                if (task.last_sent.time_since_epoch().count() == 0) { need_send = true; }
+                else
+                {
+                    auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - task.last_sent).count();
+                    if (elapsed >= task.interval_sec) need_send = true;
+                }
+
+                if (need_send)
+                {
+                    serial_at->send_data(task.cmd + "\r\n");
+                    task.sent_count++;
+                    task.last_sent = now;
+                }
+            }
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+
+    LOG_INFO("[serial_at] Sender thread exiting");
+}
+
+// ================== 接收处理 ==================
+static void handle_serial_at_data(const std::string& data)
+{
+    std::istringstream iss(data);
+    std::string line;
+
+    while (std::getline(iss, line))
+    {
+        // trim
+        line.erase(0, line.find_first_not_of(" \t\r\n"));
+        line.erase(line.find_last_not_of(" \t\r\n") + 1);
+
+        if (line.empty() || line == "OK") continue;
+
+        // ---------- CSQ ----------
+        // +CSQ: <rssi>,<ber>
+        if (line.rfind("+CSQ:", 0) == 0)
+        {
+            int rssi = -1, ber = -1;
+            if (sscanf(line.c_str(), "+CSQ: %d,%d", &rssi, &ber) == 2)
+            {
+                int dbm = (rssi >= 0 && rssi <= 31) ? (-113 + rssi * 2) : -999;
+
+                LOG_INFO("[serial_at] CSQ rssi=" + std::to_string(rssi) + " (" + std::to_string(dbm) +
+                         " dBm)"
+                         " ber=" +
+                         std::to_string(ber));
+            }
+            continue;
+        }
+
+        // ---------- QENG servingcell ----------
+        if (line.rfind("+QENG:", 0) == 0 && line.find("servingcell") != std::string::npos)
+        {
+            // 先只打原始信息，后续你再精解析
+            LOG_INFO("[serial_at] QENG: " + line);
+            continue;
+        }
+
+        // ---------- 其它 AT 回显 ----------
+        // 默认忽略，避免日志污染
+        // 如遇现场问题，可临时加 LOG_INFO 打印
+    }
+}
+
+// ================== 初始化 ==================
+void init_serial_at(const std::string& device, int baudrate)
+{
+    if (serial_at_running.load())
+    {
+        LOG_WARN("[serial_at] Already running");
+        return;
+    }
+
+    serial_at_running.store(true, std::memory_order_relaxed);
+
+    serial_at = std::make_unique<SerialPort>("serial_at", device, baudrate, 5);
+    serial_at->set_receive_callback(handle_serial_at_data);
+    serial_at->start();
+
+    serial_at_sender = std::thread(serial_at_send_loop);
+}
+
+// ================== 停止 ==================
+void stop_serial_at()
+{
+    if (!serial_at_running.load()) return;
+
+    LOG_INFO("[serial_at] Stopping...");
+
+    serial_at_running.store(false, std::memory_order_relaxed);
+
+    if (serial_at) serial_at->stop();
+
+    if (serial_at_sender.joinable()) serial_at_sender.join();
+
+    serial_at.reset();
+
+    LOG_INFO("[serial_at] Stopped cleanly");
+}

src/serial_port.cpp

@@ -0,0 +1,197 @@
+#include "serial_port.h"
+
+#include <fcntl.h>
+#include <termios.h>
+#include <unistd.h>
+
+#include <chrono>
+#include <cstring>
+#include <thread>
+
+// --------- 波特率映射 ---------
+static speed_t baud_to_speed(int baud)
+{
+    switch (baud)
+    {
+        case 9600:
+            return B9600;
+        case 19200:
+            return B19200;
+        case 38400:
+            return B38400;
+        case 57600:
+            return B57600;
+        case 115200:
+            return B115200;
+#ifdef B230400
+        case 230400:
+            return B230400;
+#endif
+        default:
+            return B115200;
+    }
+}
+
+SerialPort::SerialPort(const std::string& id, const std::string& device, int baudrate, int retry_interval)
+    : id_(id), device_(device), baudrate_(baudrate), retry_interval_(retry_interval)
+{
+}
+
+SerialPort::~SerialPort() { stop(); }
+
+void SerialPort::start()
+{
+    stop_flag_ = false;
+    reconnect_thread_ = std::thread(&SerialPort::reconnect_loop, this);
+}
+
+void SerialPort::stop()
+{
+    stop_flag_ = true;
+    running_ = false;
+
+    // 先关闭 fd，打断 reader 的阻塞 read()
+    close_port();
+
+    if (reader_thread_.joinable()) reader_thread_.join();
+    if (reconnect_thread_.joinable()) reconnect_thread_.join();
+}
+
+bool SerialPort::open_port()
+{
+    fd_ = open(device_.c_str(), O_RDWR | O_NOCTTY | O_SYNC);
+    if (fd_ < 0)
+    {
+        LOG_ERROR("[" + id_ + "] Failed to open " + device_ + ": " + strerror(errno));
+        return false;
+    }
+
+    if (!configure_port(fd_))
+    {
+        LOG_ERROR("[" + id_ + "] Failed to configure " + device_);
+        close(fd_);
+        fd_ = -1;
+        return false;
+    }
+
+    running_ = true;
+    reader_thread_ = std::thread(&SerialPort::reader_loop, this);
+
+    LOG_INFO("[" + id_ + "] Opened serial port " + device_ + " at " + std::to_string(baudrate_) + " baud");
+    return true;
+}
+
+void SerialPort::close_port()
+{
+    running_ = false;
+
+    if (fd_ >= 0)
+    {
+        close(fd_);
+        fd_ = -1;
+    }
+}
+
+bool SerialPort::is_open() const { return fd_ >= 0; }
+
+bool SerialPort::send_data(const std::vector<uint8_t>& data)
+{
+    if (fd_ < 0) return false;
+
+    std::lock_guard<std::mutex> lock(send_mutex_);
+    ssize_t n = write(fd_, data.data(), data.size());
+    return n == static_cast<ssize_t>(data.size());
+}
+
+bool SerialPort::send_data(const std::string& data)
+{
+    return send_data(std::vector<uint8_t>(data.begin(), data.end()));
+}
+
+void SerialPort::set_receive_callback(ReceiveCallback cb) { receive_callback_ = std::move(cb); }
+
+void SerialPort::set_receive_callback(ReceiveStringCallback cb)
+{
+    receive_callback_ = [cb](const std::vector<uint8_t>& data) { cb(std::string(data.begin(), data.end())); };
+}
+
+void SerialPort::reader_loop()
+{
+    std::vector<uint8_t> buffer(1024);
+
+    while (running_ && !stop_flag_)
+    {
+        int n = read(fd_, buffer.data(), buffer.size());
+
+        if (n > 0)
+        {
+            if (receive_callback_) receive_callback_({buffer.begin(), buffer.begin() + n});
+        }
+        else if (n < 0)
+        {
+            LOG_ERROR("[" + id_ + "] Read error: " + std::string(strerror(errno)));
+            break;
+        }
+        // n == 0：超时，正常情况，继续读
+    }
+
+    running_ = false;
+}
+
+static void interruptible_sleep(std::atomic<bool>& stop_flag, int seconds)
+{
+    using namespace std::chrono;
+    for (int i = 0; i < seconds * 10 && !stop_flag.load(std::memory_order_relaxed); ++i)
+        std::this_thread::sleep_for(100ms);
+}
+
+void SerialPort::reconnect_loop()
+{
+    int current_interval = retry_interval_;
+    const int max_interval = 300;
+
+    while (!stop_flag_)
+    {
+        if (open_port())
+        {
+            // 不 join reader，只等 stop_flag_
+            while (running_ && !stop_flag_) std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+            close_port();
+            LOG_WARN("[" + id_ + "] Port closed, will retry");
+            current_interval = retry_interval_;
+        }
+        else
+        {
+            LOG_INFO("[" + id_ + "] Connect failed, retry in " + std::to_string(current_interval) + "s");
+
+            interruptible_sleep(stop_flag_, current_interval);
+            current_interval = std::min(current_interval * 2, max_interval);
+        }
+    }
+}
+
+bool SerialPort::configure_port(int fd)
+{
+    struct termios tty{};
+    if (tcgetattr(fd, &tty) != 0) return false;
+
+    speed_t spd = baud_to_speed(baudrate_);
+    cfsetospeed(&tty, spd);
+    cfsetispeed(&tty, spd);
+
+    tty.c_cflag = (tty.c_cflag & ~CSIZE) | CS8;
+    tty.c_iflag &= ~IGNBRK;
+    tty.c_lflag = 0;
+    tty.c_oflag = 0;
+    tty.c_cc[VMIN] = 0;
+    tty.c_cc[VTIME] = 10;
+
+    tty.c_iflag &= ~(IXON | IXOFF | IXANY);
+    tty.c_cflag |= (CLOCAL | CREAD);
+    tty.c_cflag &= ~(PARENB | PARODD);
+    tty.c_cflag &= ~CSTOPB;
+    tty.c_cflag &= ~CRTSCTS;
+
+    return tcsetattr(fd, TCSANOW, &tty) == 0;
+}