Recherche avancée

Sur d’autres sites (11428)

• Decoding and playing audio with ffmpeg and XAudio2 - frequency ratio wrong

  9 mars, par Brent de Carteret

  I'm using ffmpeg to decode audio and output it using the XAudio2 API, it works and plays synced with the video output using the pts. But it's high pitched (i.e. sounds like chipmunks).

  


  Setting breakpoints I can see it has set the correct sample rate from the audio codec in CreateSourceVoice. I'm stumped.

  


  Any help would be much appreciated.

  


  CDVDAUDIO.cpp

  


  #include "DVDAudioDevice.h"
   
HANDLE m_hBufferEndEvent;

CDVDAudio::CDVDAudio()
{
    m_pXAudio2 = NULL;
    m_pMasteringVoice = NULL;
    m_pSourceVoice = NULL;
    m_pWfx  = NULL;
    m_VoiceCallback = NULL;    
    m_hBufferEndEvent = CreateEvent(NULL, false, false, "Buffer end event");
}
    
CDVDAudio::~CDVDAudio()
{
    m_pXAudio2 = NULL;
    m_pMasteringVoice = NULL;
    m_pSourceVoice = NULL;
    m_pWfx  = NULL;
    m_VoiceCallback = NULL;
    CloseHandle(m_hBufferEndEvent);
    m_hBufferEndEvent = NULL;
}
    
bool CDVDAudio::Create(int iChannels, int iBitrate, int iBitsPerSample, bool bPasstrough)
{
    CoInitializeEx(NULL, COINIT_MULTITHREADED);
    HRESULT hr = XAudio2Create( &m_pXAudio2, 0, XAUDIO2_DEFAULT_PROCESSOR);
    
    if (SUCCEEDED(hr))
    {
        m_pXAudio2->CreateMasteringVoice( &m_pMasteringVoice );
    }
    
    // Create source voice
    WAVEFORMATEXTENSIBLE wfx;
    memset(&wfx, 0, sizeof(WAVEFORMATEXTENSIBLE));
    
    wfx.Format.wFormatTag           = WAVE_FORMAT_PCM;
    wfx.Format.nSamplesPerSec       = iBitrate;//pFFMpegData->pAudioCodecCtx->sample_rate;//48000 by default
    wfx.Format.nChannels            = iChannels;//pFFMpegData->pAudioCodecCtx->channels;
    wfx.Format.wBitsPerSample       = 16;
    wfx.Format.nBlockAlign          = wfx.Format.nChannels*16/8;
    wfx.Format.nAvgBytesPerSec      = wfx.Format.nSamplesPerSec * wfx.Format.nBlockAlign;
    wfx.Format.cbSize               = sizeof(WAVEFORMATEXTENSIBLE)-sizeof(WAVEFORMATEX);
    wfx.Samples.wValidBitsPerSample = wfx.Format.wBitsPerSample;
    
    if(wfx.Format.nChannels == 1)
    {
        wfx.dwChannelMask = SPEAKER_MONO;
    }
    else if(wfx.Format.nChannels == 2)
    {
        wfx.dwChannelMask = SPEAKER_STEREO;
    }
    else if(wfx.Format.nChannels == 5)
    {
        wfx.dwChannelMask = SPEAKER_5POINT1;
    }
    
    wfx.SubFormat = KSDATAFORMAT_SUBTYPE_PCM;
    
    unsigned int flags = 0;//XAUDIO2_VOICE_NOSRC;// | XAUDIO2_VOICE_NOPITCH;
        
    //Source voice
    m_VoiceCallback = new StreamingVoiceCallback(this);
    hr = m_pXAudio2->CreateSourceVoice(&m_pSourceVoice,(WAVEFORMATEX*)&wfx, 0 , 1.0f, m_VoiceCallback);
        
    if (!SUCCEEDED(hr))
        return false;
    
    // Start sound
    hr = m_pSourceVoice->Start(0);
    
    if(!SUCCEEDED(hr))
        return false;
    
    return true;
}
    
DWORD CDVDAudio::AddPackets(unsigned char* data, DWORD len)
{  
    memset(&m_SoundBuffer,0,sizeof(XAUDIO2_BUFFER));
    m_SoundBuffer.AudioBytes = len;
    m_SoundBuffer.pAudioData = data;
    m_SoundBuffer.pContext = NULL;//(VOID*)data;
    XAUDIO2_VOICE_STATE state;
    
    while (m_pSourceVoice->GetState( &state ), state.BuffersQueued > 60)
    {
        WaitForSingleObject( m_hBufferEndEvent, INFINITE );
    }
    
    m_pSourceVoice->SubmitSourceBuffer( &m_SoundBuffer );
    return 0;
}
    
void CDVDAudio::Destroy()
{
    m_pMasteringVoice->DestroyVoice();
    m_pXAudio2->Release();
    m_pSourceVoice->DestroyVoice();
    delete m_VoiceCallback;
    m_VoiceCallback = NULL;
}


  


  CDVDAUdioCodecFFmpeg.cpp

  


  #include "DVDAudioCodecFFmpeg.h"
#include "Log.h"
    
CDVDAudioCodecFFmpeg::CDVDAudioCodecFFmpeg() : CDVDAudioCodec()
{
    m_iBufferSize = 0;
    m_pCodecContext = NULL;
    m_bOpenedCodec = false;
}
    
CDVDAudioCodecFFmpeg::~CDVDAudioCodecFFmpeg()
{
    Dispose();
}
    
bool CDVDAudioCodecFFmpeg::Open(AVCodecID codecID, int iChannels, int iSampleRate)
{
    AVCodec* pCodec;
    m_bOpenedCodec = false;
    av_register_all();
    pCodec = avcodec_find_decoder(codecID);
    m_pCodecContext = avcodec_alloc_context3(pCodec);//avcodec_alloc_context();
    avcodec_get_context_defaults3(m_pCodecContext, pCodec);
    
    if (!pCodec)
    {
        CLog::Log(LOGERROR, "CDVDAudioCodecFFmpeg::Open() Unable to find codec");
        return false;
    }
    
    m_pCodecContext->debug_mv = 0;
    m_pCodecContext->debug = 0;
    m_pCodecContext->workaround_bugs = 1;
    
    if (pCodec->capabilities & CODEC_CAP_TRUNCATED)
        m_pCodecContext->flags |= CODEC_FLAG_TRUNCATED;
    
    m_pCodecContext->channels = iChannels;
    m_pCodecContext->sample_rate = iSampleRate;
    //m_pCodecContext->bits_per_sample = 24;
     
    /* //FIXME BRENT
        if( ExtraData && ExtraSize > 0 )
        {
            m_pCodecContext->extradata_size = ExtraSize;
            m_pCodecContext->extradata = m_dllAvCodec.av_mallocz(ExtraSize + FF_INPUT_BUFFER_PADDING_SIZE);
            memcpy(m_pCodecContext->extradata, ExtraData, ExtraSize);
        }
    */
    
    // set acceleration
    //m_pCodecContext->dsp_mask = FF_MM_FORCE | FF_MM_MMX | FF_MM_MMXEXT | FF_MM_SSE; //BRENT
    
    if (avcodec_open2(m_pCodecContext, pCodec, NULL) < 0)
    {
        CLog::Log(LOGERROR, "CDVDAudioCodecFFmpeg::Open() Unable to open codec");
        Dispose();
        return false;
    }
    
    m_bOpenedCodec = true;
    return true;
}
    
void CDVDAudioCodecFFmpeg::Dispose()
{
    if (m_pCodecContext)
    {
        if (m_bOpenedCodec)
            avcodec_close(m_pCodecContext);
        m_bOpenedCodec = false;
        av_free(m_pCodecContext);
        m_pCodecContext = NULL;
    }
    m_iBufferSize = 0;
}

int CDVDAudioCodecFFmpeg::Decode(BYTE* pData, int iSize)
{
    int iBytesUsed;
    if (!m_pCodecContext) return -1;
    
    //Copy into a FFMpeg AVPAcket again
    AVPacket packet;
    av_init_packet(&packet);
    
    packet.data=pData;
    packet.size=iSize;
    
    int iOutputSize = AVCODEC_MAX_AUDIO_FRAME_SIZE; //BRENT
    
    iBytesUsed = avcodec_decode_audio3(m_pCodecContext, (int16_t *)m_buffer, &iOutputSize/*m_iBufferSize*/, &packet);

    m_iBufferSize = iOutputSize;//BRENT

    return iBytesUsed;
}

int CDVDAudioCodecFFmpeg::GetData(BYTE** dst)
{
    *dst = m_buffer;
    return m_iBufferSize;
}

void CDVDAudioCodecFFmpeg::Reset()
{
    if (m_pCodecContext)
        avcodec_flush_buffers(m_pCodecContext);
}

int CDVDAudioCodecFFmpeg::GetChannels()
{
    if (m_pCodecContext)
        return m_pCodecContext->channels;
    return 0;
}

int CDVDAudioCodecFFmpeg::GetSampleRate()
{
    if (m_pCodecContext)
        return m_pCodecContext->sample_rate;
    return 0;
}
    
int CDVDAudioCodecFFmpeg::GetBitsPerSample()
{
    if (m_pCodecContext)
        return 16;
    return 0;
}


  


  CDVDPlayerAudio.cpp

  


  #include "DVDPlayerAudio.h"
#include "DVDDemuxUtils.h"
#include "Log.h"
    
#include 
#include "DVDAudioCodecFFmpeg.h" //FIXME Move to a codec factory!!
    
CDVDPlayerAudio::CDVDPlayerAudio(CDVDClock* pClock) : CThread()
{
    m_pClock = pClock;
    m_pAudioCodec = NULL;
    m_bInitializedOutputDevice = false;
    m_iSourceChannels = 0;
    m_audioClock = 0;
    
    //  m_currentPTSItem.pts = DVD_NOPTS_VALUE;
    //  m_currentPTSItem.timestamp = 0;
    
    SetSpeed(DVD_PLAYSPEED_NORMAL);
      
    InitializeCriticalSection(&m_critCodecSection);
    m_messageQueue.SetMaxDataSize(10 * 16 * 1024);
    //  g_dvdPerformanceCounter.EnableAudioQueue(&m_packetQueue);
}

CDVDPlayerAudio::~CDVDPlayerAudio()
{
    //  g_dvdPerformanceCounter.DisableAudioQueue();

    // close the stream, and don't wait for the audio to be finished
    CloseStream(true);
    DeleteCriticalSection(&m_critCodecSection);
}

bool CDVDPlayerAudio::OpenStream( CDemuxStreamAudio *pDemuxStream )
{
    // should always be NULL!!!!, it will probably crash anyway when deleting m_pAudioCodec here.
    if (m_pAudioCodec)
    {
        CLog::Log(LOGFATAL, "CDVDPlayerAudio::OpenStream() m_pAudioCodec != NULL");
        return false;
    }
    
    AVCodecID codecID = pDemuxStream->codec;
    
    CLog::Log(LOGNOTICE, "Finding audio codec for: %i", codecID);
    //m_pAudioCodec = CDVDFactoryCodec::CreateAudioCodec( pDemuxStream ); 
    m_pAudioCodec = new CDVDAudioCodecFFmpeg; //FIXME BRENT Codec Factory needed!
    
    if (!m_pAudioCodec->Open(pDemuxStream->codec, pDemuxStream->iChannels, pDemuxStream->iSampleRate))
    {
        m_pAudioCodec->Dispose();
        delete m_pAudioCodec;
        m_pAudioCodec = NULL;
        return false;
    }
    
    if ( !m_pAudioCodec )
    {
        CLog::Log(LOGERROR, "Unsupported audio codec");
        return false;
    }
    
    m_codec = pDemuxStream->codec;
    m_iSourceChannels = pDemuxStream->iChannels;
    m_messageQueue.Init();
    
    CLog::Log(LOGNOTICE, "Creating audio thread");
    Create();
    
    return true;
}

void CDVDPlayerAudio::CloseStream(bool bWaitForBuffers)
{
    // wait until buffers are empty
    if (bWaitForBuffers)
        m_messageQueue.WaitUntilEmpty();
    
    // send abort message to the audio queue
    m_messageQueue.Abort();
    
    CLog::Log(LOGNOTICE, "waiting for audio thread to exit");
    
    // shut down the audio_decode thread and wait for it
    StopThread(); // will set this->m_bStop to true
    this->WaitForThreadExit(INFINITE);
    
    // uninit queue
    m_messageQueue.End();
    
    CLog::Log(LOGNOTICE, "Deleting audio codec");
    if (m_pAudioCodec)
    {
        m_pAudioCodec->Dispose();
        delete m_pAudioCodec;
        m_pAudioCodec = NULL;
    }
    
    // flush any remaining pts values
    //FlushPTSQueue(); //FIXME BRENT
}

void CDVDPlayerAudio::OnStartup()
{
    CThread::SetName("CDVDPlayerAudio");
    pAudioPacket = NULL;
    m_audioClock = 0;
    audio_pkt_data = NULL;
    audio_pkt_size = 0;
  
    //  g_dvdPerformanceCounter.EnableAudioDecodePerformance(ThreadHandle());
}

void CDVDPlayerAudio::Process()
{
    CLog::Log(LOGNOTICE, "running thread: CDVDPlayerAudio::Process()");

    int result;
    
    // silence data
    BYTE silence[1024];
    memset(silence, 0, 1024);
    
    DVDAudioFrame audioframe;
    
    __int64 iClockDiff=0;
    while (!m_bStop)
    {
        //Don't let anybody mess with our global variables
        EnterCriticalSection(&m_critCodecSection);
        result = DecodeFrame(audioframe, m_speed != DVD_PLAYSPEED_NORMAL); // blocks if no audio is available, but leaves critical section before doing so
        LeaveCriticalSection(&m_critCodecSection);
    
        if ( result & DECODE_FLAG_ERROR ) 
        {      
            CLog::Log(LOGERROR, "CDVDPlayerAudio::Process - Decode Error. Skipping audio frame");
            continue;
        }
    
        if ( result & DECODE_FLAG_ABORT )
        {
            CLog::Log(LOGDEBUG, "CDVDPlayerAudio::Process - Abort received, exiting thread");
            break;
        }
    
        if ( result & DECODE_FLAG_DROP ) //FIXME BRENT
        {
            /*  //frame should be dropped. Don't let audio move ahead of the current time thou
                //we need to be able to start playing at any time
                //when playing backwards, we try to keep as small buffers as possible
    
                // set the time at this delay
                AddPTSQueue(audioframe.pts, m_dvdAudio.GetDelay());
            */
            if (m_speed > 0)
            {
                __int64 timestamp = m_pClock->GetAbsoluteClock() + (audioframe.duration * DVD_PLAYSPEED_NORMAL) / m_speed;
                while ( !m_bStop && timestamp > m_pClock->GetAbsoluteClock() )
                    Sleep(1);
            }
            continue;
        }
    
        if ( audioframe.size > 0 ) 
        {
            // we have successfully decoded an audio frame, open up the audio device if not already done
            if (!m_bInitializedOutputDevice)
            {
                m_bInitializedOutputDevice = InitializeOutputDevice();
            }
    
            //Add any packets play
            m_dvdAudio.AddPackets(audioframe.data, audioframe.size);
    
            // store the delay for this pts value so we can calculate the current playing
            //AddPTSQueue(audioframe.pts, m_dvdAudio.GetDelay() - audioframe.duration);//BRENT
        }
    
        // if we where asked to resync on this packet, do so here
        if ( result & DECODE_FLAG_RESYNC )
        {
            CLog::Log(LOGDEBUG, "CDVDPlayerAudio::Process - Resync recieved.");
            //while (!m_bStop && (unsigned int)m_dvdAudio.GetDelay() > audioframe.duration ) Sleep(5); //BRENT
            m_pClock->Discontinuity(CLOCK_DISC_NORMAL, audioframe.pts);
        }
    
        #ifdef USEOLDSYNC
        //Clock should be calculated after packets have been added as m_audioClock points to the 
        //time after they have been played
    
        const __int64 iCurrDiff = (m_audioClock - m_dvdAudio.GetDelay()) - m_pClock->GetClock();
        const __int64 iAvDiff = (iClockDiff + iCurrDiff)/2;
    
        //Check for discontinuity in the stream, use a moving average to
        //eliminate highfreq fluctuations of large packet sizes
        if ( ABS(iAvDiff) > 5000 ) // sync clock if average diff is bigger than 5 msec 
        {
            //Wait until only the new audio frame which triggered the discontinuity is left
            //then set disc state
            while (!m_bStop && (unsigned int)m_dvdAudio.GetBytesInBuffer() > audioframe.size )
                Sleep(5);
    
            m_pClock->Discontinuity(CLOCK_DISC_NORMAL, m_audioClock - m_dvdAudio.GetDelay());
            CLog::("CDVDPlayer:: Detected Audio Discontinuity, syncing clock. diff was: %I64d, %I64d, av: %I64d", iClockDiff, iCurrDiff, iAvDiff);
            iClockDiff = 0;
        }
        else
        {
            //Do gradual adjustments (not working yet)
            //m_pClock->AdjustSpeedToMatch(iClock + iAvDiff);
            iClockDiff = iCurrDiff;
        }
        #endif
    }
}

void CDVDPlayerAudio::OnExit()
{
    //g_dvdPerformanceCounter.DisableAudioDecodePerformance();
  
    // destroy audio device
    CLog::Log(LOGNOTICE, "Closing audio device");
    m_dvdAudio.Destroy();
    m_bInitializedOutputDevice = false;

    CLog::Log(LOGNOTICE, "thread end: CDVDPlayerAudio::OnExit()");
}

// decode one audio frame and returns its uncompressed size
int CDVDPlayerAudio::DecodeFrame(DVDAudioFrame &audioframe, bool bDropPacket)
{
    CDVDDemux::DemuxPacket* pPacket = pAudioPacket;
    int n=48000*2*16/8, len;
    
    //Store amount left at this point, and what last pts was
    unsigned __int64 first_pkt_pts = 0;
    int first_pkt_size = 0; 
    int first_pkt_used = 0;
    int result = 0;
    
    // make sure the sent frame is clean
    memset(&audioframe, 0, sizeof(DVDAudioFrame));
    
    if (pPacket)
    {
        first_pkt_pts = pPacket->pts;
        first_pkt_size = pPacket->iSize;
        first_pkt_used = first_pkt_size - audio_pkt_size;
    }
     
    for (;;)
    {
        /* NOTE: the audio packet can contain several frames */
        while (audio_pkt_size > 0)
        {
            len = m_pAudioCodec->Decode(audio_pkt_data, audio_pkt_size);
            if (len < 0)
            {
                /* if error, we skip the frame */
                audio_pkt_size=0;
                m_pAudioCodec->Reset();
                break;
            }
    
            // fix for fucked up decoders //FIXME BRENT
            if( len > audio_pkt_size )
            {        
                CLog::Log(LOGERROR, "CDVDPlayerAudio:DecodeFrame - Codec tried to consume more data than available. Potential memory corruption");        
                audio_pkt_size=0;
                m_pAudioCodec->Reset();
                assert(0);
            }
    
            // get decoded data and the size of it
            audioframe.size = m_pAudioCodec->GetData(&audioframe.data);
            audio_pkt_data += len;
            audio_pkt_size -= len;
    
            if (audioframe.size <= 0)
                continue;
    
            audioframe.pts = m_audioClock;
    
            // compute duration.
            n = m_pAudioCodec->GetChannels() * m_pAudioCodec->GetBitsPerSample() / 8 * m_pAudioCodec->GetSampleRate();
            if (n > 0)
            {
                // safety check, if channels == 0, n will result in 0, and that will result in a nice divide exception
                audioframe.duration = (unsigned int)(((__int64)audioframe.size * DVD_TIME_BASE) / n);
    
                // increase audioclock to after the packet
                m_audioClock += audioframe.duration;
            }
    
            //If we are asked to drop this packet, return a size of zero. then it won't be played
            //we currently still decode the audio.. this is needed since we still need to know it's 
            //duration to make sure clock is updated correctly.
            if ( bDropPacket )
            {
                result |= DECODE_FLAG_DROP;
            }
            return result;
        }
    
        // free the current packet
        if (pPacket)
        {
            CDVDDemuxUtils::FreeDemuxPacket(pPacket); //BRENT FIXME
            pPacket = NULL;
            pAudioPacket = NULL;
        }
    
        if (m_messageQueue.RecievedAbortRequest())
            return DECODE_FLAG_ABORT;
    
        // read next packet and return -1 on error
        LeaveCriticalSection(&m_critCodecSection); //Leave here as this might stall a while
    
        CDVDMsg* pMsg;
        MsgQueueReturnCode ret = m_messageQueue.Get(&pMsg, INFINITE);
        EnterCriticalSection(&m_critCodecSection);
            
        if (MSGQ_IS_ERROR(ret) || ret == MSGQ_ABORT)
            return DECODE_FLAG_ABORT;
    
        if (pMsg->IsType(CDVDMsg::DEMUXER_PACKET))
        {
            CDVDMsgDemuxerPacket* pMsgDemuxerPacket = (CDVDMsgDemuxerPacket*)pMsg;
            pPacket = pMsgDemuxerPacket->GetPacket();
            pMsgDemuxerPacket->m_pPacket = NULL; // XXX, test
            pAudioPacket = pPacket;
            audio_pkt_data = pPacket->pData;
            audio_pkt_size = pPacket->iSize;
        }
        else
        {
            // other data is not used here, free if
            // msg itself will still be available
            pMsg->Release();
        }
 
        // if update the audio clock with the pts
        if (pMsg->IsType(CDVDMsg::DEMUXER_PACKET) || pMsg->IsType(CDVDMsg::GENERAL_RESYNC))
        {
            if (pMsg->IsType(CDVDMsg::GENERAL_RESYNC))
            { 
                //player asked us to sync on this package
                CDVDMsgGeneralResync* pMsgGeneralResync = (CDVDMsgGeneralResync*)pMsg;
                result |= DECODE_FLAG_RESYNC;
                m_audioClock = pMsgGeneralResync->GetPts();
            }
            else if (pPacket->pts != DVD_NOPTS_VALUE) // CDVDMsg::DEMUXER_PACKET, pPacket is already set above
            {
                if (first_pkt_size == 0) 
                { 
                    //first package
                    m_audioClock = pPacket->pts;        
                }
                else if (first_pkt_pts > pPacket->pts)
                { 
                    //okey first packet in this continous stream, make sure we use the time here        
                    m_audioClock = pPacket->pts;        
                }
                else if ((unsigned __int64)m_audioClock < pPacket->pts || (unsigned __int64)m_audioClock > pPacket->pts)
                {
                    //crap, moved outsided correct pts
                    //Use pts from current packet, untill we find a better value for it.
                    //Should be ok after a couple of frames, as soon as it starts clean on a packet
                    m_audioClock = pPacket->pts;
                }
                else if (first_pkt_size == first_pkt_used)
                {
                    //Nice starting up freshly on the start of a packet, use pts from it
                    m_audioClock = pPacket->pts;
                }
            }
        }
        pMsg->Release();
    }
}

void CDVDPlayerAudio::SetSpeed(int speed)
{ 
    m_speed = speed;
  
    //if (m_speed == DVD_PLAYSPEED_PAUSE) m_dvdAudio.Pause(); //BRENT FIXME
    //else m_dvdAudio.Resume();
}
    
bool CDVDPlayerAudio::InitializeOutputDevice()
{
    int iChannels = m_pAudioCodec->GetChannels();
    int iSampleRate = m_pAudioCodec->GetSampleRate();
    int iBitsPerSample = m_pAudioCodec->GetBitsPerSample();
    //bool bPasstrough = m_pAudioCodec->NeedPasstrough(); //BRENT
    
    if (iChannels == 0 || iSampleRate == 0 || iBitsPerSample == 0)
    {
        CLog::Log(LOGERROR, "Unable to create audio device, (iChannels == 0 || iSampleRate == 0 || iBitsPerSample == 0)");
        return false;
    }
    
    CLog::Log(LOGNOTICE, "Creating audio device with codec id: %i, channels: %i, sample rate: %i", m_codec, iChannels, iSampleRate);
    if (m_dvdAudio.Create(iChannels, iSampleRate, iBitsPerSample, /*bPasstrough*/0)) // always 16 bit with ffmpeg ? //BRENT Passthrough needed?
    {
        return true;
    }
    
    CLog::Log(LOGERROR, "Failed Creating audio device with codec id: %i, channels: %i, sample rate: %i", m_codec, iChannels, iSampleRate);
    return false;
}


  


• dockerized python application takes a long time to trim a video with ffmpeg

  15 avril 2024, par Ukpa Uchechi

  The project trims YouTube videos.

  


  When I ran the ffmpeg command on the terminal, it didn't take too long to respond. The code below returns the trimmed video to the front end but it takes too long to respond. A 10 mins trim length takes about 5mins to respond. I am missing something, but I can't pinpoint the issue.

  


  backend

  


  main.py

  


  import os

from flask import Flask, request, send_file
from flask_cors import CORS, cross_origin


app = Flask(__name__)
cors = CORS(app)


current_directory = os.getcwd()
folder_name = "youtube_videos"
save_path = os.path.join(current_directory, folder_name)
output_file_path = os.path.join(save_path, 'video.mp4')

os.makedirs(save_path, exist_ok=True)

def convert_time_seconds(time_str):
    hours, minutes, seconds = map(int, time_str.split(':'))
    total_seconds = (hours * 3600) + (minutes * 60) + seconds

    return total_seconds
def convert_seconds_time(total_seconds):
    new_hours = total_seconds // 3600
    total_seconds %= 3600
    new_minutes = total_seconds // 60
    new_seconds = total_seconds % 60

    new_time_str = f'{new_hours:02}:{new_minutes:02}:{new_seconds:02}'

    return new_time_str
def add_seconds_to_time(time_str, seconds_to_add):
    total_seconds = convert_time_seconds(time_str)

    total_seconds -= seconds_to_add
    new_time_str = convert_seconds_time(total_seconds)

    return new_time_str

def get_length(start_time, end_time):
    start_time_seconds = convert_time_seconds(start_time)
    end_time_seconds = convert_time_seconds(end_time)

    length = end_time_seconds - start_time_seconds

    length_str = convert_seconds_time(length)
    return length_str
    
def download_url(url):
    command = [
        "yt-dlp",
        "-g",
        url
    ]
    
    try:
        links = subprocess.run(command, capture_output=True, text=True, check=True)
        
        video, audio = links.stdout.strip().split("\n")
        
        return video, audio

    except subprocess.CalledProcessError as e:
        print(f"Command failed with return code {e.returncode}.")
        print(f"Error output: {e.stderr}")
        return None
    except ValueError:
        print("Error: Could not parse video and audio links.")
        return None
    


def download_trimmed_video(video_link, audio_link, start_time, end_time):
    new_start_time = add_seconds_to_time(start_time, 30)
    new_end_time = get_length(start_time, end_time)

    if os.path.exists(output_file_path):
        os.remove(output_file_path)


    command = [
        'ffmpeg',
        '-ss', new_start_time + '.00',
        '-i', video_link,
        '-ss', new_start_time + '.00',
        '-i', audio_link,
        '-map', '0:v',
        '-map', '1:a',
        '-ss', '30',
        '-t', new_end_time + '.00',
        '-c:v', 'libx264',
        '-c:a', 'aac',
        output_file_path
    ]
    try:
        result = subprocess.run(command, capture_output=True, text=True, check=True)

        if result.returncode == 0:
            return "Trimmed video downloaded successfully!"
        else:
            return "Error occurred while downloading trimmed video"
    except subprocess.CalledProcessError as e:
        print(f"Command failed with return code {e.returncode}.")
        print(f"Error output: {e.stderr}")


app = Flask(__name__)


@app.route('/trimvideo', methods =["POST"])
@cross_origin()
def trim_video():
    print("here")
    data = request.get_json()
    video_link, audio_link = download_url(data["url"])
    if video_link and audio_link:
        print("Downloading trimmed video...")
        download_trimmed_video(video_link, audio_link, data["start_time"], data["end_time"])
        response = send_file(output_file_path, as_attachment=True, download_name='video.mp4')
    
        response.status_code = 200

        return response
    else:
        return "Error downloading video", 400

    




if __name__ == '__main__':
    app.run(debug=True, port=5000, host='0.0.0.0')


  


  dockerfile

  


  FROM ubuntu:latest

# Update the package list and install wget and ffmpeg
RUN apt-get update \
    && apt-get install -y wget ffmpeg python3 python3-pip \
    && rm -rf /var/lib/apt/lists/*

# Download the latest version of yt-dlp and install it
RUN wget https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp -O /usr/local/bin/yt-dlp \
    && chmod a+rx /usr/local/bin/yt-dlp

WORKDIR /app

COPY main.py /app/
COPY requirements.txt /app/


RUN pip install --no-cache-dir -r requirements.txt


# Set the default command
CMD ["python3", "main.py"]


  


  requirements.txt

  


  blinker==1.7.0
click==8.1.7
colorama==0.4.6
Flask==3.0.3
Flask-Cors==4.0.0
itsdangerous==2.1.2
Jinja2==3.1.3
MarkupSafe==2.1.5
Werkzeug==3.0.2


  


  frontend

  


  App.js

  


  &#xA;import React, { useState } from &#x27;react&#x27;;&#xA;import &#x27;./App.css&#x27;;&#xA;import axios from &#x27;axios&#x27;;&#xA;async function handleSubmit(event, url, start_time, end_time, setVideoUrl, setIsSubmitted){&#xA;  event.preventDefault();&#xA;&#xA;  if( url &amp;&amp; start_time &amp;&amp; end_time){&#xA;&#xA;    try {&#xA;      setIsSubmitted(true);&#xA;    const response = await axios.post(&#x27;http://127.0.0.1:5000/trimvideo&#x27;, {&#xA;      url: url,&#xA;      start_time: start_time,&#xA;      end_time: end_time&#xA;    },&#xA;    {&#xA;      responseType: &#x27;blob&#x27;,&#xA;      headers: {&#x27;Content-Type&#x27;: &#x27;application/json&#x27;}&#xA;    }&#xA;  )&#xA;    const blob = new Blob([response.data], { type: &#x27;video/mp4&#x27; });&#xA;    const newurl = URL.createObjectURL(blob);&#xA;&#xA;&#xA;    setVideoUrl(newurl);&#xA;    } catch (error) {&#xA;      console.error(&#x27;Error trimming video:&#x27;, error);&#xA;    }&#xA;&#xA;  } else {&#xA;    alert(&#x27;Please fill all the fields&#x27;);&#xA;  }&#xA;}&#xA;&#xA;&#xA;function App() {&#xA;  const [url, setUrl] = useState(&#x27;&#x27;);&#xA;  const [startTime, setStartTime] = useState(&#x27;&#x27;);&#xA;  const [endTime, setEndTime] = useState(&#x27;&#x27;);&#xA;  const [videoUrl, setVideoUrl] = useState(&#x27;&#x27;);&#xA;  const [isSubmitted, setIsSubmitted] = useState(false);&#xA;  return (&#xA;    <div classname="App">&#xA;        <div classname="app-header">TRIM AND DOWNLOAD YOUR YOUTUBE VIDEO HERE</div>&#xA;        <input classname="input-url" placeholder="&#x27;Enter" value="{url}" />setUrl(e.target.value)}/>&#xA;        <div classname="input-container">&#xA;          <input classname="start-time-url" placeholder="start time" value="{startTime}" />setStartTime(e.target.value)}/>&#xA;          <input classname="end-time-url" placeholder="end time" value="{endTime}" />setEndTime(e.target.value)}/>&#xA;        &#xA;        </div>&#xA;        {&#xA;          !isSubmitted &amp;&amp; <button>> handleSubmit(event, url, startTime, endTime, setVideoUrl, setIsSubmitted)} className=&#x27;trim-button&#x27;>Trim</button>&#xA;        }&#xA;&#xA;        {&#xA;         ( isSubmitted &amp;&amp; !videoUrl) &amp;&amp;   <div classname="dot-pulse"></div>&#xA;        }&#xA;&#xA;&#xA;        {&#xA;          videoUrl &amp;&amp; <video controls="controls" autoplay="autoplay" width="500" height="360">&#xA;          <source src="{videoUrl}" type="&#x27;video/mp4&#x27;"></source>&#xA;        </video>&#xA;        }&#xA;&#xA;        &#xA;    </div>&#xA;  );&#xA;}&#xA;&#xA;export default App;&#xA;

  


• Issues with Publishing and Subscribing Rates for H.264 Video Streaming over RabbitMQ

  7 octobre 2024, par Luis

  I am working on a project to stream an H.264 video file using RabbitMQ (AMQP protocol) and display it in a web application. The setup involves capturing video frames, encoding them, sending them to RabbitMQ, and then consuming and decoding them on the web application side using Flask and Flask-SocketIO.

  


  However, I am encountering performance issues with the publishing and subscribing rates in RabbitMQ. I cannot seem to achieve more than 10 messages per second. This is not sufficient for smooth video streaming.
I need help to diagnose and resolve these performance bottlenecks.

  


  Here is my code :

  


  

  • Video Capture and Publishing Script :

  


  


  # RabbitMQ setup
RABBITMQ_HOST = 'localhost'
EXCHANGE = 'DRONE'
CAM_LOCATION = 'Out_Front'
KEY = f'DRONE_{CAM_LOCATION}'
QUEUE_NAME = f'DRONE_{CAM_LOCATION}_video_queue'

# Path to the H.264 video file
VIDEO_FILE_PATH = 'videos/FPV.h264'

# Configure logging
logging.basicConfig(level=logging.INFO)

@contextmanager
def rabbitmq_channel(host):
    """Context manager to handle RabbitMQ channel setup and teardown."""
    connection = pika.BlockingConnection(pika.ConnectionParameters(host))
    channel = connection.channel()
    try:
        yield channel
    finally:
        connection.close()

def initialize_rabbitmq(channel):
    """Initialize RabbitMQ exchange and queue, and bind them together."""
    channel.exchange_declare(exchange=EXCHANGE, exchange_type='direct')
    channel.queue_declare(queue=QUEUE_NAME)
    channel.queue_bind(exchange=EXCHANGE, queue=QUEUE_NAME, routing_key=KEY)

def send_frame(channel, frame):
    """Encode the video frame using FFmpeg and send it to RabbitMQ."""
    ffmpeg_path = 'ffmpeg/bin/ffmpeg.exe'
    cmd = [
        ffmpeg_path,
        '-f', 'rawvideo',
        '-pix_fmt', 'rgb24',
        '-s', '{}x{}'.format(frame.shape[1], frame.shape[0]),
        '-i', 'pipe:0',
        '-f', 'h264',
        '-vcodec', 'libx264',
        '-pix_fmt', 'yuv420p',
        '-preset', 'ultrafast',
        'pipe:1'
    ]
    
    start_time = time.time()
    process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = process.communicate(input=frame.tobytes())
    encoding_time = time.time() - start_time
    
    if process.returncode != 0:
        logging.error("ffmpeg error: %s", err.decode())
        raise RuntimeError("ffmpeg error")
    
    frame_size = len(out)
    logging.info("Sending frame with shape: %s, size: %d bytes", frame.shape, frame_size)
    timestamp = time.time()
    formatted_timestamp = datetime.fromtimestamp(timestamp).strftime('%H:%M:%S.%f')
    logging.info(f"Timestamp: {timestamp}") 
    logging.info(f"Formatted Timestamp: {formatted_timestamp[:-3]}")
    timestamp_bytes = struct.pack('d', timestamp)
    message_body = timestamp_bytes + out
    channel.basic_publish(exchange=EXCHANGE, routing_key=KEY, body=message_body)
    logging.info(f"Encoding time: {encoding_time:.4f} seconds")

def capture_video(channel):
    """Read video from the file, encode frames, and send them to RabbitMQ."""
    if not os.path.exists(VIDEO_FILE_PATH):
        logging.error("Error: Video file does not exist.")
        return
    cap = cv2.VideoCapture(VIDEO_FILE_PATH)
    if not cap.isOpened():
        logging.error("Error: Could not open video file.")
        return
    try:
        while True:
            start_time = time.time()
            ret, frame = cap.read()
            read_time = time.time() - start_time
            if not ret:
                break
            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            frame_rgb = np.ascontiguousarray(frame_rgb) # Ensure the frame is contiguous
            send_frame(channel, frame_rgb)
            cv2.imshow('Video', frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
            logging.info(f"Read time: {read_time:.4f} seconds")
    finally:
        cap.release()
        cv2.destroyAllWindows()


  


  

  • the backend (flask) :

  


  


  app = Flask(__name__)
CORS(app)
socketio = SocketIO(app, cors_allowed_origins="*")

RABBITMQ_HOST = 'localhost'
EXCHANGE = 'DRONE'
CAM_LOCATION = 'Out_Front'
QUEUE_NAME = f'DRONE_{CAM_LOCATION}_video_queue'

def initialize_rabbitmq():
    connection = pika.BlockingConnection(pika.ConnectionParameters(RABBITMQ_HOST))
    channel = connection.channel()
    channel.exchange_declare(exchange=EXCHANGE, exchange_type='direct')
    channel.queue_declare(queue=QUEUE_NAME)
    channel.queue_bind(exchange=EXCHANGE, queue=QUEUE_NAME, routing_key=f'DRONE_{CAM_LOCATION}')
    return connection, channel

def decode_frame(frame_data):
    # FFmpeg command to decode H.264 frame data
    ffmpeg_path = 'ffmpeg/bin/ffmpeg.exe'
    cmd = [
        ffmpeg_path,
        '-f', 'h264',
        '-i', 'pipe:0',
        '-pix_fmt', 'bgr24',
        '-vcodec', 'rawvideo',
        '-an', '-sn',
        '-f', 'rawvideo',
        'pipe:1'
    ]
    process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    start_time = time.time()  # Start timing the decoding process
    out, err = process.communicate(input=frame_data)
    decoding_time = time.time() - start_time  # Calculate decoding time
    
    if process.returncode != 0:
        print("ffmpeg error: ", err.decode())
        return None
    frame_size = (960, 1280, 3)  # frame dimensions expected by the frontend
    frame = np.frombuffer(out, np.uint8).reshape(frame_size)
    print(f"Decoding time: {decoding_time:.4f} seconds")
    return frame

def format_timestamp(ts):
    dt = datetime.fromtimestamp(ts)
    return dt.strftime('%H:%M:%S.%f')[:-3]

def rabbitmq_consumer():
    connection, channel = initialize_rabbitmq()
    for method_frame, properties, body in channel.consume(QUEUE_NAME):
        message_receive_time = time.time()  # Time when the message is received

        # Extract the timestamp from the message body
        timestamp_bytes = body[:8]
        frame_data = body[8:]
        publish_timestamp = struct.unpack('d', timestamp_bytes)[0]

        print(f"Message Receive Time: {message_receive_time:.4f} ({format_timestamp(message_receive_time)})")
        print(f"Publish Time: {publish_timestamp:.4f} ({format_timestamp(publish_timestamp)})")

        frame = decode_frame(frame_data)
        decode_time = time.time() - message_receive_time  # Calculate decode time

        if frame is not None:
            _, buffer = cv2.imencode('.jpg', frame)
            frame_data = buffer.tobytes()
            socketio.emit('video_frame', {'frame': frame_data, 'timestamp': publish_timestamp}, namespace='/')
            emit_time = time.time()  # Time after emitting the frame

            # Log the time taken to emit the frame and its size
            rtt = emit_time - publish_timestamp  # Calculate RTT from publish to emit
            print(f"Current Time: {emit_time:.4f} ({format_timestamp(emit_time)})")
            print(f"RTT: {rtt:.4f} seconds")
            print(f"Emit time: {emit_time - message_receive_time:.4f} seconds, Frame size: {len(frame_data)} bytes")
        channel.basic_ack(method_frame.delivery_tag)

@app.route('/')
def index():
    return render_template('index.html')

@socketio.on('connect')
def handle_connect():
    print('Client connected')

@socketio.on('disconnect')
def handle_disconnect():
    print('Client disconnected')

if __name__ == '__main__':
    consumer_thread = threading.Thread(target=rabbitmq_consumer)
    consumer_thread.daemon = True
    consumer_thread.start()
    socketio.run(app, host='0.0.0.0', port=5000)



  


  How can I optimize the publishing and subscribing rates to handle a higher number of messages per second ?

  


  Any help or suggestions would be greatly appreciated !

  


  I attempted to use threading and multiprocessing to handle multiple frames concurrently and I tried to optimize the frame decoding function to make it faster but with no success.