#include <assert.h>
#include <ifaddrs.h>
#include <sys/time.h>
#include <iostream>
#include <memory>
#include <string.h>
#include <unistd.h>
#include <telux/cv2x/Cv2xRadio.hpp>
#include <chrono>
#include "../../common/utils/Utils.hpp"
#include <csignal>
#include <cstdlib>
#include <cstdint>
#include <string>
#include <stdio.h>
#include <stdlib.h>
//#include <ros/package.h>
#include <fstream>
#include <sstream>
#include <vector>
#include <iterator>
#include <thread>
#include <cstring>



using std::array;
using std::cerr;
using std::cout;
using std::endl;
using std::promise;
using std::shared_ptr;
using telux::common::ErrorCode;
using telux::common::Status;
using telux::cv2x::Cv2xFactory;
using telux::cv2x::Cv2xStatus;
using telux::cv2x::Cv2xStatusType;
using telux::cv2x::ICv2xRxSubscription;
using telux::cv2x::TrafficCategory;
using telux::cv2x::TrafficIpType;

static constexpr uint16_t RX_PORT_NUM = 9000u;
static constexpr uint32_t G_BUF_LEN = 3000u;
static constexpr uint32_t NUM_TEST_ITERATIONS = 100u;

static Cv2xStatus gCv2xStatus;
static promise<ErrorCode> gCallbackPromise;
static shared_ptr<ICv2xRxSubscription> gRxSub;
static uint32_t gPacketsReceived = 0u;
static array<char, G_BUF_LEN> gBuf;
volatile sig_atomic_t receivedSignal = 0;   //判断中断信号用

std::ofstream logfile;
std::stringstream logbuffer;


void SaveLog()
{
    std::ofstream logfile("/data/log/example.csv",std::ios::out | std::ios::trunc);
    if (!logfile.is_open()) {
        std::cerr << "无法打开文件" << std::endl;
        exit(0);
    }
    logfile << "time\data" << endl;
    while (!logbuffer.tellp() )
    {
        usleep(1000u);
    }
    while (logbuffer.tellp() > 0)
    {
        logfile<<logbuffer.str();
        logbuffer.str("");
        logbuffer.clear();
        usleep(100000u);
    }
    std::cout << logbuffer.str();

    logfile<<logbuffer.str();
    logbuffer.str("");
    logbuffer.clear();
    logfile.close();
}
// Returns current timestamp
static uint64_t getCurrentTimestamp(void) {
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return (uint64_t)tv.tv_sec * 1000000ull + (uint64_t)tv.tv_usec;
}

// Resets the global callback promise
static inline void resetCallbackPromise(void) {
    gCallbackPromise = promise<ErrorCode>();
}

// Callback function for Cv2xRadioManager->requestCv2xStatus()
static void cv2xStatusCallback(Cv2xStatus status, ErrorCode error){
    if (ErrorCode::SUCCESS == error) {
        gCv2xStatus = status;
    }
    gCallbackPromise.set_value(error);
}

// Callback function for Cv2xRadio->createRxSubscription() and Cv2xRadio->closeRxSubscription()
static void rxSubCallback(shared_ptr<ICv2xRxSubscription> rxSub, ErrorCode error) {
    if (ErrorCode::SUCCESS == error) {
        gRxSub = rxSub;
    }
    gCallbackPromise.set_value(error);
}

// Function for reading from Rx socket
static void sampleRx(void){

    char* endPtr;
    
    int sock = gRxSub->getSock();

    cout << "sampleRx(" << sock << ")" << endl;
    // Attempt to read from socket
    int n = recv(sock, gBuf.data(), gBuf.max_size(), 0);
    long long time2 = getCurrentTimestamp();
    unsigned long long intValue16 = strtoull(gBuf.data(), NULL, 10);
    //................................
    long long num1 = std::strtoll(gBuf.data(), &endPtr, 10);
    

    // 解析第二个数值
    long long num2 = std::strtoll(endPtr + 1, nullptr, 10);

    // 打印结果
    //std::cout << "First number: " << num1 << std::endl;
    //std::cout << "Second number: " << num2 << std::endl;
    //................................
    //logbuffer<< num1  << "\t"<< num2  << endl;
    if (n < 0) {
        cerr << "Error occurred reading from socket[" << sock << "]" << endl;
    }
    else {
        cout << __FUNCTION__ << ": Received " << n << " bytes" << endl;
        ++gPacketsReceived;
        std::cout << "收到缓冲区内的数据" << gBuf.data() << std::endl;
        //std::cout << "缓冲区内转换的数据" << intValue16 << endl;
        std::cout << "打印接收时间戳" << time2 << std::endl;    //打印接收时间戳
        std::cout << "时间差（单位us）" << (time2- intValue16) << std::endl;
        std::cout <<  gPacketsReceived << std::endl;

        logbuffer<< num2  << "\t"<< (time2- intValue16)  << endl;

    }
}
void signalHandler(int signum)
{
	std::cout << "中断开启，恢复状态" <<std::endl;
    if (signum == SIGINT) {
        receivedSignal = 1;
    }
    
    
}
int main(int argc, char *argv[]) {
    std::signal(SIGINT, signalHandler);
    cout << "Running Sample C-V2X RX app" << endl;
    std::thread log_thread(SaveLog);
    std::vector<std::string> groups{"radio", "system", "gps"};
    if (-1 == Utils::setSupplementaryGroups(groups)){
        cout << "Adding supplementary group failed!" << std::endl;
    }

    // Get handle to Cv2xRadioManager
    auto & cv2xFactory = Cv2xFactory::getInstance();
    auto cv2xRadioManager = cv2xFactory.getCv2xRadioManager();

    // Wait for radio manager to complete initialization
    if (not cv2xRadioManager->isReady()) {
        if (cv2xRadioManager->onReady().get()) {
            cout << "C-V2X Radio Manager is ready" << endl;
        }
        else {
            cerr << "C-V2X Radio Manager initialization failed, exiting" << endl;
            return EXIT_FAILURE;
        }
    }

    // Get C-V2X status and make sure Rx is enabled
    assert(Status::SUCCESS == cv2xRadioManager->requestCv2xStatus(cv2xStatusCallback));
    assert(ErrorCode::SUCCESS == gCallbackPromise.get_future().get());

    if (Cv2xStatusType::ACTIVE == gCv2xStatus.rxStatus) {
        cout << "C-V2X RX status is active" << endl;
    }
    else {
        cerr << "C-V2X RX is inactive" << endl;
        return EXIT_FAILURE;
    }

    // Get handle to Cv2xRadio
    auto cv2xRadio = cv2xRadioManager->getCv2xRadio(TrafficCategory::SAFETY_TYPE);

    // Wait for radio to complete initialization
    if (not cv2xRadio->isReady()) {
        if (Status::SUCCESS == cv2xRadio->onReady().get()) {
            cout << "C-V2X Radio is ready" << endl;
        }
        else {
            cerr << "C-V2X Radio initialization failed." << endl;
            return EXIT_FAILURE;
        }
    }
    // Create new Rx subscription
    resetCallbackPromise();
    assert(Status::SUCCESS == cv2xRadio->createRxSubscription(TrafficIpType::TRAFFIC_NON_IP,
                                                              RX_PORT_NUM,
                                                              rxSubCallback));
    assert(ErrorCode::SUCCESS == gCallbackPromise.get_future().get());
    while(!receivedSignal)
    {
    // Read from the RX socket in a loop
        
        sampleRx();
    }
    // Close subscription
    resetCallbackPromise();
    assert(Status::SUCCESS == cv2xRadio->closeRxSubscription(gRxSub,
                                                             rxSubCallback));
    assert(ErrorCode::SUCCESS == gCallbackPromise.get_future().get());

    cout << "Done." << endl;
    logfile.close();
    return EXIT_SUCCESS;
}