test_blink_webrtc_sans_aud.py
Rama main · réplica de lectura del repositorio original.
"""Issue #18 : les Blink Mini 2 / Mini 2K+ restaient sur « Réveil de la
caméra… » en WebRTC, alors que MSE (découpage par ffmpeg) les affichait.
_PisteH264 ne fermait une unité d'accès que sur un AUD (NAL 9), émis par les
Mini d'origine : sans AUD, aucune image complète n'atteignait le navigateur.
Le découpage suit désormais ITU-T H.264 §7.4.1.2.3."""
import asyncio
import subprocess
import tempfile
import unittest
from pathlib import Path
import blink_ts_demux
import blink_webrtc
from test_webrtc_roundtrip import trouver_ffmpeg
def _nal(entete: int, charge: bytes = b"\x00") -> bytes:
return b"\x00\x00\x00\x01" + bytes([entete]) + charge
AUD = _nal(0x09, b"\xf0")
SEI = _nal(0x06, b"\x05")
SPS = _nal(0x67, b"\x64\x00\x28")
PPS = _nal(0x68, b"\xee")
# Premier bit après l'en-tête = first_mb_in_slice ue(v) : 1 -> vaut 0.
IDR = _nal(0x65, b"\x88\x84")
IDR_SUITE = _nal(0x65, b"\x1a\x84") # first_mb_in_slice > 0 : même image
P = _nal(0x41, b"\x9a\x02")
P_SUITE = _nal(0x41, b"\x12\x02")
def decouper_annexb(flux: bytes) -> list:
"""NAL units d'un flux Annexe B brut, start code de 4 octets remis."""
positions = []
i = flux.find(b"\x00\x00\x01")
while i != -1:
positions.append(i)
i = flux.find(b"\x00\x00\x01", i + 3)
nals = []
for n, debut in enumerate(positions):
fin = positions[n + 1] if n + 1 < len(positions) else len(flux)
if fin < len(flux) and flux[fin - 1] == 0:
fin -= 1
nals.append(b"\x00\x00\x00\x01" + flux[debut + 3:fin])
return nals
class TestsDebutUniteAcces(unittest.TestCase):
def test_nal_non_vcl_ouvrant_une_unite(self):
for nal in (AUD, SEI, SPS, PPS, _nal(0x0E), _nal(0x12)):
with self.subTest(type_=blink_ts_demux.type_nal(nal)):
self.assertTrue(blink_ts_demux.debut_unite_acces(nal))
def test_tranche_selon_first_mb_in_slice(self):
self.assertTrue(blink_ts_demux.debut_unite_acces(IDR))
self.assertTrue(blink_ts_demux.debut_unite_acces(P))
self.assertFalse(blink_ts_demux.debut_unite_acces(IDR_SUITE))
self.assertFalse(blink_ts_demux.debut_unite_acces(P_SUITE))
def test_start_code_trois_octets_et_nal_tronquee(self):
self.assertTrue(blink_ts_demux.debut_unite_acces(b"\x00\x00\x01\x65\x88"))
self.assertFalse(blink_ts_demux.debut_unite_acces(b"\x00\x00\x01\x65"))
def test_autres_types_n_ouvrent_pas_d_unite(self):
for entete in (0x0A, 0x0B, 0x0C, 0x13): # fin de séquence/flux, remplissage, auxiliaire
with self.subTest(entete=entete):
self.assertFalse(blink_ts_demux.debut_unite_acces(_nal(entete)))
@unittest.skipUnless(blink_webrtc.DISPONIBLE, "aiortc optionnel non installé")
class TestsDecoupageSansAUD(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.track = blink_webrtc._PisteH264(asyncio.StreamReader())
self.track._synchroniser_enregistrement = lambda _unite: None
async def asyncTearDown(self):
self.track.fermer()
with self.assertRaises(asyncio.CancelledError):
await self.track._tache
def _unites(self) -> list:
unites = []
while not self.track._file.empty():
item = self.track._file.get_nowait()
if item is not None:
unites.append(item)
return unites
def _alimenter(self, nals_pts) -> list:
for pts, nal in nals_pts:
self.track._traiter_nal(pts, nal)
self.track._flush_unite_courante()
return self._unites()
async def test_flux_sans_aud_decoupe_par_image_multi_tranches(self):
unites = self._alimenter([
(0, SPS), (0, PPS), (0, IDR), (0, IDR_SUITE),
(3000, P), (3000, P_SUITE),
(6000, SEI), (6000, P),
])
self.assertEqual(unites, [
(0, SPS + PPS + IDR + IDR_SUITE),
(3000, P + P_SUITE),
(6000, SEI + P),
])
self.assertEqual(self.track.sps_pps, SPS + PPS)
self.assertTrue(self.track.sps_pps_pret.is_set())
async def test_flux_avec_aud_inchange(self):
unites = self._alimenter([
(0, AUD), (0, SEI), (0, SPS), (0, PPS), (0, IDR), (0, IDR_SUITE),
(3000, AUD), (3000, P), (3000, P_SUITE),
])
self.assertEqual(unites, [
(0, AUD + SEI + SPS + PPS + IDR + IDR_SUITE),
(3000, AUD + P + P_SUITE),
])
async def test_pts_de_l_unite_est_celui_de_sa_premiere_nal(self):
unites = self._alimenter([(None, SPS), (0, PPS), (0, IDR), (3000, P)])
self.assertEqual([pts for pts, _ in unites], [None, 3000])
async def test_unite_jamais_close_est_plafonnee(self):
enorme = _nal(0x65, b"\x1a" + bytes(1024))
self.track._traiter_nal(0, IDR)
with self.assertRaisesRegex(RuntimeError, "sans fin détectable"):
for _ in range(blink_webrtc.FILE_IMAGES_MAX_OCTETS // len(enorme) + 2):
self.track._traiter_nal(0, enorme)
async def test_vrai_flux_libx264_sans_aud_decode_image_par_image(self):
ffmpeg = trouver_ffmpeg()
if ffmpeg is None:
self.skipTest("ffmpeg non installé")
images = 12
with tempfile.TemporaryDirectory(prefix="blink-sans-aud-") as dossier:
sortie = Path(dossier) / "flux.h264"
# x264 n'émet pas d'AUD par défaut (aud=0) ; quatre tranches par
# image vérifient qu'une image n'est pas coupée entre ses tranches.
subprocess.run([
ffmpeg, "-hide_banner", "-loglevel", "error", "-nostdin",
"-f", "lavfi", "-i", f"testsrc=size=320x240:rate=10,format=yuv420p",
"-frames:v", str(images), "-c:v", "libx264", "-profile:v", "high",
"-x264-params", "aud=0:slices=4", "-bf", "0", "-f", "h264", str(sortie),
], check=True, timeout=60)
nals = decouper_annexb(sortie.read_bytes())
self.assertNotIn(9, {blink_ts_demux.type_nal(nal) for nal in nals})
unites = self._alimenter([(None, nal) for nal in nals])
self.assertEqual(len(unites), images)
# Chaque unité doit être une image complète et décodable à elle seule
# (après les précédentes), comme l'exige le paquétiseur RTP d'aiortc.
decodeur = blink_webrtc.av.CodecContext.create("h264", "r")
decodees = 0
for _pts, unite in unites:
decodees += len(decodeur.decode(blink_webrtc.av.Packet(unite)))
decodees += len(decodeur.decode(None))
self.assertEqual(decodees, images)
if __name__ == "__main__":
unittest.main()