server1/Server/ThreadRecord/CreateDBThread.cpp

#include "CreateDBThread.h"

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

#include "CalculateAudio.h"
#include "spdlog.h"

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

}

CreateDBThread::~CreateDBThread()
{

}

/* 停止线程 */
void CreateDBThread::thread_stop()
{
    if(m_isRunning.load())
    {
        m_isRunning.store(false);
    }
    /* 环形阻塞的线程 */
    m_queueRealTimeData.exit();
}

void CreateDBThread::thread_stop_block()
{
    thread_stop();
    while(true)
    {
        if(m_isStoped.load())
        {
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}



/* 设置数据 */
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_TRACE(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::setRealTimeData(const AudioSrcData& srcData)
{
    auto oldData = m_queueRealTimeData.push_pop(new AudioSrcData(srcData));
    if(oldData != nullptr)
    {
        /* 这里一般不会满，会被实时消耗，如果满了，说明出现问题了 */
        SPDLOG_LOGGER_WARN(m_logger, "{} 实时数据环形队列已满，出队一个元素，时间: {}, 大小: {}", 
            m_logBase, oldData->startTime.toString("yyyy-MM-dd hh:mm:ss.zzz").toStdString(), oldData->dataSize);
        delete oldData;
        oldData = nullptr;
    }

    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;
}


/**
 * @brief 获取最新的音量包结果
 * 
 * @param listData 
 * @param count 需要拷贝的数目，如果暂时没有这么多数据，则返回false
 * @return true 获取成功
 * @return false 无数据或者数据不足
 */
bool CreateDBThread::getLatestRealTimeResult(std::list<OneDBData>& listData, const int getCount)
{
    std::lock_guard<std::mutex> lock(m_mutexRealTimeResult);
    if(getCount <= 0)
    {
        return false;
    }

    if(m_listRealTimeResult.empty() || static_cast<int>(m_listRealTimeResult.size()) < getCount)
    {
        // SPDLOG_LOGGER_TRACE(m_logger, "{} 最新实时音量数据为空", m_logBase);
        return false;
    }

    auto it = m_listRealTimeResult.end();
    if(listData.empty())
    {
        SPDLOG_LOGGER_TRACE(m_logger, "{} 获取最新实时音量数据，列表为空，拷贝最新的{}个数据", m_logBase, getCount);
        for(int i = 0; i < getCount; ++i)
        {
            it--;
        }
    }else 
    {
        /* 先向前移动getCount个数据，判断新数据是否满足 */
        auto oldData = listData.back();
        for(int i = 0; i < getCount; ++i)
        {
            it--;
        }
        if(it->startTime <= oldData.startTime)
        {
            return false;
        }
        /* 继续向前查找已发送的时间点 */
        /* 找到需要开始拷贝的元素位置 */
        while(it != m_listRealTimeResult.begin())
        {
            if(it->startTime <= oldData.startTime)
            {
                break; // 找到需要开始拷贝的元素位置
            }
            it--;
        }
        /* 上面的循环无法判断begin()，这里单独判断一下 */
        if(it == m_listRealTimeResult.begin())
        {
            if(it->startTime <= oldData.startTime)
            {
                /* begin()是最新的，指向下一个可以拷贝的数据 */
                it++;
            }else 
            {
                /* 列表中所有的数据都比已发送的数据新，只拷贝最新的getCount个数据 */
                it = m_listRealTimeResult.end();
                for(int i = 0; i < getCount; ++i)
                {
                    it--;
                }
            }
        }else 
        {
            /* 移动到下一个可以拷贝的新数据位置 */
            it++;
        }
        
    }
    
    /* 拷贝数据 */
    listData.clear();
    for(int i = 0; it != m_listRealTimeResult.end() && i < getCount; ++it, ++i)
    {
        listData.push_back(*it);
    }

    // SPDLOG_LOGGER_TRACE(m_logger, "{} 获取最新实时音量数据成功，数量: {}", m_logBase, listData.size());

    return true;
}

/* 获取一个最新音量值 */
bool CreateDBThread::getLatestRealTimeResult(OneDBData& resultData)
{
    std::lock_guard<std::mutex> lock(m_mutexRealTimeResult);
    if(m_listRealTimeResult.empty())
    {
        // SPDLOG_LOGGER_TRACE(m_logger, "{} 最新实时音量数据为空", m_logBase);
        return false;
    }
    resultData = m_listRealTimeResult.front();
    m_listRealTimeResult.pop_front();
    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)
    {
        
        /*--------------------------------------------------------------
        * 计算实时音量值(这个数据量很小，大概只有33ms，所以阻塞在这里)
        *--------------------------------------------------------------*/
        /* 判断实时数据列表是否有数据，没有数据则阻塞，直到有数据 */
        auto rtData = m_queueRealTimeData.front();
        if(rtData == nullptr)
        {
            SPDLOG_LOGGER_DEBUG(m_logger, "{} 实时数据队列为空，等待数据", m_logBase);
            continue;
        }
        // std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
        /* 计算实时音量值 */
        calculateRealTimeVolume();

        // std::chrono::steady_clock::time_point rtEndTime = std::chrono::steady_clock::now();
        // std::chrono::microseconds rtDuration = std::chrono::duration_cast<std::chrono::microseconds>(rtEndTime - startTime);
        // SPDLOG_LOGGER_DEBUG(m_logger, "{} 计算实时音量包完成，耗时: {} 微秒", m_logBase, rtDuration.count());
        /*--------------------------------------------------------------
        * 计算常规音量值，1秒计算一次
        *--------------------------------------------------------------*/
        calculateVolumeData();
        
        // std::chrono::steady_clock::time_point endTime = std::chrono::steady_clock::now();
        // std::chrono::microseconds duration = std::chrono::duration_cast<std::chrono::microseconds>(endTime - rtEndTime);
        // SPDLOG_LOGGER_DEBUG(m_logger, "{} 计算1秒的音量包完成，耗时: {} 微秒", m_logBase, duration.count());
    }
    /* 清理数据 */
    clearData();
    SPDLOG_LOGGER_WARN(m_logger, "➢ {} 计算音量线程结束 ", m_logBase);
    
}



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

    m_queueRealTimeData.setQueueCapacity(queueSize);

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

    /* 一秒钟数据大小，单位：字节 */
    m_oneSecondSize = m_sampleRate * m_numChannels * (m_bitsPerSample / 8);
    /* 计算每个音量值需要的数据大小，单位数据单位: short，一个通道的大小 */
    m_oneDBLengthOfSrcData = m_sampleRate / VOLUME_INFO_NUM;
    m_singleDataLength = m_oneSecondSize / m_numChannels;

    /* 计算音量的环形队列元素数目，默认是180个，即180秒 */
    m_queueResultData.mutex.lock();
    m_queueResultData.setDefaultValue(nullptr);
    m_queueResultData.setQueueCapacity(queueSize);
    m_queueResultData.mutex.unlock();

    m_numMaxResultData = 60;

    return true;
}

/* 清理数据 */
void CreateDBThread::clearData()
{
    /* 清理环形队列 */
    while(!m_queueAudioData.isEmpty())
    {
        AudioSrcData* data = m_queueAudioData.front_pop();
        if(data != nullptr)
        {
            delete data;
            data = nullptr;
        }
    }
    
    std::lock_guard<std::mutex> lock(m_queueResultData.mutex);
    while(!m_queueResultData.isEmpty())
    {
        auto pData = m_queueResultData.front_pop();
        if(pData != nullptr)
        {
            delete pData;
            pData = nullptr;
        }
    }
    if(m_result != nullptr)
    {
        delete m_result;
        m_result = nullptr;
    }
    /* 清理实时数据 */
    while(!m_queueRealTimeData.isEmpty())
    {
        AudioSrcData* data = m_queueRealTimeData.front_pop();
        if(data != nullptr)
        {
            delete data;
            data = nullptr;
        }
    }
    m_listRealTimeResult.clear();

}

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

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

    StAudioNum audioInfo;
    audioInfo.cardRoadInfo = 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)
    {
        // 采样点最大值
        short sMaxA, sMaxB, sRMSA, sRMSB;
        audioCor.CorrelateChunks(pWaveVu + iCurPos, (pWaveVu + iCurPos + 1), oneDBSize, &sMaxA, &sMaxB, &sRMSA, &sRMSB, audioInfo);
        
        /* 这里乘2是要增加两个通道的数据大小 */
        iCurPos += (oneDBSize * 2);
        m_result->aryLeftDB[i] = calculateDB(sMaxA);
        m_result->aryRightDB[i] = calculateDB(sMaxB);
        // fmt::print("音量计算: sMaxA: {}, sMaxB: {}, sRMSA: {}, sRMSB: {}, LDB:{}, RDB:{}\n", sMaxA, sMaxB, sRMSA, sRMSB, 
        //     m_result->aryLeftDB[i], m_result->aryRightDB[i]);

        // 获取反相值，-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;
}


/* 计算音量数据 */
bool CreateDBThread::calculateVolumeData()
{
    /* 判断是否够一秒的数据 */
    if(m_queueAudioData.isEmpty())
    {
        return false;;
    }

    auto audioData = m_queueAudioData.front_pop();

    if(audioData == nullptr)
    {
        SPDLOG_LOGGER_WARN(m_logger, "{} 从环形队列中取出数据失败，可能是队列为空", m_logBase);
        // std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 等待一段时间再尝试
        return false;;
    }
    /* 计算音量和反相，结果存储在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;

    return true;
}



/* 计算实时音量值 */
bool CreateDBThread::calculateRealTimeVolume()
{
    /* 一秒钟平分30个音量值，每个音量值占有的长度 */
    const int oneDBSize = m_oneDBLengthOfSrcData;

    while(!m_queueRealTimeData.isEmpty())
    {
        /* 取出一个数据，没有数据则阻塞住 */
        AudioSrcData* audioData = m_queueRealTimeData.front_pop_noBlock();
        if(audioData == nullptr)
        {
            continue;
        }
        
        /* 计算音量和反相的计算类 */
        CAudio2ChanCorrelator audioCor;
        StAudioNum audioInfo;
        audioInfo.cardRoadInfo = m_threadInfo.cardRoadInfo; // 设置通道信息
        audioInfo.nTotal = m_singleDataLength;

        short* pData = reinterpret_cast<short*>(audioData->pData);
        // 采样点最大值
        short sMaxA, sMaxB, sRMSA, sRMSB;
        audioCor.CorrelateChunks(pData, pData + 1, oneDBSize, &sMaxA, &sMaxB, &sRMSA, &sRMSB, audioInfo);
        
        /* 这里乘2是要增加两个通道的数据大小 */
        int leftDB = calculateDB(sMaxA);
        int rightDB = calculateDB(sMaxB);
        
        // fmt::print("音量计算: sMaxA: {}, sMaxB: {}, sRMSA: {}, sRMSB: {}, LDB:{}, RDB:{}\n", sMaxA, sMaxB, sRMSA, sRMSB, 
        //     m_result->aryLeftDB[i], m_result->aryRightDB[i]);

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

        OneDBData oneDBData;
        oneDBData.leftDB = leftDB;
        oneDBData.rightDB = rightDB;
        oneDBData.phase = dReversed;
        oneDBData.startTime = audioData->startTime;
        oneDBData.endTime = audioData->endTime;
        /* 保存结果 */
        std::lock_guard<std::mutex> lock(m_mutexRealTimeResult);
        m_listRealTimeResult.push_back(oneDBData);
        
        while(static_cast<int>(m_listRealTimeResult.size()) > m_numMaxResultData)
        {
            /* 超过最大数量，删除最早的一个 */
            m_listRealTimeResult.pop_front();
        }

        /* 删除元素内存 */
        delete audioData;
        audioData = nullptr;
    }

    return true;
}