From 5e12029c21519c73293b2ec840b5adc8d04a8ffe Mon Sep 17 00:00:00 2001 From: WyattBlue Date: Sat, 31 Jan 2026 13:39:04 -0500 Subject: [PATCH] Make codec/hwaccel pure --- av/codec/{hwaccel.pyx => hwaccel.py} | 83 +++++++++++++++++----------- 1 file changed, 50 insertions(+), 33 deletions(-) rename av/codec/{hwaccel.pyx => hwaccel.py} (67%) diff --git a/av/codec/hwaccel.pyx b/av/codec/hwaccel.py similarity index 67% rename from av/codec/hwaccel.pyx rename to av/codec/hwaccel.py index e9a6bd194..6508a5867 100644 --- a/av/codec/hwaccel.pyx +++ b/av/codec/hwaccel.py @@ -1,12 +1,12 @@ import weakref from enum import IntEnum -cimport libav as lib - -from av.codec.codec cimport Codec -from av.dictionary cimport _Dictionary -from av.error cimport err_check -from av.video.format cimport get_video_format +import cython +import cython.cimports.libav as lib +from cython.cimports.av.codec.codec import Codec +from cython.cimports.av.dictionary import _Dictionary +from cython.cimports.av.error import err_check +from cython.cimports.av.video.format import get_video_format from av.dictionary import Dictionary @@ -29,34 +29,41 @@ class HWDeviceType(IntEnum): ohcodec = 14 # TODO: When ffmpeg major is changed, check this enum. + class HWConfigMethod(IntEnum): none = 0 - hw_device_ctx = lib.AV_CODEC_HW_CONFIG_METHOD_HW_DEVICE_CTX # This is the only one we support. + hw_device_ctx = ( + lib.AV_CODEC_HW_CONFIG_METHOD_HW_DEVICE_CTX + ) # This is the only one we support. hw_frame_ctx = lib.AV_CODEC_HW_CONFIG_METHOD_HW_FRAMES_CTX internal = lib.AV_CODEC_HW_CONFIG_METHOD_INTERNAL ad_hoc = lib.AV_CODEC_HW_CONFIG_METHOD_AD_HOC -cdef object _cinit_sentinel = object() -cdef object _singletons = weakref.WeakValueDictionary() +_cinit_sentinel = cython.declare(object, object()) +_singletons = cython.declare(object, weakref.WeakValueDictionary()) + -cdef HWConfig wrap_hwconfig(lib.AVCodecHWConfig *ptr): +@cython.cfunc +def wrap_hwconfig(ptr: cython.pointer[lib.AVCodecHWConfig]) -> HWConfig: try: - return _singletons[ptr] + return _singletons[cython.cast(cython.int, ptr)] except KeyError: pass - cdef HWConfig config = HWConfig(_cinit_sentinel) + config: HWConfig = HWConfig(_cinit_sentinel) config._init(ptr) - _singletons[ptr] = config + _singletons[cython.cast(cython.int, ptr)] = config return config -cdef class HWConfig: +@cython.cclass +class HWConfig: def __init__(self, sentinel): if sentinel is not _cinit_sentinel: raise RuntimeError("Cannot instantiate CodecContext") - cdef void _init(self, lib.AVCodecHWConfig *ptr): + @cython.cfunc + def _init(self, ptr: cython.pointer[lib.AVCodecHWConfig]) -> cython.void: self.ptr = ptr def __repr__(self): @@ -64,7 +71,7 @@ def __repr__(self): f"self.ptr:x}>" + f"is_supported={self.is_supported} at 0x{cython.cast(int, self.ptr):x}>" ) @property @@ -84,21 +91,28 @@ def is_supported(self): return bool(self.ptr.methods & lib.AV_CODEC_HW_CONFIG_METHOD_HW_DEVICE_CTX) -cpdef hwdevices_available(): - result = [] - - cdef lib.AVHWDeviceType x = lib.AV_HWDEVICE_TYPE_NONE +@cython.ccall +def hwdevices_available(): + result: list = [] + x: lib.AVHWDeviceType = lib.AV_HWDEVICE_TYPE_NONE while True: x = lib.av_hwdevice_iterate_types(x) if x == lib.AV_HWDEVICE_TYPE_NONE: break result.append(lib.av_hwdevice_get_type_name(HWDeviceType(x))) - return result -cdef class HWAccel: - def __init__(self, device_type, device=None, allow_software_fallback=True, options=None, flags=None): +@cython.cclass +class HWAccel: + def __init__( + self, + device_type, + device=None, + allow_software_fallback=True, + options=None, + flags=None, + ): if isinstance(device_type, HWDeviceType): self._device_type = device_type elif isinstance(device_type, str): @@ -112,11 +126,11 @@ def __init__(self, device_type, device=None, allow_software_fallback=True, optio self.allow_software_fallback = allow_software_fallback self.options = {} if not options else dict(options) self.flags = 0 if not flags else flags - self.ptr = NULL + self.ptr = cython.NULL self.config = None - def _initialize_hw_context(self, Codec codec not None): - cdef HWConfig config + def _initialize_hw_context(self, codec: Codec): + config: HWConfig for config in codec.hardware_configs: if not (config.ptr.methods & lib.AV_CODEC_HW_CONFIG_METHOD_HW_DEVICE_CTX): continue @@ -127,20 +141,23 @@ def _initialize_hw_context(self, Codec codec not None): raise NotImplementedError(f"No supported hardware config for {codec}") self.config = config - - cdef char *c_device = NULL + c_device: cython.p_char = cython.NULL if self._device: device_bytes = self._device.encode() c_device = device_bytes - cdef _Dictionary c_options = Dictionary(self.options) + c_options: _Dictionary = Dictionary(self.options) err_check( lib.av_hwdevice_ctx_create( - &self.ptr, config.ptr.device_type, c_device, c_options.ptr, self.flags + cython.address(self.ptr), + config.ptr.device_type, + c_device, + c_options.ptr, + self.flags, ) ) - def create(self, Codec codec not None): + def create(self, codec: Codec): """Create a new hardware accelerator context with the given codec""" if self.ptr: raise RuntimeError("Hardware context already initialized") @@ -149,11 +166,11 @@ def create(self, Codec codec not None): device_type=self._device_type, device=self._device, allow_software_fallback=self.allow_software_fallback, - options=self.options + options=self.options, ) ret._initialize_hw_context(codec) return ret def __dealloc__(self): if self.ptr: - lib.av_buffer_unref(&self.ptr) + lib.av_buffer_unref(cython.address(self.ptr))