Skip to content

Protocol API

fpga_verification.protocols.avalon_st.intel_video

IntelVIPFrameCodec

Pure adapters between explicit frame data and Intel VIP packets.

Source code in src/fpga_verification/protocols/avalon_st/intel_video/codec.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class IntelVIPFrameCodec:
    """Pure adapters between explicit frame data and Intel VIP packets."""

    def __init__(self, fmt):
        if not isinstance(fmt, VideoFormat):
            raise TypeError("fmt must be a VideoFormat")
        self.fmt = fmt
        self.payload_codec = VideoPayloadCodec(fmt)

    @property
    def symbols_per_beat(self):
        return self.fmt.samples_per_beat

    def control_packet(
        self,
        size,
        interlacing=VIPInterlacing.PROGRESSIVE_FRAME,
    ):
        size = _require_frame_size(size)
        return VIPControlPacket(
            width=size.width,
            height=size.height,
            interlacing=interlacing,
        )

    def frame_to_video_packet(self, frame, size):
        size = _require_frame_size(size)
        return VIPVideoPacket(
            self.payload_codec.frame_to_symbols(frame, size)
        )

    def video_packet_to_frame(self, packet, size):
        size = _require_frame_size(size)
        if not isinstance(packet, VIPVideoPacket):
            raise TypeError("packet must be a VIPVideoPacket")
        return self.payload_codec.symbols_to_frame(packet.payload, size)

    def validate_control_packet(self, packet, size):
        size = _require_frame_size(size)
        if not isinstance(packet, VIPControlPacket):
            raise TypeError("packet must be a VIPControlPacket")
        if packet.width != size.width or packet.height != size.height:
            raise ValueError(
                "VIP control resolution mismatch: "
                f"got {packet.width}x{packet.height}, "
                f"expected {size.width}x{size.height}"
            )
        return True

    def frame_to_vip(
        self,
        frame,
        size,
        interlacing=VIPInterlacing.PROGRESSIVE_FRAME,
        user_packets=None,
    ):
        size = _require_frame_size(size)
        video_packet = self.frame_to_video_packet(frame, size)
        return VIPFrame(
            width=size.width,
            height=size.height,
            pixels=list(video_packet.payload),
            interlacing=interlacing,
            user_packets=list(user_packets or []),
        )

    encode_frame = frame_to_vip

    def vip_to_frame(self, vip_frame, size):
        _require_frame_size(size)
        if not isinstance(vip_frame, VIPFrame):
            raise TypeError("vip_frame must be a VIPFrame")
        return self.video_packet_to_frame(vip_frame.video_packet(), size)

    decode_frame = vip_to_frame

    def frame_to_packets(
        self,
        frame,
        size,
        interlacing=VIPInterlacing.PROGRESSIVE_FRAME,
        user_packets=None,
    ):
        size = _require_frame_size(size)
        return [
            *list(user_packets or []),
            self.control_packet(size, interlacing),
            self.frame_to_video_packet(frame, size),
        ]

    def packets_to_frame(self, packets, size):
        size = _require_frame_size(size)
        packets = list(packets)
        for packet in packets:
            if not isinstance(packet, VIPPacket):
                raise TypeError("packets must contain VIPPacket instances")

        video_packets = [
            packet for packet in packets if isinstance(packet, VIPVideoPacket)
        ]
        if len(video_packets) != 1:
            raise ValueError(
                f"expected exactly one VIP video packet, got {len(video_packets)}"
            )
        return self.video_packet_to_frame(video_packets[0], size)

    def frame_to_packet_symbols(self, frame, size, **kwargs):
        return [
            packet.to_symbols(self.symbols_per_beat)
            for packet in self.frame_to_packets(frame, size, **kwargs)
        ]

    def packet_symbols_to_frame(self, packet_symbols, size):
        packets = [
            vip_packet_from_symbols(symbols, self.symbols_per_beat)
            for symbols in packet_symbols
        ]
        return self.packets_to_frame(packets, size)

VIPProtocolChecker

Validate control/video ordering without converting packets to frames.

The checker is intentionally separate from the stateless frame codec. One checker instance represents one observed packet stream.

Source code in src/fpga_verification/protocols/avalon_st/intel_video/checker.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class VIPProtocolChecker:
    """Validate control/video ordering without converting packets to frames.

    The checker is intentionally separate from the stateless frame codec. One
    checker instance represents one observed packet stream.
    """

    def __init__(self, fmt):
        if not isinstance(fmt, VideoFormat):
            raise TypeError("fmt must be a VideoFormat")
        self.fmt = fmt
        self._control_size = None
        self.check_video_packet_size = False
        self.log = logging.getLogger("cocotb.fpga_verification.vip_protocol_checker")

    @property
    def control_size(self):
        return self._control_size

    def reset(self):
        self._control_size = None

    def expected_video_symbols(self, size=None):
        size = self._control_size if size is None else size
        if not isinstance(size, FrameSize):
            raise TypeError("size must be a FrameSize")
        return self.fmt.frame_symbol_count(size)

    def observe(self, packet):
        if not isinstance(packet, VIPPacket):
            raise TypeError("packet must be a VIPPacket")

        if isinstance(packet, VIPControlPacket):
            try:
                self._control_size = FrameSize(packet.width, packet.height)
            except (TypeError, ValueError) as exc:
                raise VIPProtocolError(
                    "VIP control packet must declare a positive frame size"
                ) from exc
            return packet

        if isinstance(packet, VIPVideoPacket):
            if self._control_size is None:
                raise VIPProtocolError(
                    "VIP video packet received before any control packet"
                )

            actual = len(packet.payload)
            expected = self.expected_video_symbols()
            if actual != expected:
                size = self._control_size
                msg = "VIP video payload does not match the active control " \
                        f"resolution {size.width}x{size.height}: " \
                        f"got {actual} symbols, expected {expected}. " \
                        "A new control packet is required before a frame-size " \
                        "change."
                if self.check_video_packet_size:
                    raise VIPProtocolError(msg)
                else:
                    self.log.warning(msg)
        return packet

VIPProtocolError

Bases: ValueError

Intel VIP packet ordering or declared-frame-size violation.

Source code in src/fpga_verification/protocols/avalon_st/intel_video/checker.py 
23
24
class VIPProtocolError(ValueError):
    """Intel VIP packet ordering or declared-frame-size violation."""