server1/Server/ThreadRecord/CreateDBThread.cpp

#include "CreateDBThread.h"

#include "GlobalVariable.h"
#include "AudioData.h"
#include "GlobalInfo.h"

#include "CalculateAudio.h"

CreateDBThread::CreateDBThread(RecordThreadInfo_t& threadInfo)
    : BaseRecordThread(threadInfo)
{

}

CreateDBThread::~CreateDBThread()
{

}



/* 设置数据 */
bool CreateDBThread::setData(const AudioSrcData& srcData)
{
    if(srcData.pData == nullptr || srcData.dataSize == 0)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "{} 设置数据失败，srcData为空或dataSize为0", m_logBase);
        return false;
    }
    /* 创建一个新的AudioSrcData对象 */
    AudioSrcData* audioData = new AudioSrcData(srcData);
    if(audioData == nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "{} 创建AudioSrcData对象失败", m_logBase);
        return false;
    }
    /* 判断环形队列是否满 */
    if(m_queueAudioData.isFull())
    {
        /* 出队一个最早的元素 */
        AudioSrcData* oldData = m_queueAudioData.front_pop();
        SPDLOG_LOGGER_WARN(m_logger, "{} 环形队列已满，出队一个元素，时间: {}, 大小: {}", 
            m_logBase, oldData->startTime.toString("yyyy-MM-dd hh:mm:ss.zzz").toStdString(), oldData->dataSize);
        if(oldData != nullptr){
            delete oldData;
            oldData = nullptr;
        }
    }
    /* 入队新的数据 */
    if(!m_queueAudioData.push_noBlock(audioData))
    {
        SPDLOG_LOGGER_ERROR(m_logger, "{} 数据加入环形队列失败", m_logBase);
        delete audioData; // 入队失败，释放内存
        audioData = nullptr;
        return false;
    }

    return true;
}


/* 获取最新的结果，根据时间进行对比，最新的时间比传入的晚，就是有新的数据了 */
bool CreateDBThread::getLatestResult(OneSecondData& resultData)
{
    if(resultData.startTime.isNull())
    {
        // SPDLOG_LOGGER_ERROR(m_logger, "{} 获取最新结果失败，传入的时间戳为空", m_logBase);
        return false;
    }
    std::lock_guard<std::mutex> lock(m_queueResultData->mutex);
    auto lastSecondData = m_queueResultData->back();
    if(lastSecondData == nullptr)
    {
        SPDLOG_LOGGER_ERROR(m_logger, "{} 获取最新结果失败，结果队列为空", m_logBase);
        return false;
    }
    /* 比较时间戳 */
    if(lastSecondData->startTime <= resultData.startTime)
    {
        return false;
    }
    resultData = *lastSecondData;
    return true;
}

/* 获取最新的结果，让整个环形队列相等 */
bool CreateDBThread::getLatestResult(RingQueueManualMutex<OneSecondData*>& resultQueue)
{
    std::lock_guard<std::mutex> lock(m_queueResultData->mutex);
    if(m_queueResultData->isEmpty())
    {
        // SPDLOG_LOGGER_TRACE(m_logger, "{} 获取最新数据，数据队列为空", m_logBase);
        return false;
    }
    /* 目标队列为空，全部拷贝 */
    if(resultQueue.isEmpty())
    {
        SPDLOG_LOGGER_DEBUG(m_logger, "{} 获取最新数据，目标队列为空，拷贝整个结果队列", m_logBase);
        for(int i = 0; i < m_queueResultData->QueueSize(); ++i)
        {
            OneSecondData* data = m_queueResultData->at(i);
            if(data != nullptr)
            {
                OneSecondData* newData = new OneSecondData(*data); // 深拷贝
                resultQueue.push(newData);
            }
        }
        return true;
    }

    /* 队列不为空，查找队列中相等的位置，拷贝后面的数据，不拷贝整个队列，减少开销
     * 这里直接从最新的数据往前找 */
    auto itBack = resultQueue.back();
    int index = m_queueResultData->QueueSize() - 1;
    while(index >= 0)
    {
        OneSecondData* data = m_queueResultData->at(index);
        if(data == nullptr)
        {
            --index;
            continue;
        }
        /* 找到相等的位置 */
        if(data->startTime == itBack->startTime)
        {
            break;
        }
        --index;
    }
    if(index < 0)
    {
        SPDLOG_LOGGER_WARN(m_logger, "{} 获取最新数据失败，未找到相等的时间戳，将清空队列，拷贝全部数据", m_logBase);
        SPDLOG_LOGGER_WARN(m_logger, "队列大小: {}, 目标队列大小: {}", m_queueResultData->QueueSize(), resultQueue.QueueSize());
        /* 清空目标队列，将所有的数据全部拷贝 */
        while(!resultQueue.isEmpty())
        {
            OneSecondData* data = resultQueue.front_pop();
            if(data != nullptr)
            {
                delete data;
                data = nullptr;
            }
        }
        /* 拷贝全部数据 */
        for(int i = 0; i < m_queueResultData->QueueSize(); ++i)
        {
            OneSecondData* data = m_queueResultData->at(i);
            if(data != nullptr)
            {
                OneSecondData* newData = new OneSecondData(*data); // 深拷贝
                resultQueue.push(newData);
            }
        }

        return true;
    }
    else if(index == m_queueResultData->QueueSize() - 1)
    {
        // 已经是最新的数据了，不需要拷贝
        // SPDLOG_LOGGER_DEBUG(m_logger, "{} 获取最新数据，已经是最新的数据了", m_logBase);
        return false;
    }
    /* 拷贝数据 */
    for(; index < m_queueResultData->QueueSize(); ++index)
    {
        OneSecondData* data = m_queueResultData->at(index);
        if(data == nullptr)
        {
            continue;
        }
        OneSecondData* newData = new OneSecondData(*data); // 深拷贝
        auto fornt = resultQueue.push(newData);
        if(fornt != nullptr)
        {
            delete fornt; // 出队的元素是之前的结果，释放内存
            fornt = nullptr;
        }
    }

    return true;
}


/* 计算音量和反相的线程函数 */
void CreateDBThread::task()
{
    SPDLOG_LOGGER_INFO(m_logger, "➢ {} 开始计算音量线程", m_logBase);
    /* 初始化一些数据 */
    if(!initData())
    {
        SPDLOG_LOGGER_ERROR(m_logger, "{} 初始化数据失败", m_logBase);
        return;
    }

    while(m_isRunning)
    {
        /* 取出一个数据，没有数据则阻塞住 */
        AudioSrcData* audioData = m_queueAudioData.front_pop();
        // SPDLOG_LOGGER_DEBUG(m_logger, "{} 从环形队列中取出数据，队列大小: {}", m_logBase, m_queueAudioData.QueueSize());

        if(audioData == nullptr)
        {
            SPDLOG_LOGGER_WARN(m_logger, "{} 从环形队列中取出数据失败，可能是队列为空", m_logBase);
            std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 等待一段时间再尝试
            continue;
        }
        /* 计算音量和反相，结果存储在m_result */
        if(!CreateDBPhase(audioData))
        {
            SPDLOG_LOGGER_ERROR(m_logger, "{} 计算音量和反相失败", m_logBase);
        }
        /* 将结果放入结果队列 */
        if(m_result != nullptr)
        {
            std::lock_guard<std::mutex> lock(m_queueResultData->mutex);
            auto result = m_queueResultData->push(m_result);
            if(result != nullptr)
            {
                // SPDLOG_LOGGER_DEBUG(m_logger, "{} 队列已满，出队一个元素", m_logBase);
                delete result; // 出队的元素是之前的结果，释放内存
                result = nullptr;
            }
            m_result = nullptr; // 清空结果指针，等待下一次使用
        }

        delete audioData; // 释放内存
        audioData = nullptr;
        
        // SPDLOG_LOGGER_DEBUG(m_logger, "{} 创建音量的线程处理数据完成，队列大小: {}", 
        //     m_logBase, m_queueResultData->QueueSize());
    }
    /* 清理数据 */
    clearData();
    SPDLOG_LOGGER_WARN(m_logger, "➢ {} 计算音量线程结束 ", m_logBase);
    
}



/* 初始化一些数据 */
bool CreateDBThread::initData()
{
    int queueSize = GInfo.queueElementCount();
    /* 初始化一些数据 */
    m_queueAudioData.clearQueue();
    m_queueAudioData.setQueueCapacity(queueSize);

    m_sampleRate = GInfo.sampleRate();          /* 采样率 */
    m_numChannels = GInfo.numChannels();        /* 声道数 */
    m_bitsPerSample = GInfo.bitsPerSample();    /* 每个采样点的位数 */

    /* 一秒钟数据大小，单位：字节 */
    m_oneSecondSize = m_sampleRate * m_numChannels * (m_bitsPerSample / 8);

    /* 计算音量和反相的环形队列元素数目，默认是180个，即180秒 */
    m_queueAudioDataCapacity = queueSize; 
    m_queueResultData = new RingQueueManualMutex<OneSecondData*>(m_queueAudioDataCapacity);
    m_queueResultData->mutex.lock();
    m_queueResultData->setDefaultValue(nullptr);
    m_queueResultData->mutex.unlock();

    return true;
}

/* 清理数据 */
void CreateDBThread::clearData()
{
    /* 清理环形队列 */
    while(!m_queueAudioData.isEmpty())
    {
        AudioSrcData* data = m_queueAudioData.front_pop();
        if(data != nullptr)
        {
            delete data;
            data = nullptr;
        }
    }
    // if(m_remainData != nullptr)
    // {
    //     delete m_remainData;
    //     m_remainData = nullptr;
    // }
}

/* 计算音量和反相 */
bool CreateDBThread::CreateDBPhase(AudioSrcData* audioData)
{
    short* pWaveVu = (short*)(audioData->pData);

    /* 一秒钟平分30个音量值，每个音量值占有的长度 */
    const int oneDBSize = m_oneDBLengthOfSrcData;;

    StAudioNum audioInfo;
    audioInfo.roadInfo = m_threadInfo.cardRoadInfo; // 设置通道信息
    audioInfo.nTotal = m_singleDataLength;
    int iCurPos = 0;
    /* 存储结果 */
    m_result = new OneSecondData();
    /* 计算音量和反相的计算类 */
    CAudio2ChanCorrelator audioCor;
    /* 音量值计算，好多个字节计算出一个最大音量值 */
    for(int i = 0; i < VOLUME_INFO_NUM; ++i)
    {
        // oneDBSize / 2 除以2是因为左右声道同时计算
        // 采样点最大值
        short sMaxA, sMaxB, sRMSA, sRMSB;
        audioCor.CorrelateChunks(pWaveVu + iCurPos, (pWaveVu + iCurPos + 1), oneDBSize / 2, &sMaxA, &sMaxB, &sRMSA, &sRMSB, audioInfo);

        iCurPos += oneDBSize;
        m_result->aryLeftDB[i] = calculateDB(sMaxA);
        m_result->aryRightDB[i] = calculateDB(sMaxB);

        // 获取反相值，-100 到 100；反相是左右声道比对得到的值，所以只有一个
        int iReversed = audioCor.GetCorrelationLevel();
        if(iReversed > REVERSED_MAX_VALUE || iReversed < REVERSED_MIN_VALUE)
        {
            iReversed = 0;
        }

        // 和反相阀值比较，来判定是否为反相
        float dReversed = iReversed / 100.00;
        m_result->aryPhase[i] = dReversed;
    }

    /* 设置时间戳 */
    m_result->startTime = audioData->startTime;
    m_result->endTime = audioData->endTime;
    return true;
}