Skip to content

Make codec/hwaccel pure #2139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
WyattBlue merged 1 commit into from
Jan 31, 2026
+50 −33
Merged

Make codec/hwaccel pure #2139

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 50 additions & 33 deletions av/codec/hwaccel.pyx → av/codec/hwaccel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -29,42 +29,49 @@ 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[<int>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[<int>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):
return (
f"<av.{self.__class__.__name__} "
f"device_type={lib.av_hwdevice_get_type_name(self.device_type)} "
f"format={self.format.name if self.format else None} "
f"is_supported={self.is_supported} at 0x{<int>self.ptr:x}>"
f"is_supported={self.is_supported} at 0x{cython.cast(int, self.ptr):x}>"
)

@property
Expand All @@ -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):
Expand All @@ -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
Expand All @@ -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")
Expand All @@ -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))
Loading