bonus-edge-proxy/src/rknn/video_service.cc

// video_service.cc (修改后)
#include "video_service.h"
#include "opencv2/imgproc/imgproc.hpp"
#include "spdlog/spdlog.h"
#include <stdio.h>

VideoService::VideoService(std::unique_ptr<IAnalysisModule> module,
                           std::string input_url, std::string output_rtsp_url,
                           nlohmann::json module_config)
    : module_(std::move(module)), input_url_(input_url),
      output_rtsp_url_(output_rtsp_url),
      module_config_(std::move(module_config)), running_(false) {
  log_prefix_ = "[VideoService: " + input_url + "]";
  spdlog::info("{} Created. Input: {}, Output: {}", log_prefix_,
               input_url_.c_str(), output_rtsp_url_.c_str());
}

VideoService::~VideoService() {
  if (running_) {
    stop();
  }
}

bool VideoService::start() {
  if (!module_ || !module_->init(module_config_)) {
    spdlog::error("{} Failed to initialize analysis module!", log_prefix_);
    return false;
  }
  spdlog::info("{} Analysis module initialized successfully.", log_prefix_);

  std::string gst_input_pipeline = "rtspsrc location=" + input_url_ +
                                   " latency=0 protocols=tcp ! "
                                   "rtph265depay ! "
                                   "h265parse ! "
                                   "mppvideodec format=16 ! "
                                   "videoconvert ! "
                                   "video/x-raw,format=BGR ! "
                                   "appsink";

  spdlog::info("Try to Open RTSP Stream");
  capture_.open(gst_input_pipeline, cv::CAP_GSTREAMER);

  if (!capture_.isOpened()) {
    printf("Error: Could not open RTSP stream: %s\n", input_url_.c_str());
    return false;
  } else {
    spdlog::info("RTSP Stream Opened!");
  }

  frame_width_ = static_cast<int>(capture_.get(cv::CAP_PROP_FRAME_WIDTH));
  frame_height_ = static_cast<int>(capture_.get(cv::CAP_PROP_FRAME_HEIGHT));
  frame_fps_ = capture_.get(cv::CAP_PROP_FPS);
  if (frame_fps_ <= 0)
    frame_fps_ = 25.0;

  if (frame_width_ == 0 || frame_height_ == 0) {
    spdlog::error("{} Failed to get valid frame width or height from GStreamer "
                  "pipeline (got {}x{}).",
                  log_prefix_, frame_width_, frame_height_);
    spdlog::error("{} This usually means the RTSP stream is unavailable or the "
                  "GStreamer input pipeline (mppvideodec?) failed.",
                  log_prefix_);

    cv::Mat test_frame;
    if (capture_.read(test_frame) && !test_frame.empty()) {
      frame_width_ = test_frame.cols;
      frame_height_ = test_frame.rows;
      spdlog::info(
          "{} Successfully got frame size by reading first frame: {}x{}",
          log_prefix_, frame_width_, frame_height_);

      {
        std::lock_guard<std::mutex> lock(frame_mutex_);
        latest_frame_ = test_frame;
        new_frame_available_ = true;
      }
      frame_cv_.notify_one();

    } else {
      spdlog::error(
          "{} Failed to read first frame to determine size. Aborting.",
          log_prefix_);
      capture_.release();
      return false;
    }
  }

  printf("RTSP stream opened successfully! (%dx%d @ %.2f FPS)\n", frame_width_,
         frame_height_, frame_fps_);

  std::string gst_pipeline = "appsrc ! "
                             "queue max-size-buffers=2 leaky=downstream ! "
                             "video/x-raw,format=BGR ! "
                             "videoconvert ! "
                             "video/x-raw,format=NV12 ! "
                             "mpph264enc gop=25 rc-mode=fixqp qp-init=26 ! "
                             "h264parse ! "
                             "rtspclientsink location=" +
                             output_rtsp_url_ + " latency=0 protocols=tcp";

  printf("Using GStreamer output pipeline: %s\n", gst_pipeline.c_str());

  writer_.open(gst_pipeline, cv::CAP_GSTREAMER, 0, frame_fps_,
               cv::Size(frame_width_, frame_height_), true);

  if (!writer_.isOpened()) {
    printf("Error: Could not open VideoWriter with GStreamer pipeline.\n");
    capture_.release();
    return false;
  }
  printf("VideoWriter opened successfully.\n");

  running_ = true;
  reading_thread_ = std::thread(&VideoService::reading_loop, this);
  processing_thread_ = std::thread(&VideoService::processing_loop, this);
  printf("Processing thread started.\n");

  return true;
}

void VideoService::stop() {
  printf("Stopping VideoService...\n");
  running_ = false;

  frame_cv_.notify_all();

  if (reading_thread_.joinable()) {
    reading_thread_.join();
  }

  if (processing_thread_.joinable()) {
    processing_thread_.join();
  }
  printf("Processing thread joined.\n");

  if (capture_.isOpened()) {
    capture_.release();
  }
  if (writer_.isOpened()) {
    writer_.release();
  }
  module_->stop();
  module_.reset();

  printf("VideoService stopped.\n");
}

void VideoService::reading_loop() {
  cv::Mat frame;
  spdlog::info("Reading thread started.");

  while (running_) {
    if (!capture_.read(frame)) {
      spdlog::warn(
          "Reading loop: Failed to read frame from capture. Stopping service.");
      running_ = false;
      break;
    }

    if (frame.empty()) {
      continue;
    }

    {
      std::lock_guard<std::mutex> lock(frame_mutex_);
      latest_frame_ = frame;
      new_frame_available_ = true;
    }

    frame_cv_.notify_one();
  }

  frame_cv_.notify_all(); // 确保 processing_loop 也会退出
  spdlog::info("Reading loop finished.");
}

void VideoService::processing_loop() {
  cv::Mat frame;

  while (running_) {
    {
      // 1. (不变) 获取帧
      std::unique_lock<std::mutex> lock(frame_mutex_);

      frame_cv_.wait(lock, [&] { return new_frame_available_ || !running_; });

      if (!running_) {
        break;
      }

      frame = latest_frame_.clone();
      new_frame_available_ = false;
    }

    if (frame.empty()) {
      continue;
    }
    if (!module_->process(frame)) {
      // 模块报告处理失败
      spdlog::warn("{} Module failed to process frame. Skipping.", log_prefix_);
    }
    if (writer_.isOpened()) {
      writer_.write(frame);
    }
  }

  spdlog::info("VideoService: Processing loop finished.");
}