Fix #2161 null check for get_media_type_str

av/codec/codec.py

@@ -172,7 +172,8 @@ def type(self):
         E.g: ``'audio'``, ``'video'``, ``'subtitle'``.
 
         """
-        return lib.av_get_media_type_string(self.ptr.type)
+        media_type = lib.av_get_media_type_string(self.ptr.type)
+        return "unknown" if media_type == cython.NULL else media_type
 
     @property
     def id(self):

av/codec/codec.pyi

@@ -61,7 +61,9 @@ class Codec:
     @property
     def long_name(self) -> str: ...
     @property
-    def type(self) -> Literal["video", "audio", "data", "subtitle", "attachment"]: ...
+    def type(
+        self,
+    ) -> Literal["video", "audio", "data", "subtitle", "attachment", "unknown"]: ...
     @property
     def id(self) -> int: ...
     frame_rates: list[Fraction] | None

av/stream.py

@@ -275,7 +275,8 @@ def type(self):
 
         :type: Literal["audio", "video", "subtitle", "data", "attachment"]
         """
-        return lib.av_get_media_type_string(self.ptr.codecpar.codec_type)
+        media_type = lib.av_get_media_type_string(self.ptr.codecpar.codec_type)
+        return "unknown" if media_type == cython.NULL else media_type
 
 
 @cython.cclass

av/stream.pyi

@@ -48,7 +48,7 @@ class Stream:
     disposition: Disposition
     frames: int
     language: str | None
-    type: Literal["video", "audio", "data", "subtitle", "attachment"]
+    type: Literal["video", "audio", "data", "subtitle", "attachment", "unknown"]
 
     # From context
     codec_tag: str

tests/test_streams.py

@@ -1,4 +1,6 @@
+import io
 import os
+import struct
 
 import pytest
 
@@ -8,6 +10,63 @@
 from .common import fate_suite
 
 
+def _crc32_mpeg(data: bytes) -> int:
+    crc = 0xFFFFFFFF
+    for byte in data:
+        crc ^= byte << 24
+        for _ in range(8):
+            if crc & 0x80000000:
+                crc = (crc << 1) ^ 0x04C11DB7
+            else:
+                crc <<= 1
+            crc &= 0xFFFFFFFF
+    return crc
+
+
+def _make_ts_packet(pid: int, payload: bytes, pusi: bool = False, cc: int = 0) -> bytes:
+    pid_hi = (pid >> 8) & 0x1F
+    pid_lo = pid & 0xFF
+    if pusi:
+        pid_hi |= 0x40
+    header = bytes([0x47, pid_hi, pid_lo, 0x10 | (cc & 0x0F)])
+    pkt = header + payload
+    return (pkt + b"\xff" * (188 - len(pkt)))[:188]
+
+
+def _make_unknown_stream_ts() -> bytes:
+    """Build a minimal MPEG-TS byte string whose only stream has AVMEDIA_TYPE_UNKNOWN.
+
+    Uses stream_type=0x82 in the PMT, which FFmpeg does not map to any known
+    media type, resulting in ``stream.type == "unknown"``.
+    """
+    pat_data = bytes(
+        [0x00, 0xB0, 0x0D, 0x00, 0x01, 0xC1, 0x00, 0x00, 0x00, 0x01, 0xE1, 0x00]
+    )
+    pat_data += struct.pack(">I", _crc32_mpeg(pat_data))
+    pat_payload = bytes([0x00]) + pat_data
+
+    # fmt: off
+    pmt_data = bytes([0x02, 0xB0, 0x12, 0x00, 0x01, 0xC1, 0x00, 0x00,
+                      0xE1, 0x02, 0xF0, 0x00, 0x82, 0xE1, 0x02, 0xF0, 0x00,
+    ])
+    # fmt: on
+    pmt_data += struct.pack(">I", _crc32_mpeg(pmt_data))
+    pmt_payload = bytes([0x00]) + pmt_data
+
+    pes_header = bytes([0x00, 0x00, 0x01, 0xBD, 0x00, 0x0A, 0x80, 0x00, 0x00])
+    pes_data = pes_header + b"\xaa" * (184 - len(pes_header))
+
+    packets: list[bytes] = []
+    for i in range(2):
+        packets.append(_make_ts_packet(0x0000, pat_payload, pusi=True, cc=i))
+    for i in range(2):
+        packets.append(_make_ts_packet(0x0100, pmt_payload, pusi=True, cc=i))
+    for i in range(200):
+        packets.append(_make_ts_packet(0x0102, pes_data, pusi=(i % 10 == 0), cc=i % 16))
+
+    return b"".join(packets)
+
+
 class TestStreams:
     @pytest.fixture(autouse=True)
     def cleanup(self):
@@ -300,3 +359,12 @@ def test_attachment_stream(self) -> None:
             assert att.name == "attachment.txt"
             assert att.mimetype == "text/plain"
             assert att.data == b"hello\n"
+
+    def test_unknown_stream_type(self) -> None:
+        ts_data = _make_unknown_stream_ts()
+
+        with av.open(io.BytesIO(ts_data), format="mpegts") as container:
+            assert len(container.streams) == 1
+            stream = container.streams[0]
+            assert stream.type == "unknown"
+            assert type(stream) is av.stream.Stream