All/Python/local/camera/decoder/lib.py

from threading import Lock

__all__ = ["b2ToInt", "b4ToInt", "RtpData", "VideoManager", "Decoder"]


def b4ToInt(buffer: "bytes") -> "int":
    return (buffer[0] << 24) + (buffer[1] << 16) + (buffer[2] << 8) + buffer[3]


def b2ToInt(buffer: "bytes") -> "int":
    return (buffer[0] << 8) + buffer[1]


class RtpData:
    def __init__(self, *, data=b"", packType=0, naluType=0, pts=0, seq=0, isFrameEnd=False):
        # packType: 0(first), 1(center), 2(last), naluType: SPS, PPS, SEI...
        self.data, self.packType, self.naluType = data, packType, naluType
        self.pts, self.seq, self.isFrameEnd = pts, seq, isFrameEnd

    def __str__(self):
        return f"<RtpData: packType:{self.packType}, naluType:{self.naluType}, data len:{len(self.data)}>"


class VideoManager:
    def __init__(self):
        self.packs: "list[RtpData]" = []
        self.lock: "Lock" = Lock()
        self.rtpCv = None


class Decoder:
    """
    Reference：https://datatracker.ietf.org/doc/html/rfc7798
    """

    def __init__(self):
        self.Prefix = b"\x00\x00\x00\x01"
        self.Manager: "VideoManager" = VideoManager()

    def parse(self, rtp):
        raise NotImplementedError("parse must be implemented before use")