lib/python3.10/site-packages/av/error.pyx

cimport libav as lib
from libc.stdio cimport fprintf, stderr
from libc.stdlib cimport free, malloc

from av.logging cimport get_last_error

import errno
import os
import sys
import traceback
from threading import local

# Will get extended with all of the exceptions.
__all__ = [
    "ErrorType", "FFmpegError", "LookupError", "HTTPError", "HTTPClientError",
    "UndefinedError",
]


cpdef code_to_tag(int code):
    """Convert an integer error code into 4-byte tag.

    >>> code_to_tag(1953719668)
    b'test'

    """
    return bytes((
        code & 0xff,
        (code >> 8) & 0xff,
        (code >> 16) & 0xff,
        (code >> 24) & 0xff,
    ))

cpdef tag_to_code(bytes tag):
    """Convert a 4-byte error tag into an integer code.

    >>> tag_to_code(b'test')
    1953719668

    """
    if len(tag) != 4:
        raise ValueError("Error tags are 4 bytes.")
    return (
        (tag[0]) +
        (tag[1] << 8) +
        (tag[2] << 16) +
        (tag[3] << 24)
    )


class FFmpegError(Exception):
    """Exception class for errors from within FFmpeg.

    .. attribute:: errno

        FFmpeg's integer error code.

    .. attribute:: strerror

        FFmpeg's error message.

    .. attribute:: filename

        The filename that was being operated on (if available).

    .. attribute:: log

        The tuple from :func:`av.logging.get_last_log`, or ``None``.

    """

    def __init__(self, code, message, filename=None, log=None):
        self.errno = code
        self.strerror = message

        args = [code, message]
        if filename or log:
            args.append(filename)
            if log:
                args.append(log)
        super(FFmpegError, self).__init__(*args)
        self.args = tuple(args)  # FileNotFoundError/etc. only pulls 2 args.

    @property
    def filename(self):
        try:
            return self.args[2]
        except IndexError:
            pass

    @property
    def log(self):
        try:
            return self.args[3]
        except IndexError:
            pass

    def __str__(self):
        msg = ""
        if self.errno is not None:
            msg = f"{msg}[Errno {self.errno}] "
        if self.strerror is not None:
            msg = f"{msg}{self.strerror}"
        if self.filename:
            msg = f"{msg}: {self.filename!r}"
        if self.log:
            msg = f"{msg}; last error log: [{self.log[1].strip()}] {self.log[2].strip()}"

        return msg


# Our custom error, used in callbacks.
cdef int c_PYAV_STASHED_ERROR = tag_to_code(b"PyAV")
cdef str PYAV_STASHED_ERROR_message = "Error in PyAV callback"


# Bases for the FFmpeg-based exceptions.
class LookupError(FFmpegError, LookupError):
    pass


class HTTPError(FFmpegError):
    pass


class HTTPClientError(FFmpegError):
    pass


# Tuples of (enum_name, enum_value, exc_name, exc_base).
_ffmpeg_specs = (
    ("BSF_NOT_FOUND", -lib.AVERROR_BSF_NOT_FOUND, "BSFNotFoundError", LookupError),
    ("BUG", -lib.AVERROR_BUG, None, RuntimeError),
    ("BUFFER_TOO_SMALL", -lib.AVERROR_BUFFER_TOO_SMALL, None, ValueError),
    ("DECODER_NOT_FOUND", -lib.AVERROR_DECODER_NOT_FOUND, None, LookupError),
    ("DEMUXER_NOT_FOUND", -lib.AVERROR_DEMUXER_NOT_FOUND, None, LookupError),
    ("ENCODER_NOT_FOUND", -lib.AVERROR_ENCODER_NOT_FOUND, None, LookupError),
    ("EOF", -lib.AVERROR_EOF, "EOFError", EOFError),
    ("EXIT", -lib.AVERROR_EXIT, None, None),
    ("EXTERNAL", -lib.AVERROR_EXTERNAL, None, None),
    ("FILTER_NOT_FOUND", -lib.AVERROR_FILTER_NOT_FOUND, None, LookupError),
    ("INVALIDDATA", -lib.AVERROR_INVALIDDATA, "InvalidDataError", ValueError),
    ("MUXER_NOT_FOUND", -lib.AVERROR_MUXER_NOT_FOUND, None, LookupError),
    ("OPTION_NOT_FOUND", -lib.AVERROR_OPTION_NOT_FOUND, None, LookupError),
    ("PATCHWELCOME", -lib.AVERROR_PATCHWELCOME, "PatchWelcomeError", None),
    ("PROTOCOL_NOT_FOUND", -lib.AVERROR_PROTOCOL_NOT_FOUND, None, LookupError),
    ("UNKNOWN", -lib.AVERROR_UNKNOWN, None, None),
    ("EXPERIMENTAL", -lib.AVERROR_EXPERIMENTAL, None, None),
    ("INPUT_CHANGED", -lib.AVERROR_INPUT_CHANGED, None, None),
    ("OUTPUT_CHANGED", -lib.AVERROR_OUTPUT_CHANGED, None, None),
    ("HTTP_BAD_REQUEST", -lib.AVERROR_HTTP_BAD_REQUEST, "HTTPBadRequestError", HTTPClientError),
    ("HTTP_UNAUTHORIZED", -lib.AVERROR_HTTP_UNAUTHORIZED, "HTTPUnauthorizedError", HTTPClientError),
    ("HTTP_FORBIDDEN", -lib.AVERROR_HTTP_FORBIDDEN, "HTTPForbiddenError", HTTPClientError),
    ("HTTP_NOT_FOUND", -lib.AVERROR_HTTP_NOT_FOUND, "HTTPNotFoundError", HTTPClientError),
    ("HTTP_OTHER_4XX", -lib.AVERROR_HTTP_OTHER_4XX, "HTTPOtherClientError", HTTPClientError),
    ("HTTP_SERVER_ERROR", -lib.AVERROR_HTTP_SERVER_ERROR, "HTTPServerError", HTTPError),
    ("PYAV_CALLBACK", c_PYAV_STASHED_ERROR, "PyAVCallbackError", RuntimeError),
)

cdef sentinel = object()


class EnumType(type):
    def __new__(mcl, name, bases, attrs, *args):
        # Just adapting the method signature.
        return super().__new__(mcl, name, bases, attrs)

    def __init__(self, name, bases, attrs, items):
        self._by_name = {}
        self._by_value = {}
        self._all = []

        for spec in items:
            self._create(*spec)

    def _create(self, name, value, doc=None, by_value_only=False):
        # We only have one instance per value.
        try:
            item = self._by_value[value]
        except KeyError:
            item = self(sentinel, name, value, doc)
            self._by_value[value] = item

        return item

    def __len__(self):
        return len(self._all)

    def __iter__(self):
        return iter(self._all)

    def __getitem__(self, key):
        if isinstance(key, str):
            return self._by_name[key]

        if isinstance(key, int):
            try:
                return self._by_value[key]
            except KeyError:
                pass

            raise KeyError(key)

        if isinstance(key, self):
            return key

        raise TypeError(f"{self.__name__} indices must be str, int, or itself")

    def _get(self, long value, bint create=False):
        try:
            return self._by_value[value]
        except KeyError:
            pass

        if not create:
            return

        return self._create(f"{self.__name__.upper()}_{value}", value, by_value_only=True)

    def get(self, key, default=None, create=False):
        try:
            return self[key]
        except KeyError:
            if create:
                return self._get(key, create=True)
            return default


cdef class EnumItem:
    """An enumeration of FFmpeg's error types.

.. attribute:: tag

    The FFmpeg byte tag for the error.

.. attribute:: strerror

    The error message that would be returned.

"""
    cdef readonly str name
    cdef readonly int value

    def __cinit__(self, sentinel_, str name, int value, doc=None):
        if sentinel_ is not sentinel:
            raise RuntimeError(f"Cannot instantiate {self.__class__.__name__}.")

        self.name = name
        self.value = value
        self.__doc__ = doc

    def __repr__(self):
        return f"<{self.__class__.__module__}.{self.__class__.__name__}:{self.name}(0x{self.value:x})>"

    def __str__(self):
        return self.name

    def __int__(self):
        return self.value

    @property
    def tag(self):
        return code_to_tag(self.value)


ErrorType = EnumType("ErrorType", (EnumItem, ), {"__module__": __name__}, [x[:2] for x in _ffmpeg_specs])


for enum in ErrorType:
    # Mimick the errno module.
    globals()[enum.name] = enum
    if enum.value == c_PYAV_STASHED_ERROR:
        enum.strerror = PYAV_STASHED_ERROR_message
    else:
        enum.strerror = lib.av_err2str(-enum.value)


# Mimick the builtin exception types.
# See https://www.python.org/dev/peps/pep-3151/#new-exception-classes
# Use the named ones we have, otherwise default to OSError for anything in errno.

# See this command for the count of POSIX codes used:
#
#    egrep -IR 'AVERROR\(E[A-Z]+\)' vendor/ffmpeg-4.2 |\
#        sed -E 's/.*AVERROR\((E[A-Z]+)\).*/\1/' | \
#        sort | uniq -c
#
# The biggest ones that don't map to PEP 3151 builtins:
#
#    2106 EINVAL -> ValueError
#     649 EIO    -> IOError (if it is distinct from OSError)
#    4080 ENOMEM -> MemoryError
#     340 ENOSYS -> NotImplementedError
#      35 ERANGE -> OverflowError

classes = {}


def _extend_builtin(name, codes):
    base = getattr(__builtins__, name, OSError)
    cls = type(name, (FFmpegError, base), dict(__module__=__name__))

    # Register in builder.
    for code in codes:
        classes[code] = cls

    # Register in module.
    globals()[name] = cls
    __all__.append(name)

    return cls


# PEP 3151 builtins.
_extend_builtin("PermissionError", (errno.EACCES, errno.EPERM))
_extend_builtin("BlockingIOError", (errno.EAGAIN, errno.EALREADY, errno.EINPROGRESS, errno.EWOULDBLOCK))
_extend_builtin("ChildProcessError", (errno.ECHILD, ))
_extend_builtin("ConnectionAbortedError", (errno.ECONNABORTED, ))
_extend_builtin("ConnectionRefusedError", (errno.ECONNREFUSED, ))
_extend_builtin("ConnectionResetError", (errno.ECONNRESET, ))
_extend_builtin("FileExistsError", (errno.EEXIST, ))
_extend_builtin("InterruptedError", (errno.EINTR, ))
_extend_builtin("IsADirectoryError", (errno.EISDIR, ))
_extend_builtin("FileNotFoundError", (errno.ENOENT, ))
_extend_builtin("NotADirectoryError", (errno.ENOTDIR, ))
_extend_builtin("BrokenPipeError", (errno.EPIPE, errno.ESHUTDOWN))
_extend_builtin("ProcessLookupError", (errno.ESRCH, ))
_extend_builtin("TimeoutError", (errno.ETIMEDOUT, ))

# Other obvious ones.
_extend_builtin("ValueError", (errno.EINVAL, ))
_extend_builtin("MemoryError", (errno.ENOMEM, ))
_extend_builtin("NotImplementedError", (errno.ENOSYS, ))
_extend_builtin("OverflowError", (errno.ERANGE, ))

# The rest of them (for now)
_extend_builtin("OSError", [code for code in errno.errorcode if code not in classes])

# Classes for the FFmpeg errors.
for enum_name, code, name, base in _ffmpeg_specs:
    name = name or enum_name.title().replace("_", "") + "Error"

    if base is None:
        bases = (FFmpegError,)
    elif issubclass(base, FFmpegError):
        bases = (base,)
    else:
        bases = (FFmpegError, base)

    cls = type(name, bases, {"__module__": __name__})

    # Register in builder.
    classes[code] = cls

    # Register in module.
    globals()[name] = cls
    __all__.append(name)

del _ffmpeg_specs


# Storage for stashing.
cdef object _local = local()
cdef int _err_count = 0

cdef int stash_exception(exc_info=None):
    global _err_count

    existing = getattr(_local, "exc_info", None)
    if existing is not None:
        fprintf(stderr, "PyAV library exception being dropped:\n")
        traceback.print_exception(*existing)
        _err_count -= 1  # Balance out the +=1 that is coming.

    exc_info = exc_info or sys.exc_info()
    _local.exc_info = exc_info
    if exc_info:
        _err_count += 1

    return -c_PYAV_STASHED_ERROR


cdef int _last_log_count = 0

cpdef int err_check(int res, filename=None) except -1:
    """Raise appropriate exceptions from library return code."""

    global _err_count
    global _last_log_count

    # Check for stashed exceptions.
    if _err_count:
        exc_info = getattr(_local, "exc_info", None)
        if exc_info is not None:
            _err_count -= 1
            _local.exc_info = None
            raise exc_info[0], exc_info[1], exc_info[2]

    if res >= 0:
        return res

    # Grab details from the last log.
    log_count, last_log = get_last_error()
    if log_count > _last_log_count:
        _last_log_count = log_count
        log = last_log
    else:
        log = None

    cdef int code = -res
    cdef char* error_buffer = <char*>malloc(lib.AV_ERROR_MAX_STRING_SIZE * sizeof(char))
    if error_buffer == NULL:
        raise MemoryError()

    try:
        if code == c_PYAV_STASHED_ERROR:
            message = PYAV_STASHED_ERROR_message
        else:
            lib.av_strerror(res, error_buffer, lib.AV_ERROR_MAX_STRING_SIZE)
            # Fallback to OS error string if no message
            message = error_buffer or os.strerror(code)

        cls = classes.get(code, UndefinedError)
        raise cls(code, message, filename, log)
    finally:
        free(error_buffer)


class UndefinedError(FFmpegError):
    """Fallback exception type in case FFmpeg returns an error we don't know about."""
    pass