q-tools/py-packages/ffmpeg-parser/ffmpegparser/ffprobeparser.py

#!/usr/bin/env python3

import subprocess
import sys
import os
import time
import parse
import argparse
import json
from ansi.colour import fg, fx
from datetime import datetime


class Probe:
    def __init__(self):
        parser = argparse.ArgumentParser(
            description="Readable version of ffprobe. Give filename as argument."
        )
        parser.add_argument(
            "-c", dest="colorize", action="store_true", default=False, help="Colorize"
        )
        parser.add_argument("filename", help="File to probe")
        parsed = parser.parse_args()

        self.filename = parsed.filename
        self.colorize = parsed.colorize
        self.run_probe()
        self.print()

    def run_probe(self):
        process = subprocess.Popen(
            [
                "",
                "-v",
                "error",
                "-hide_banner",
                "-of",
                "default=noprint_wrappers=0",
                "-print_format",
                "json",
                "-show_format",
                "-show_streams",
                self.filename,
            ],
            executable="ffprobe",
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        self.raw_data, errors = process.communicate()
        if len(errors) > 0:
            print(errors)
        self.json_data = json.loads(self.raw_data)

    def print(self):
        # ~ print(self.json_data['format'])
        self.json_data["format"]["time"] = self._timestr(
            self.json_data["format"].get("duration", "0")
        )
        self.json_data["format"]["hrate"] = self._speedfmt(
            self.json_data["format"].get("bit_rate", "0")
        )
        self.json_data["format"]["hsize"] = self._sizefmt(
            float(self.json_data["format"].get("size", 0))
        )
        self.json_data["format"].update(self._get_colors())

        msg = """{Y}==== Format ===={z}
   File: {filename}
 Format: {format_long_name} ({format_name})
 Length: {time}
   Size: {hsize}
Bitrate: {hrate}
Streams: {nb_streams}""".format(
            **self.json_data["format"]
        )
        print(msg)
        for stream in self.json_data["streams"]:
            original = stream.copy()
            stream.update(self._get_colors())

            if stream["codec_type"] == "video":
                msg = self.format_video(stream)
            elif stream["codec_type"] == "audio":
                msg = self.format_audio(stream)
            elif stream["codec_type"] == "subtitle":
                msg = self.format_subtitle(stream)
            else:
                msg = "Unrecognized stream\n{}".format(json.dumps(original, indent=2))

            print(msg)

    def format_subtitle(self, stream):

        stream["tags-lang"] = stream.get("tags", {}).get("language", "NA")
        msg = """{Y}==== Stream:{index} ===={z}
    Type: {G}{codec_type}{z}
   Codec: {codec_long_name} ({codec_name})
Language: {tags-lang}""".format(
            **stream
        )
        return msg

    def format_video(self, stream):
        stream["hrate"] = self._speedfmt(stream.get("bit_rate", 0))
        stream["fps"] = "{:.2f}".format(
            float(stream["r_frame_rate"].split("/")[0])
            / float(stream["r_frame_rate"].split("/")[1])
        )
        msg = """{Y}==== Stream:{index} ===={z}
       Type: {G}{codec_type}{z}
      Codec: {codec_long_name} ({codec_name})
    Bitrate: {hrate}
 Resolution: {width}x{height}
     Aspect: {display_aspect_ratio}
    Profile: {profile}
        FPS: {r_frame_rate} ({fps})""".format(
            **stream
        )

        return msg

    def format_audio(self, stream):
        stream["hrate"] = self._speedfmt(stream.get("bit_rate", 0))
        msg = """{Y}==== Stream:{index} ===={z}
       Type: {G}{codec_type}{z}
      Codec: {codec_long_name} ({codec_name})
    Bitrate: {hrate}
Sample Rate: {sample_rate} Hz
   Channels: {channels} ({channel_layout})""".format(
            **stream
        )

        return msg

    def _timestr(self, sec):
        msec = int((float(sec) - int(float(sec))) * 1000)
        sec = int(float(sec))
        try:
            hours = int(sec) // 3600 % 24
            minutes = int(sec) // 60 % 60
            seconds = int(sec) % 60

            return "{:02d}:{:02d}:{:02d}.{:03d}".format(hours, minutes, seconds, msec)
        except Exception:
            return sec

    def _mbstr(self, b):

        try:
            return "{:.1f}".format((float(b) / (1024 ** 2)))
        except Exception:
            return b

    def _speedfmt(self, b):

        return self._sizefmt(float(b), suffix="bit/s")
        try:
            return "{:.1f}".format((float(b) / (1024 ** 2)))
        except Exception:
            return b

    def _sizefmt(self, num, suffix="B"):

        num = float(num)
        for unit in ["", "K", "M", "G", "T", "P", "E", "Z"]:
            if abs(num) < 1024.0:
                return "%3.1f %s%s" % (num, unit, suffix)
            num /= 1024.0
        return "%.1f %s%s" % (num, "Y", suffix)

    def _get_colors(self):

        if self.colorize:
            return {
                "Y": fg.boldyellow,
                "G": fg.boldgreen,
                "z": fx.reset,
            }
        else:
            return {
                "Y": "",
                "G": "",
                "z": "",
            }


def main():
    Probe()


if __name__ == "__main__":
    main()