新增网络质量监测日志

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

@@ -10,6 +10,7 @@
 #include "logger.hpp"
 #include "mqtt_client_wrapper.hpp"
 #include "rtmp_manager.hpp"
+#include "serial_AT.hpp"
 
 std::atomic<bool> g_running(true);
 
@@ -48,6 +49,7 @@ int main()
         return -1;
     }
 
+    init_serial_at("/dev/ttyUSB3", 115200);
     // ---------- GStreamer ----------
     RTMPManager::init();
 
@@ -70,6 +72,8 @@ int main()
     // ---------- 退出 ----------
     LOG_INFO("[MAIN] Shutdown requested. Stopping services...");
 
+    stop_serial_at();
+
     RTMPManager::stop_all();
 
     if (mqtt_thread.joinable()) mqtt_thread.join();

src/rtmp_manager.cpp

@@ -63,41 +63,114 @@ void RTMPManager::init()
 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 width = cam.width;
+    const int out_width = cam.width;  // 推流分辨率（如 1280 / 960 / 720）
-    const int height = cam.height;
+    const int out_height = cam.height;
     const int fps = cam.fps;
     const int bitrate = cam.bitrate;
 
-    // MediaMTX 中的 stream key
+    // 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 说明：
+     * A 方案 Pipeline：
-     * v4l2src -> mpph264enc -> h264parse -> flvmux -> rtmpsink
      *
-     * - 不使用 tee（降低死锁概率）
+     * v4l2src (原生 1280x960)
-     * - 不引入音频（MediaMTX 不强制要求）
+     *   -> videoscale + caps（编码前缩放）
-     * - 纯视频、纯 RTMP、纯 TCP
+     *   -> queue（低延迟）
+     *   -> mpph264enc（CBR）
+     *   -> h264parse
+     *   -> flvmux
+     *   -> rtmpsink
      */
     std::string pipeline_str = "v4l2src name=src device=" + cam.device +
-                               " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
+                               " io-mode=dmabuf "
-                               ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
+
+                               // ⭐ 相机真实输出：原生 1280x960
+                               "! video/x-raw,format=NV12,"
+                               "width=1280,height=960,"
+                               "framerate=" +
+                               std::to_string(fps) +
                                "/1 "
-                               " ! mpph264enc bps=" +
+
-                               std::to_string(bitrate) + " gop=" + std::to_string(fps) +
+                               // ⭐ 编码前缩放（关键）
-                               " rc-mode=cbr "
+                               "! videoscale "
-                               " ! h264parse name=parse "
+                               "! video/x-raw,"
-                               " ! flvmux streamable=true "
+                               "width=" +
-                               " ! rtmpsink location=\"" +
+                               std::to_string(out_width) + ",height=" + std::to_string(out_height) +
+                               " "
+
+                               // ⭐ 低延迟队列，防止下游阻塞反噬 v4l2
+                               "! queue max-size-buffers=2 max-size-time=0 leaky=downstream "
+
+                               // ⭐ 硬件编码，SIM 卡友好
+                               "! mpph264enc "
+                               "rc-mode=cbr "
+                               "bps=" +
+                               std::to_string(bitrate) +
+                               " "
+                               "gop=" +
+                               std::to_string(fps) +
+                               " "
+                               "header-mode=each-idr "
+                               "profile=baseline "
+
+                               "! h264parse config-interval=1 "
+                               "! flvmux streamable=true "
+                               "! rtmpsink location=\"" +
                                rtmp_url +
                                "\" "
-                               "   sync=false async=false";
+                               "sync=false async=false";
 
     GError* error = nullptr;
     GstElement* pipeline = gst_parse_launch(pipeline_str.c_str(), &error);

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\"", 30, -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;
+}