PyAV

av.open(file, mode='r', **kwargs)

av.open("/path/to/some.mp4")  # File
av.open("https://file-examples-com.github.io/uploads/2017/04/file_example_MP4_480_1_5MG.mp4")  # URL
av.open("rtsp://192.168.0.50:554/media/1/1/Profile1")  # RTSP
av.open("rtsp://192.168.0.95:8554/live.sdp")  # RTSP
av.open("/dev/video0", format="v4l2", options={"video_size": "640x480"})  # Linux
av.open("default:none", format="avfoundation", options={"video_size": "640x480"})  # macOS
av.open("video=Integrated Camera", format="dshow", options={"video_size": "640x480"})  # Windows

import numpy as np
import cv2

import av

video0 = 'small.mp4'
video1 = 'bus-fire.mp4'
video2 = 'rtsp://admin:[email protected]/stream1'

container = av.open(video1, options={'rtsp_transport': 'udp'})
container.streams.video[0].thread_type = 'AUTO'  # Go faster!

for frame in container.decode(video=0):

    print(frame)
    array = frame.to_ndarray(format='bgr24')

    cv2.imshow('frame', array)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

import av

container = av.open("My_Super_Video.mp4")
total_frames = container.streams.video[0].frames

import av
import av.datasets

content = av.datasets.curated("pexels/time-lapse-video-of-night-sky-857195.mp4")
with av.open(content) as container:
    # Signal that we only want to look at keyframes.
    stream = container.streams.video[0]
    stream.codec_context.skip_frame = "NONKEY"

    for frame in container.decode(stream):

        print(frame)

        # We use `frame.pts` as `frame.index` won't make must sense with the `skip_frame`.
        frame.to_image().save(
            "night-sky.{:04d}.jpg".format(frame.pts),
            quality=80,
        )

import av
import time
import numpy as np
import cv2

video0 = 'small.mp4'
video1 = 'bus-fire.mp4'
video2 = 'rtsp://admin:[email protected]/stream1'

video_src = video2
video_index = 0
frame_format = 'bgr24'
frame_width = 1270
frame_height = 720
options = {
    'rtsp_transport': 'udp',
    'fflags': 'nobuffer'
}
container_options = {}
stream_options = []

container = av.open(video_src, options=options,
                    container_options=container_options,
                    stream_options=stream_options)
container.streams.video[video_index].thread_type = 'AUTO'  # Go faster! (SLICE, AUTO, FRAME, NONE)
container.streams.video[video_index].codec_context.flags = 'LOW_DELAY'
frames = container.decode(video=video_index)

while True:

    begin = time.time()
    frame = next(frames)

    last_frame = frame.to_ndarray(width=frame_width,
                                  height=frame_height,
                                  format=frame_format,
                                  interpolation='FAST_BILINEAR')
    last_index = frame.index
    last_pts = frame.pts

    end = time.time()
    elapsed = int((end-begin)*1000)
    print(f'index={last_index}, pts={last_pts}, shape={last_frame.shape}, elapsed={elapsed}ms')

    cv2.imshow('frame', last_frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

# -*- coding: utf-8 -*-

from argparse import Namespace
from typing import Callable

import av
import cv2
import numpy as np

from ddrm.logging.logging import logger


class CamberApp:
    def __init__(self, args: Namespace, printer: Callable[..., None] = print):
        assert args is not None
        assert printer is not None

        assert isinstance(args.debug, bool)
        assert isinstance(args.verbose, int)
        assert isinstance(args.bind, str)
        assert isinstance(args.port, int)
        assert isinstance(args.timeout, float)
        assert isinstance(args.source, str)
        assert isinstance(args.format, str)
        assert isinstance(args.destination, str)

        self._debug = args.debug
        self._verbose = args.verbose
        self._bind = args.bind
        self._port = args.port
        self._timeout = args.timeout
        self._source = args.source
        self._format = args.format
        self._destination = args.destination

        logger.info(f"Debug enabled: {self._debug}")
        logger.info(f"Verbose level: {self._verbose}")
        logger.info(f"Web Server bind: {self._bind}")
        logger.info(f"Web Server port: {self._port}")
        logger.info(f"Web Server timeout: {self._timeout}s")
        logger.info(f"Source URL: {self._source}")
        logger.info(f"Destination format: {self._format}")
        logger.info(f"Destination URL: {self._destination}")

    def run(self) -> int:
        format_options = {'rtsp_transport': 'tcp', 'fflags': 'nobuffer'}

        logger.debug("Open input ..")
        input_container = av.open(self._source, options=format_options)

        logger.debug("Open output ..")
        output_container = av.open(self._destination, mode="w", format=self._format)

        video_stream = None
        for stream in input_container.streams:
            if stream.type == "video":
                video_stream = stream
                break

        if video_stream is None:
            logger.error("Not found video stream from source")
            return 1

        video_stream.thread_type = "AUTO"
        # video_stream.no_buffer = True
        # , 'flags': 'no_buffer'
        video_stream.codec_context.low_delay = True
        logger.debug(f"Find video stream: {video_stream}")
        # output_stream = output_container.add_stream(template=video_stream)

        output_stream = output_container.add_stream('libx264')
        output_stream.width = video_stream.width
        output_stream.height = video_stream.height
        output_stream.pix_fmt = 'yuv420p'
        output_stream.options = {
            'preset': 'fast',
            'crf': '28',
            'turn': 'zerolatency',
        }

        # ffmpeg time is complicated
        # more at https://github.com/PyAV-Org/PyAV/blob/main/docs/api/time.rst
        # our situation is the "encoding" one

        # this is independent of the "fps" you give above
        # 1/1000 means milliseconds (and you can use that, no problem)
        # 1/2 means half a second (would be okay for the delays we use below)
        # 1/30 means ~33 milliseconds
        # you should use the least fraction that makes sense for you

        logger.warning(f"input_stream.time_base: {video_stream.time_base}")
        logger.warning(f"output_stream.time_base: {output_stream.time_base}")
        # output_stream.time_base = video_stream.time_base

        # this says when to show the next frame
        # (increment by how long the current frame will be shown)
        my_pts = 0  # [seconds]
        # below we'll calculate that into our chosen time base

        frame_i = 0

        for packet in input_container.demux(video_stream):

            # We need to skip the "flushing" packets that `demux` generates.
            if packet.dts is None:
                continue

            # Discard frames from previous packets to get the latest frame.
            frames = [frame for frame in packet.decode()]
            if not frames:
                continue

            if len(frames) >= 2:
                # Discard frames from previous packets to get the latest frame.
                logger.debug(f"Discard {len(frames) - 1} frames")

            latest_frame = frames[-1]
            logger.debug(f"Latest frame: {latest_frame}")

            image = latest_frame.to_ndarray(format="bgr24")

            # gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
            # colord_gray = cv2.cvtColor(gray, cv2.COLOR_GRAY2RGB)

            result_image = image
            av_frame = av.VideoFrame.from_ndarray(result_image, format="bgr24")

            # av_frame.pts = latest_frame.pts
            # av_frame.time_base = latest_frame.time_base

            # # seconds -> counts of time_base
            # av_frame.pts = int(round(my_pts / output_stream.time_base))
            #
            # # increment by display time to pre-determine next frame's PTS
            # my_pts += 1.0 if ((frame_i // 3) % 2 == 0) else 0.5
            # frame_i += 1

            logger.debug(f"Latest av frame: {av_frame}")

            for output_packet in output_stream.encode(av_frame):
                output_container.mux(output_packet)

        input_container.close()
        output_container.mux(output_stream.encode(None))
        output_container.close()

        return 0


def camber_main(args: Namespace, printer: Callable[..., None] = print) -> int:
    assert args is not None
    assert printer is not None
    return CamberApp(args).run()

from av.video.frame import VideoFrame
# ...
assert isinstance(frame, VideoFrame)
frame.to_ndarray(format="bgr24")

# Disable PyAV's logging framework as it can lead to thread deadlocks.
av.logging.restore_default_callback()

import av.logging

def do_log(message):
    av.logging.log(av.logging.INFO, 'test', message)

with av.logging.Capture() as logs:
    frame.to_ndarray(format="bgr24")

av.logging.set_level(av.logging.ERROR)

logging.getLogger('libav').setLevel(logging.ERROR)

# -*- coding: utf-8 -*-

import os
from importlib import import_module

def get_module_directory(module) -> str:
    module_path = getattr(module, "__path__", None)
    if module_path:
        assert isinstance(module_path, list)
        return module_path[0]

    module_file = getattr(module, "__file__", None)
    if module_file:
        assert isinstance(module_file, str)
        return os.path.dirname(module_file)

    raise RuntimeError(f"The '{module.__name__}' module path is unknown")

# -*- coding: utf-8 -*-

import re
import os
from importlib import import_module
from recc.package.package_utils import get_module_directory


def find_ffmpeg_library_path(library_name: str) -> str:
    av_module = import_module("av")
    av_core_module = import_module("av._core")

    av_module_dir = get_module_directory(av_module)
    if av_module_dir[-1] == "/":
        av_module_dir = av_module_dir[:-1]
    av_lib_module_dir = av_module_dir + ".libs"
    if not os.path.isdir(av_lib_module_dir):
        raise FileNotFoundError(f"Not found module directory: {av_lib_module_dir}")

    versions = getattr(av_core_module, "library_versions")
    assert isinstance(versions, dict)
    if library_name not in versions:
        raise ValueError(f"Unknown library name: {library_name}")

    major = versions[library_name][0]
    minor = versions[library_name][1]
    patch = versions[library_name][2]
    assert isinstance(major, int)
    assert isinstance(minor, int)
    assert isinstance(patch, int)

    regex_pattern = r"^{name}-(.*)\.so\.{major}\.{minor}\.{patch}$".format(
        name=library_name, major=major, minor=minor, patch=patch
    )
    matcher = re.compile(regex_pattern)

    for file in os.listdir(av_lib_module_dir):
        if matcher.match(file):
            return os.path.join(av_lib_module_dir, file)

    raise FileNotFoundError(f"Not found ffmpeg library: {library_name}")


# noinspection SpellCheckingInspection
def find_libavutil_path() -> str:
    return find_ffmpeg_library_path("libavutil")

# -*- coding: utf-8 -*-

from ctypes import CDLL
from recc_plugin_vms.av.modules import find_libavutil_path
from av.logging import ERROR


# noinspection SpellCheckingInspection
def av_log_set_level(level: int) -> None:
    libavutil = CDLL(find_libavutil_path())
    try:
        libavutil.av_log_set_level(level)
    finally:
        del libavutil


# noinspection SpellCheckingInspection
def silent_av_warnings() -> None:
    # e.g.
    # [swscaler @ 0x5569f73ced40]
    # deprecated pixel format used, make sure you did set range correctly
    av_log_set_level(ERROR)

silent_av_warnings()