ffmpeg_streaming.ffprobe

Complexity

Total Complexity

13

Size/Duplication

Total Lines	74
Duplicated Lines	0 %

Importance

Changes

0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	FFProbe.save_as_json()	0	3	2
A	FFProbe.__init__()	0	9	2
A	FFProbe.streams()	0	2	1
A	FFProbe.all()	0	2	1
A	FFProbe.format()	0	2	1
A	FFProbe.video_size()	0	9	3
A	FFProbe.bitrate()	0	10	2

"""
ffmpeg_streaming._ffprobe
~~~~~~~~~~~~

Probe the video


:copyright: (c) 2020 by Amin Yazdanpanah.
:website: https://www.aminyazdanpanah.com
:email: contact@aminyazdanpanah.com
:license: MIT, see LICENSE for more details.
"""

import json
import logging
import subprocess

from ._media_streams import Streams
from ._media_property import Size, Bitrate


class FFProbe:
    def __init__(self, filename, cmd='ffprobe'):
        commands = [cmd, '-show_format', '-show_streams', '-of', 'json', filename]
        logging.info("ffprobe running command: {}".format(" ".join(commands)))
        process = subprocess.Popen(commands, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.out, err = process.communicate()
        if process.returncode != 0:
            logging.error(str(self.out) + str(err))
            raise RuntimeError('ffprobe', self.out, err)
        logging.info("ffprobe executed command successfully!")

    def streams(self):
        return Streams(json.loads(self.out.decode('utf-8'))['streams'])

    def format(self):
        return json.loads(self.out.decode('utf-8'))['format']

    def all(self):
        return json.loads(self.out.decode('utf-8'))

    def save_as_json(self, path):
        with open(path, 'w') as probe:
            probe.write(self.out.decode('utf-8'))

    @property
    def video_size(self) -> Size:
        width = int(self.streams().video().get('width', 0))
        height = int(self.streams().video().get('height', 0))

        if width == 0 or height == 0:
            raise RuntimeError('It could not determine the value of width/height')

        return Size(width, height)

    @property
    def bitrate(self, _type: str = "k") -> Bitrate:
        overall = int(self.format().get('bit_rate', 0))
        video = int(self.streams().video().get('bit_rate', 0))
        audio = int(self.streams().audio().get('bit_rate', 0))

        if overall == 0:
            raise RuntimeError('It could not determine the value of bitrate')

        return Bitrate(video, audio, overall, type=_type)


def ffprobe(filename, cmd='ffprobe'):
    return FFProbe(filename, cmd)


__all__ = [
    'FFProbe'
]