def execute_ffmpeg()

in broadcast-monitoring/src/audio_detect/app/audio_detect.py [0:0]

• 14 lines of code
• 6 McCabe index (conditional complexity)

def execute_ffmpeg(input_file, processors=None, **kwargs):
    """
    Run ffmpeg with a set of audio processors to add filters to a call
    and process the results into a dict
    """
    if processors is None:
        processors = [
            AudioProcessor('volumedetect', with_volumedetect, parse_volume_output),
            AudioProcessor('silencedetect', with_silencedetect, parse_silence_output)
        ]

    stream = ffmpeg.input(input_file)

    for with_filter in [ap.with_filter for ap in processors]:
        stream = with_filter(stream, **kwargs)

    ret_code, out = ffmpeg.output(stream, '-', format='null').run(quiet=True)

    if ret_code:
        raise RuntimeError

    output_lines = out.decode('utf-8').splitlines()

    return {ap.name: ap.output_processor(output_lines) for ap in processors}