env/kimodo/lib/python3.10/site-packages/av/error.py

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

import cython
from cython.cimports import libav as lib
from cython.cimports.av.logging import get_last_error
from cython.cimports.libc.stdio import fprintf, stderr
from cython.cimports.libc.stdlib import free, malloc

# Will get extended with all the exceptions.
__all__ = [
    "ErrorType",
    "FFmpegError",
    "LookupError",
    "HTTPError",
    "HTTPClientError",
    "UndefinedError",
]
sentinel = cython.declare(object, object())


@cython.ccall
def code_to_tag(code: cython.int) -> bytes:
    """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,
        )
    )


@cython.ccall
def tag_to_code(tag: bytes) -> cython.int:
    """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().__init__(*args)
        self.args = tuple(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.
c_PYAV_STASHED_ERROR = cython.declare(cython.int, tag_to_code(b"PyAV"))
PYAV_STASHED_ERROR_message = cython.declare(str, "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).
# tuple[str, int, str | None, Exception | none]
# fmt: off
_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),
)
# fmt: on


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)


@cython.cclass
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.
    """

    name = cython.declare(str, visibility="readonly")
    value = cython.declare(cython.int, visibility="readonly")

    def __cinit__(self, sentinel_, name: str, value: cython.int, 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)

classes: dict = {}


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

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

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

    return cls


_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,))
_extend_builtin("MemoryError", (errno.ENOMEM,))
_extend_builtin("NotImplementedError", (errno.ENOSYS,))
_extend_builtin("OverflowError", (errno.ERANGE,))
_extend_builtin("OSError", [code for code in errno.errorcode if code not in classes])


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

        return msg


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

    pass


classes[errno.EINVAL] = ArgumentError
globals()["ArgumentError"] = ArgumentError
__all__.append("ArgumentError")


# 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.
_local = cython.declare(object, local())
_err_count = cython.declare(cython.int, 0)
_last_log_count = cython.declare(cython.int, 0)


@cython.cfunc
def stash_exception(exc_info=None) -> cython.int:
    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


@cython.ccall
@cython.exceptval(-1, check=False)
def err_check(res: cython.int, filename=None) -> cython.int:
    """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[1].with_traceback(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

    code: cython.int = -res
    error_buffer: cython.p_char = cython.cast(
        cython.p_char, malloc(lib.AV_ERROR_MAX_STRING_SIZE * cython.sizeof(char))
    )
    if error_buffer == cython.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)