USB_UVC_DETECT/MppDecoder/MppManager.cpp

#include "MppManager.h"
#include "../DataManager/DataManager.h"
#include "../DataManager/DataPackage.h"
#include "../ThreadGuardian/ThreadGuardian.h"
#include "../RGAColorTransfer/RgaColorTransfer.h"
#include <thread>
#include "../ImageTest/ImageTest.h"
// ...existing code...

void MppManager::updateThreadStatus(int threadIndex)
{
    std::string threadName = "mppThread_" + std::to_string(threadIndex);
    ThreadGuardian::getInstance().updateThreadHeartbeat(threadName);
}

MppManager::~MppManager()
{

    for (int i = 0; i < m_threads.size(); i++)
    {
        m_bThreadSwitch = false;
        if (m_threads[i].joinable())
        {
            m_threads[i].join();
        }
    }
}

void MppManager::addMppDecode()
{
    m_vThreadSwitch.resize(3);
    m_bThreadSwitch = true;
    m_threads.emplace_back(&MppManager::decodeThread, this, 0);
    int index = m_threads.size() - 1;
    ThreadGuardian::getInstance().registerMppThread(index, std::bind(&MppManager::restartDecodeThread, this, index));

    m_threads.emplace_back(&MppManager::decodeThread, this, 1);
    index = m_threads.size() - 1;
    ThreadGuardian::getInstance().registerMppThread(index, std::bind(&MppManager::restartDecodeThread, this, index));

    m_threads.emplace_back(&MppManager::decodeThread, this, 2);
    index = m_threads.size() - 1;
    ThreadGuardian::getInstance().registerMppThread(index, std::bind(&MppManager::restartDecodeThread, this, index));
}

void MppManager::restartDecodeThread(int threadIndex)
{
    // 先停止当前线程
    // if (m_threads[threadIndex].joinable())
    // {
    //     m_vThreadSwitch[threadIndex] = false;
    //     m_threads[threadIndex].join();
    // }

    // 重新启动线程
    if (threadIndex >= 0 && threadIndex < m_threads.size())
    {
        m_threads[threadIndex] = std::thread(&MppManager::decodeThread, this, threadIndex);
        std::cout << "MPP线程 " << threadIndex << " 已重新启动" << std::endl;
    }
}

void MppManager::decodeThread(int index)
{
    MppDecoder decoder;
    decoder.init();
    RgaColorTransfer colorTransfer;
    unsigned char *rgbData = nullptr;
    int width = 0, height = 0;
    DataManager::getInstance().addDataPipe<DataPackagePtr>("resized" + std::to_string(index));

    // test
    UsbTest::HighResolutionTimer timer;

    while (m_bThreadSwitch)
    {
        DataPackagePtr dataPackage = nullptr;
        // update thread status
        updateThreadStatus(index);
        if (DataManager::getInstance().popData("uvc" + std::to_string(index), dataPackage))
        {
            decoder.decodeJpegToRgb((const char *)dataPackage->pJpegData, dataPackage->nJpegSize, (unsigned char **)&dataPackage->pRGBData, &width, &height);
            if (width != dataPackage->nWidth || height != dataPackage->nHeight)
            {
                std::cerr << "the decode result is different from defined !!" << std::endl;
            }

            std::cout << "Decoded frame " << index << std::endl;

            // color transfer
            resize_image(dataPackage->pRGBData, width, height, dataPackage->pResizeData, dataPackage->nResizeWidth, dataPackage->nResizeHeight);

            // ImageTest::saveImageFromData((unsigned char *)dataPackage->pResizeData, dataPackage->nResizeWidth, dataPackage->nResizeHeight);

            std::string logMessage = "Decoded frame " + std::to_string(index) + " degree : " + std::to_string(dataPackage->dDegree) + " " + UsbTest::GlobalResolutionTimer::getInstance().pressStopWatchString();
            // UsbTest::TimeRecorder::getInstance().recordTime(logMessage);

            // push data to the next pipe
            DataManager::getInstance().pushData("resized" + std::to_string(index), dataPackage);
        }

        // delay 10ms
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
    // 释放资源
    if (rgbData)
    {
        delete[] rgbData;
    }
    DataManager::getInstance().eraseDataPipe("mpp" + std::to_string(index));
}