Make video/codeccontext pure

Author: WyattBlue

av/audio/codeccontext.py

@@ -10,7 +10,7 @@
 @cython.cclass
 class AudioCodecContext(CodecContext):
     @cython.cfunc
-    def _prepare_frames_for_encode(self, input_frame: Frame | None):
+    def _prepare_frames_for_encode(self, input_frame: Frame | None) -> list:
         frame: AudioFrame | None = input_frame
         allow_var_frame_size: cython.bint = (
             self.ptr.codec.capabilities & lib.AV_CODEC_CAP_VARIABLE_FRAME_SIZE

av/codec/context.pxd

@@ -40,7 +40,7 @@ cdef class CodecContext:
     # TODO: Remove the `Packet` from `_setup_decoded_frame` (because flushing packets
     # are bogus). It should take all info it needs from the context and/or stream.
     cdef _prepare_and_time_rebase_frames_for_encode(self, Frame frame)
-    cdef _prepare_frames_for_encode(self, Frame frame)
+    cdef list _prepare_frames_for_encode(self, Frame frame)
     cdef _setup_encoded_packet(self, Packet)
     cdef _setup_decoded_frame(self, Frame, Packet)
 

av/codec/context.py

@@ -386,7 +386,7 @@ def _send_packet_and_recv(self, packet: Packet | None):
         return out
 
     @cython.cfunc
-    def _prepare_frames_for_encode(self, frame: Frame | None):
+    def _prepare_frames_for_encode(self, frame: Frame | None) -> list:
         return [frame]
 
     @cython.cfunc

av/container/input.py

@@ -236,11 +236,11 @@ def seek(
         self,
         offset,
         *,
-        backward: bint = True,
-        any_frame: bint = False,
+        backward: cython.bint = True,
+        any_frame: cython.bint = False,
         stream: Stream | None = None,
-        unsupported_frame_offset: bint = False,
-        unsupported_byte_offset: bint = False,
+        unsupported_frame_offset: cython.bint = False,
+        unsupported_byte_offset: cython.bint = False,
     ):
         """seek(offset, *, backward=True, any_frame=False, stream=None)
 

av/video/codeccontext.pxd

@@ -16,18 +16,11 @@ cdef struct AVCodecPrivateData:
 
 
 cdef class VideoCodecContext(CodecContext):
-
     cdef AVCodecPrivateData _private_data
-
     cdef VideoFormat _format
     cdef _build_format(self)
-
     cdef int last_w
     cdef int last_h
     cdef readonly VideoReformatter reformatter
-
-    # For encoding.
     cdef readonly int encoded_frame_count
-
-    # For decoding.
     cdef VideoFrame next_frame

av/video/codeccontext.pyx av/video/codeccontext.pyav/video/codeccontext.pyx renamed to av/video/codeccontext.py

@@ -1,51 +1,70 @@
-cimport libav as lib
-from libc.stdint cimport int64_t
-
-from av.codec.context cimport CodecContext
-from av.codec.hwaccel cimport HWAccel, HWConfig
-from av.error cimport err_check
-from av.frame cimport Frame
-from av.packet cimport Packet
-from av.utils cimport avrational_to_fraction, to_avrational
-from av.video.format cimport VideoFormat, get_pix_fmt, get_video_format
-from av.video.frame cimport VideoFrame, alloc_video_frame
-from av.video.reformatter cimport VideoReformatter
-
-
-cdef lib.AVPixelFormat _get_hw_format(lib.AVCodecContext *ctx, const lib.AVPixelFormat *pix_fmts) noexcept:
+import cython
+import cython.cimports.libav as lib
+from cython.cimports.av.codec.context import CodecContext
+from cython.cimports.av.codec.hwaccel import HWAccel, HWConfig
+from cython.cimports.av.error import err_check
+from cython.cimports.av.frame import Frame
+from cython.cimports.av.packet import Packet
+from cython.cimports.av.utils import avrational_to_fraction, to_avrational
+from cython.cimports.av.video.format import VideoFormat, get_pix_fmt, get_video_format
+from cython.cimports.av.video.frame import VideoFrame, alloc_video_frame
+from cython.cimports.av.video.reformatter import VideoReformatter
+from cython.cimports.libc.stdint import int64_t
+
+
+@cython.cfunc
+@cython.exceptval(check=False)
+def _get_hw_format(
+    ctx: cython.pointer[lib.AVCodecContext],
+    pix_fmts: cython.pointer[cython.const[lib.AVPixelFormat]],
+) -> lib.AVPixelFormat:
     # In the case where we requested accelerated decoding, the decoder first calls this function
     # with a list that includes both the hardware format and software formats.
     # First we try to pick the hardware format if it's in the list.
     # However, if the decoder fails to initialize the hardware, it will call this function again,
     # with only software formats in pix_fmts. We return ctx->sw_pix_fmt regardless in this case,
     # because that should be in the candidate list. If not, we are out of ideas anyways.
-    cdef AVCodecPrivateData* private_data = <AVCodecPrivateData*>ctx.opaque
-    i = 0
+    private_data: cython.pointer[AVCodecPrivateData] = cython.cast(
+        cython.pointer[AVCodecPrivateData], ctx.opaque
+    )
+    i: cython.int = 0
     while pix_fmts[i] != -1:
         if pix_fmts[i] == private_data.hardware_pix_fmt:
             return pix_fmts[i]
         i += 1
-    return ctx.sw_pix_fmt if private_data.allow_software_fallback else lib.AV_PIX_FMT_NONE
-
+    return (
+        ctx.sw_pix_fmt if private_data.allow_software_fallback else lib.AV_PIX_FMT_NONE
+    )
 
-cdef class VideoCodecContext(CodecContext):
 
+@cython.cclass
+class VideoCodecContext(CodecContext):
     def __cinit__(self, *args, **kwargs):
         self.last_w = 0
         self.last_h = 0
 
-    cdef _init(self, lib.AVCodecContext *ptr, const lib.AVCodec *codec, HWAccel hwaccel):
-        CodecContext._init(self, ptr, codec, hwaccel)  # TODO: Can this be `super`?
+    @cython.cfunc
+    def _init(
+        self,
+        ptr: cython.pointer[lib.AVCodecContext],
+        codec: cython.pointer[cython.const[lib.AVCodec]],
+        hwaccel: HWAccel | None,
+    ):
+        CodecContext._init(self, ptr, codec, hwaccel)
 
         if hwaccel is not None:
             try:
                 self.hwaccel_ctx = hwaccel.create(self.codec)
                 self.ptr.hw_device_ctx = lib.av_buffer_ref(self.hwaccel_ctx.ptr)
                 self.ptr.pix_fmt = self.hwaccel_ctx.config.ptr.pix_fmt
                 self.ptr.get_format = _get_hw_format
-                self._private_data.hardware_pix_fmt = self.hwaccel_ctx.config.ptr.pix_fmt
-                self._private_data.allow_software_fallback = self.hwaccel.allow_software_fallback
-                self.ptr.opaque = &self._private_data
+                self._private_data.hardware_pix_fmt = (
+                    self.hwaccel_ctx.config.ptr.pix_fmt
+                )
+                self._private_data.allow_software_fallback = (
+                    self.hwaccel.allow_software_fallback
+                )
+                self.ptr.opaque = cython.address(self._private_data)
             except NotImplementedError:
                 # Some streams may not have a hardware decoder. For example, many action
                 # cam videos have a low resolution mjpeg stream, which is usually not
@@ -60,20 +79,20 @@ def __cinit__(self, *args, **kwargs):
         self._build_format()
         self.encoded_frame_count = 0
 
-    cdef _prepare_frames_for_encode(self, Frame input):
-        if not input:
+    @cython.cfunc
+    def _prepare_frames_for_encode(self, input: Frame | None) -> list:
+        if input is None or not input:
             return [None]
 
-        cdef VideoFrame vframe = input
-
         if self._format is None:
             raise ValueError("self._format is None, cannot encode")
 
+        vframe: VideoFrame = input
         # Reformat if it doesn't match.
         if (
-            vframe.format.pix_fmt != self._format.pix_fmt or
-            vframe.width != self.ptr.width or
-            vframe.height != self.ptr.height
+            vframe.format.pix_fmt != self._format.pix_fmt
+            or vframe.width != self.ptr.width
+            or vframe.height != self.ptr.height
         ):
             if not self.reformatter:
                 self.reformatter = VideoReformatter()
@@ -84,73 +103,74 @@ def __cinit__(self, *args, **kwargs):
 
         # There is no pts, so create one.
         if vframe.ptr.pts == lib.AV_NOPTS_VALUE:
-            vframe.ptr.pts = <int64_t>self.encoded_frame_count
+            vframe.ptr.pts = cython.cast(int64_t, self.encoded_frame_count)
 
         self.encoded_frame_count += 1
-
         return [vframe]
 
-    cdef Frame _alloc_next_frame(self):
+    @cython.cfunc
+    def _alloc_next_frame(self) -> Frame:
         return alloc_video_frame()
 
-    cdef _setup_decoded_frame(self, Frame frame, Packet packet):
+    @cython.cfunc
+    def _setup_decoded_frame(self, frame: Frame, packet: Packet):
         CodecContext._setup_decoded_frame(self, frame, packet)
-        cdef VideoFrame vframe = frame
+        vframe: VideoFrame = frame
         vframe._init_user_attributes()
 
-    cdef _transfer_hwframe(self, Frame frame):
+    @cython.cfunc
+    def _transfer_hwframe(self, frame: Frame):
         if self.hwaccel_ctx is None:
             return frame
-
         if frame.ptr.format != self.hwaccel_ctx.config.ptr.pix_fmt:
             # If we get a software frame, that means we are in software fallback mode, and don't actually
             # need to transfer.
             return frame
 
-        cdef Frame frame_sw
-
-        frame_sw = self._alloc_next_frame()
-
+        frame_sw: Frame = self._alloc_next_frame()
         err_check(lib.av_hwframe_transfer_data(frame_sw.ptr, frame.ptr, 0))
-
-        # TODO: Is there anything else to transfer?!
+        # TODO: Is there anything else to transfer?
         frame_sw.pts = frame.pts
-
         return frame_sw
 
-    cdef _build_format(self):
-        self._format = get_video_format(<lib.AVPixelFormat>self.ptr.pix_fmt, self.ptr.width, self.ptr.height)
+    @cython.cfunc
+    def _build_format(self):
+        self._format = get_video_format(
+            cython.cast(lib.AVPixelFormat, self.ptr.pix_fmt),
+            self.ptr.width,
+            self.ptr.height,
+        )
 
     @property
     def format(self):
         return self._format
 
     @format.setter
-    def format(self, VideoFormat format):
+    def format(self, format: VideoFormat):
         self.ptr.pix_fmt = format.pix_fmt
         self.ptr.width = format.width
         self.ptr.height = format.height
         self._build_format()  # Kinda wasteful.
 
     @property
     def width(self):
-        if self.ptr is NULL:
+        if self.ptr is cython.NULL:
             return 0
         return self.ptr.width
 
     @width.setter
-    def width(self, unsigned int value):
+    def width(self, value: cython.uint):
         self.ptr.width = value
         self._build_format()
 
     @property
     def height(self):
-        if self.ptr is NULL:
+        if self.ptr is cython.NULL:
             return 0
         return self.ptr.height
 
     @height.setter
-    def height(self, unsigned int value):
+    def height(self, value: cython.uint):
         self.ptr.height = value
         self._build_format()
 
@@ -160,13 +180,13 @@ def bits_per_coded_sample(self):
         The number of bits per sample in the codedwords. It's mandatory for this to be set for some formats to decode properly.
 
         Wraps :ffmpeg:`AVCodecContext.bits_per_coded_sample`.
-        
+
         :type: int
         """
         return self.ptr.bits_per_coded_sample
-      
+
     @bits_per_coded_sample.setter
-    def bits_per_coded_sample(self, int value):
+    def bits_per_coded_sample(self, value: cython.int):
         if self.is_encoder:
             raise ValueError("Not supported for encoders")
 
@@ -194,11 +214,11 @@ def framerate(self):
 
         :type: fractions.Fraction
         """
-        return avrational_to_fraction(&self.ptr.framerate)
+        return avrational_to_fraction(cython.address(self.ptr.framerate))
 
     @framerate.setter
     def framerate(self, value):
-        to_avrational(value, &self.ptr.framerate)
+        to_avrational(value, cython.address(self.ptr.framerate))
 
     @property
     def rate(self):
@@ -213,37 +233,39 @@ def rate(self, value):
     def gop_size(self):
         """
         Sets the number of frames between keyframes. Used only for encoding.
-        
+
         :type: int
         """
         if self.is_decoder:
             raise RuntimeError("Cannot access 'gop_size' as a decoder")
         return self.ptr.gop_size
 
     @gop_size.setter
-    def gop_size(self, int value):
+    def gop_size(self, value: cython.int):
         if self.is_decoder:
             raise RuntimeError("Cannot access 'gop_size' as a decoder")
         self.ptr.gop_size = value
 
     @property
     def sample_aspect_ratio(self):
-        return avrational_to_fraction(&self.ptr.sample_aspect_ratio)
+        return avrational_to_fraction(cython.address(self.ptr.sample_aspect_ratio))
 
     @sample_aspect_ratio.setter
     def sample_aspect_ratio(self, value):
-        to_avrational(value, &self.ptr.sample_aspect_ratio)
+        to_avrational(value, cython.address(self.ptr.sample_aspect_ratio))
 
     @property
     def display_aspect_ratio(self):
-        cdef lib.AVRational dar
-
+        dar: lib.AVRational
         lib.av_reduce(
-            &dar.num, &dar.den,
+            cython.address(dar.num),
+            cython.address(dar.den),
             self.ptr.width * self.ptr.sample_aspect_ratio.num,
-            self.ptr.height * self.ptr.sample_aspect_ratio.den, 1024*1024)
+            self.ptr.height * self.ptr.sample_aspect_ratio.den,
+            1024 * 1024,
+        )
 
-        return avrational_to_fraction(&dar)
+        return avrational_to_fraction(cython.address(dar))
 
     @property
     def has_b_frames(self):