#include <opencv2/core/core.hpp>
#include <opencv2/imgproc.hpp>
#include <opencv2/videoio.hpp>
#include <opencv2/opencv.hpp>
#include "vp_rtsp_ffmpeg_src_node.h"
#include "../utils/vp_utils.h"
#include <cstdio>
#include <memory>
#include <stdexcept>
#include <iostream>
#include <vector>
#include <thread>
#include <unistd.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <cstdlib> // for setting environment variable

namespace vp_nodes
{
    vp_rtsp_ffmpeg_src_node::vp_rtsp_ffmpeg_src_node(std::string node_name,
                                                     int channel_index,
                                                     std::string rtsp_url,
                                                     float resize_ratio,
                                                     int skip_interval,
                                                     bool use_gpu,
                                                     std::string ffmpeg_format,
                                                     std::string ffmpeg_pix_fmt) : vp_src_node(node_name, channel_index, resize_ratio),
                                                                                   rtsp_url(rtsp_url),
                                                                                   skip_interval(skip_interval),
                                                                                   use_gpu(use_gpu),
                                                                                   ffmpeg_format(ffmpeg_format),
                                                                                   ffmpeg_pix_fmt(ffmpeg_pix_fmt)
    {
        assert(skip_interval >= 0 && skip_interval <= 9);
        VP_INFO(vp_utils::string_format("[%s] [%s]", node_name.c_str(), ffmpeg_format.c_str()));
        this->initialized();
    }

    std::vector<int> vp_rtsp_ffmpeg_src_node::initialize_stream_properties()
    {
        try
        {
            // FFmpeg command to probe stream metadata
            std::string ffprobe_command = vp_utils::string_format(ffprobe_template, rtsp_url.c_str());

            // Execute command and capture output
            FILE *pipe = popen(ffprobe_command.c_str(), "r");
            if (!pipe)
            {
                throw std::runtime_error("Failed to execute ffprobe command.");
            }

            char buffer[128];
            std::string result = "";
            while (fgets(buffer, sizeof(buffer), pipe) != nullptr)
            {
                result += buffer;
            }
            pclose(pipe);
            VP_INFO(vp_utils::string_format("FFPROBE RESULT:%s url: %s", result, rtsp_url.c_str()));

            // Parse the output (expected format: width,height,r_frame_rate)
            std::stringstream ss(result);
            std::string width_str, height_str, fps_str;
            std::getline(ss, width_str, ',');
            std::getline(ss, height_str, ',');
            std::getline(ss, fps_str, ',');

            // Parse width and height
            int video_width = std::stoi(width_str);
            int video_height = std::stoi(height_str);
            int fps = 25;

            // Parse r_frame_rate (e.g., "299/12" or "25")
            if (fps_str.find('/') != std::string::npos)
            {
                // Handle fractional frame rates
                std::stringstream fps_ss(fps_str);
                std::string numerator_str, denominator_str;
                std::getline(fps_ss, numerator_str, '/');
                std::getline(fps_ss, denominator_str, '/');

                // Convert to integer values
                int numerator = std::stoi(numerator_str);
                int denominator = std::stoi(denominator_str);

                // Calculate and round to the nearest integer
                fps = static_cast<int>(std::round(static_cast<float>(numerator) / denominator));
            }
            else
            {
                // Handle integer frame rates
                fps = std::stoi(fps_str);
            }

            // Optional: Correct minor deviations (e.g., if fps is close to 25)
            if (std::abs(fps - 25) < 1)
            {
                fps = 25; // Force fps to 25 if very close
            }

            // Debug log for parsed properties
            VP_INFO(vp_utils::string_format("Stream URL:%s Stream properties - Width: %d, Height: %d, FPS: %d",
                                            rtsp_url.c_str(), video_width, video_height, fps));

            return std::vector<int>{video_width, video_height, fps};
        }
        catch (...)
        {
            return std::vector<int>{0, 0, 0};
        }
    }

    vp_rtsp_ffmpeg_src_node::~vp_rtsp_ffmpeg_src_node()
    {
        deinitialized();
    }

    // 子进程启动 ffmpeg 并返回管道文件描述符
    int vp_rtsp_ffmpeg_src_node::start_ffmpeg_thread(const std::string &rtsp_url, int width, int height)
    {

        int pipe_fd[2];
        if (pipe(pipe_fd) == -1)
        {
            perror("pipe");
            return -1;
        }

        pid_t pid = fork();
        if (pid == -1)
        {
            perror("fork");
            return -1;
        }

        if (pid == 0) // 子进程
        {
            close(pipe_fd[0]);               // 关闭读端
            dup2(pipe_fd[1], STDOUT_FILENO); // 将管道写端绑定到标准输出

            // 根据 channel_index 奇偶来设置 GPU
            //int gpu_id = channel_index % total_gpus; // 根据 channel_index 和 GPU 数量进行分配
            // 设置环境变量以选择 GPU
            //setenv("CUDA_VISIBLE_DEVICES", std::to_string(gpu_id).c_str(), 1);

            // 构造 ffmpeg 命令
            execlp("ffmpeg", ("ffmpeg-" + std::to_string(channel_index)).c_str(),
                   "-loglevel", "warning",
                   "-hwaccel", "cuda",
                   "-rtsp_transport", "tcp",
                   "-i", rtsp_url.c_str(),
                   "-f", ffmpeg_format.c_str(),
                   "-pix_fmt", ffmpeg_pix_fmt.c_str(),
                   "-s", (std::to_string(width) + "x" + std::to_string(height)).c_str(),
                   "pipe:1",
                   nullptr);
            exit(EXIT_FAILURE); // 若 execlp 失败
        }
        else
        {                      // 父进程
            close(pipe_fd[1]); // 关闭写端
            return pipe_fd[0]; // 返回读端文件描述符
        }
    }

    // define how to read video from rtsp stream, create frame meta etc.
    // please refer to the implementation of vp_node::handle_run.
    void vp_rtsp_ffmpeg_src_node::handle_run()
    {
        std::vector<int> stream_infos = initialize_stream_properties();

        original_width = stream_infos[0];
        original_height = stream_infos[1];
        original_fps = stream_infos[2];


        int video_width = original_width;
        int video_height = original_height;
        // set true fps because skip some frames
        int fps = original_fps / (skip_interval + 1);

        cv::Mat frame(original_height, original_width, CV_8UC3);

        int skip = 0;
        int pipe_fd = -1;

        int frame_size = original_width * original_height * 3; // 假设 BGR24
        std::vector<unsigned char> frame_buffer(frame_size);

        bool pipe_ok = true;

        if (use_gpu)
        {
            pipe_fd = start_ffmpeg_thread(rtsp_url, original_width, original_height);
            VP_INFO(std::to_string(pipe_fd));
            if (pipe_fd < 0)
            {
                VP_ERROR(vp_utils::string_format("Failed to start cuda ffmpeg thread for  [%s]", rtsp_url.c_str()));
                use_gpu = false;
            }
        }

        while (alive)
        {
            // check if need workc
            gate.knock();

            if (!use_gpu)
            {
                // try to open capture
                if (!rtsp_capture.isOpened())
                {
                    if (!rtsp_capture.open(this->rtsp_url, cv::CAP_FFMPEG))
                    {
                        VP_WARN(vp_utils::string_format("[%s] open rtsp failed, try again...", node_name.c_str()));
                        continue;
                    }
                }
                if (video_width == 0 || video_height == 0 || fps == 0)
                {
                    video_width = rtsp_capture.get(cv::CAP_PROP_FRAME_WIDTH);
                    video_height = rtsp_capture.get(cv::CAP_PROP_FRAME_HEIGHT);
                    fps = rtsp_capture.get(cv::CAP_PROP_FPS);

                    original_fps = fps;
                    original_width = video_width;
                    original_height = video_height;

                    // set true fps because skip some frames
                    fps = fps / (skip_interval + 1);
                }
                // stream_info_hooker activated if need
                vp_stream_info stream_info{channel_index, original_fps, original_width, original_height, to_string()};
                invoke_stream_info_hooker(node_name, stream_info);

                rtsp_capture >> frame;
            }
            else
            {

                if (!pipe_ok)  // restart pipe_fd
                {
                    pipe_fd = start_ffmpeg_thread(rtsp_url, original_width, original_height);

                }

                size_t total_read = 0;

                // 循环读取，确保一帧完整数据
                while (total_read < frame_size)
                {
                    ssize_t bytes_read = read(pipe_fd, frame_buffer.data() + total_read, frame_size - total_read);
                    if (bytes_read < 0)
                    { // 读取错误
                        if (errno == EINTR)
                        {
                            // 系统调用被中断，重试读取
                            VP_DEBUG(vp_utils::string_format("Interrupted read for  [%s],re etrying...", rtsp_url.c_str()));
                            continue;
                        }
                        else if (errno == EAGAIN)
                        {
                            // 非阻塞模式下无数据可读
                            VP_DEBUG(vp_utils::string_format("Io data available yet for  [%s],re etrying...", rtsp_url.c_str()));
                            std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 等待片刻再试
                            continue;
                        }
                        else
                        {
                            // 其他不可恢复错误，记录日志并跳过帧
                            VP_DEBUG(vp_utils::string_format("Read error for [%s].skipping frame...", rtsp_url.c_str()));
                            break;
                        }
                    }
                    else if (bytes_read == 0)
                    { // 流结束
                        VP_DEBUG(vp_utils::string_format("Stream ended for [%s].skipping frame...", rtsp_url.c_str()));
                        pipe_ok = false;
                        continue; // 结束读取
                    }

                    total_read += bytes_read;
                }

                // 如果未能成功读取完整帧，记录日志并跳过
                if (total_read < frame_size)
                {
                    VP_DEBUG(vp_utils::string_format("Incomplete frame read for [%s].skipping frame...", rtsp_url.c_str()));

                }

                // 将数据拷贝到 OpenCV Mat
                memcpy(frame.data, frame_buffer.data(), frame_size);
            }

            if (frame.empty())
            {
                VP_WARN(vp_utils::string_format("[%s] reading frame empty, total frame==>%d", node_name.c_str(), frame_index));
                continue;
            }

            // need skip
            if (skip < skip_interval)
            {
                skip++;
                continue;
            }
            skip = 0;

            cv::Mat resize_frame;
            if (this->resize_ratio != 1.0f)
            {
                cv::resize(frame, resize_frame, cv::Size(), resize_ratio, resize_ratio);
            }
            else
            {
                resize_frame = frame.clone(); // clone!;
            }
            // set true size because resize
            video_width = resize_frame.cols;
            video_height = resize_frame.rows;

            this->frame_index++;
            // 插入此处
            if (frame_index % 1000 == 0) {
                VP_INFO(vp_utils::string_format(
                    "Periodic Log - Channel Index: %d, Frame Index: %d, RTSP URL: %s, Use GPU: %s",
                    channel_index, frame_index, rtsp_url.c_str(), use_gpu ? "true" : "false"));
            }
            // create frame meta
            auto out_meta =
                std::make_shared<vp_objects::vp_frame_meta>(resize_frame, this->frame_index, this->channel_index, video_width, video_height, fps);

            if (out_meta != nullptr)
            {
                this->out_queue.push(out_meta);

                // handled hooker activated if need
                if (this->meta_handled_hooker)
                {
                    meta_handled_hooker(node_name, out_queue.size(), out_meta);
                }

                // important! notify consumer of out_queue in case it is waiting.
                this->out_queue_semaphore.signal();
                VP_DEBUG(vp_utils::string_format("[%s] after handling meta, out_queue.size()==>%d", node_name.c_str(), out_queue.size()));
            }
        }
        if (use_gpu)
        {
            close(pipe_fd);
        }

        // send dead flag for dispatch_thread
        this->out_queue.push(nullptr);
        this->out_queue_semaphore.signal();
    }

    // return stream url
    std::string vp_rtsp_ffmpeg_src_node::to_string()
    {
        return rtsp_url;
    }
}