blink_webrtc.py
Rama main · réplica de lectura del repositorio original.
"""Direct WebRTC : passerelle sans reencodage entre le flux Blink (immis,
via blinkpy) et un navigateur, en alternative a /live-mse (MSE + ffmpeg).
Mesure sur prototype le 2026-09-03 (cf. BACKLOG.md) : premiere image 2.3 a
2.9x plus rapide qu'en MSE, sans le cout CPU d'un reencodage - MSE doit
attendre le SPS pour ecrire un entete MP4 (empty_moov) avant d'emettre quoi
que ce soit, WebRTC negocie les parametres du codec a part (SDP), donc peut
demarrer des la premiere image-cle.
Optionnel par construction : DISPONIBLE reste False si aiortc n'est pas
installe (pip install -r requirements.txt pas encore fait, ou plateforme ou
la roue binaire n'existe pas). serve.py doit retomber sur MSE dans ce cas,
jamais faire echouer son propre import pour ce module.
Ecueils reels rencontres en construisant ceci, tous geres ici :
- PyAV (av.open()) extrait SPS/PPS dans codec_context.extradata au lieu de
les laisser dans le flux de paquets demuxe, et sa sonde (analyzeduration/
probesize) doit rester assez genereuse pour les capturer fiablement -
baisser ce palier a l'aveugle a echoue en reel sur jardin (2026-09-03) :
negociation SDP reussie, mais extradata incomplet, plus aucune image
jamais decodee. D'ou blink_ts_demux.py : un demultiplexeur MPEG-TS/PES/NAL
minimal mais correct (PAT -> PMT -> PID video -> flux elementaire), qui
extrait SPS/PPS des les tout premiers octets, sans la marge de securite
qu'il fallait garder pour la sonde de conteneur de PyAV.
- aiortc n'annonce en SDP que du H.264 Baseline (42001f/42e01f, code en dur
dans aiortc/codecs/__init__.py) alors que les cameras Blink filment en
High (souvent 640028) : le navigateur negocie Baseline, recoit du High,
ignore tout en silence. Aucune solution officielle coté aiortc (issue
aiortc/aiortc#944, fermee sans suite ; la reserve du mainteneur porte sur
l'encodage, pas sur ce cas de passthrough). D'ou _enregistrer_profil_h264,
qui ajoute le vrai profil (lu dans le SPS) a la table de capacites
d'aiortc avant de repondre - un detail d'implementation non documente,
susceptible de casser a une future mise a jour d'aiortc sans avertissement.
- RTCPeerConnection() sans configuration retombe sur un serveur STUN public
par defaut (stun.l.google.com) : setLocalDescription() n'y renvoie
qu'apres la collecte de candidats, un blocage reseau/pare-feu dessus a
laisse MODULE_SLOT/le verrou disque tenus indefiniment en reel
(2026-09-03). D'ou iceServers=[] (navigateur et serveur sur le meme
reseau local, un STUN n'a de toute facon rien a apporter) et les deux
plafonds durs (NEGOCIATION_MAX_SECONDS, SESSION_MAX_SECONDS) plus bas,
memes principes que LIVE_FIRST_FRAME_SECONDS/LIVE_MAX_SECONDS en MSE."""
import asyncio
from contextlib import suppress
import time
from fractions import Fraction
from pathlib import Path
from typing import Awaitable, Callable, Optional
import runtime
try:
import av
from aiortc import (
MediaStreamTrack,
RTCConfiguration,
RTCPeerConnection,
RTCSessionDescription,
)
from aiortc.codecs import CODECS
from aiortc import rtp as _rtp
from aiortc.mediastreams import MediaStreamError
from aiortc.sdp import H264Profile, SessionDescription, parse_h264_profile_level_id
from aiortc.rtcrtpparameters import (
RTCRtcpFeedback,
RTCRtpCodecCapability,
RTCRtpCodecParameters,
)
import blink_ts_demux
DISPONIBLE = True
except ImportError:
DISPONIBLE = False
# Mêmes plafonds durs qu'en MSE (LIVE_FIRST_FRAME_SECONDS/LIVE_MAX_SECONDS,
# serve.py), pour la même raison : sans borne, un blocage réseau ou une
# session jamais fermée proprement retient MODULE_SLOT/le verrou disque
# indéfiniment. NEGOCIATION_MAX_SECONDS reste large : offre/réponse SDP et
# poignée de main DTLS n'ont normalement rien à attendre de long.
NEGOCIATION_MAX_SECONDS = 15
# Distinct de NEGOCIATION_MAX_SECONDS depuis le 2026-09-03 (BACKLOG.md) :
# les deux partageaient jusque-la le meme plafond de 15s pour l'attente du
# SPS/PPS (sps_pps_pret), confondant deux choses sans rapport - la vitesse
# d'une negociation SDP/DTLS purement locale, et le temps que met une
# camera sur batterie a se reveiller. 15s suffit a la premiere, pas
# toujours a la seconde (timeout constate en reel sur jardin) : memes
# raison et valeur que LIVE_FIRST_FRAME_SECONDS cote MSE (serve.py), qui
# accorde deja cette patience pour la meme raison.
PREMIERE_IMAGE_MAX_SECONDS = 40
# Silence en cours de session (apres le premier SPS/PPS, donc une fois
# connectionState deja passe a "connected") : distinct de SESSION_MAX_SECONDS
# (420s), qui borne la duree totale d'un direct sain, pas le temps qu'on
# tolere sans le moindre octet. Sans ce plafond-la, _lire() (ci-dessous)
# reste bloque sur son `await reader.read()` tant que le relais Blink ne
# ferme pas proprement la connexion TCP - jamais garanti (WinError 10054
# "connexion fermee par l'hote distant" vu dans les logs asyncio APRES
# coup, pas au moment du blocage). Sans signal explicite, aiortc ne
# remarque rien de son cote (pas d'etat "disconnected", cf. plus haut) :
# connectionState restait a "connected" jusqu'au seul filet qui restait,
# SESSION_MAX_SECONDS - 5 minutes bloque sur "Reconnexion..." pour
# l'utilisateur (constate en reel, Salon, 2026-09-04). Meme ordre de
# grandeur qu'ATTENTE_HUB_MAX_SECONDS/le timeout de _stop_stream (serve.py,
# 20s chacun) plutot qu'une valeur inventee ici.
SILENCE_FLUX_MAX_SECONDS = 20
# Même valeur et même raison que LIVE_MAX_SECONDS (serve.py) : Blink ferme
# lui-même une session au bout d'environ six minutes de vidéo (mesuré le
# 2026-09-28), sept minutes le laissent aller au bout.
SESSION_MAX_SECONDS = 420
FERMETURE_MAX_SECONDS = 10
# Tampon de lecture (jitter buffer) avant d'emettre la toute premiere image,
# puis chaque image suivante cadencee sur son PTS a partir de cette meme
# ancre. Mesure en reel sur jardin (camera a pile, 2026-09-03, cf.
# BACKLOG.md) : un trou reseau regulier d'environ 1,0s (0,62 a 0,86s mesures
# sur 6 cycles), tres probablement le wifi de la camera qui s'endort par
# intervalles pour economiser la batterie - le PTS camera lui reste
# parfaitement regulier (30 fps sans saut), ce n'est donc pas un probleme
# d'encodage. Sans tampon, recv() renvoyait une image des sa
# demultiplexation : le moindre trou se voyait directement comme un arret de
# lecture (lecture saccadee, signale par l'utilisateur). Cout : un peu de la
# latence gagnee sur MSE, qui reste tres largement devant (jardin
# ~5,6s + 1,2s contre 15,8s en MSE). Applique uniformement (pas seulement
# aux cameras a pile) : plus simple, et Salon a largement la marge pour
# l'absorber sans redevenir plus lent que MSE.
TAMPON_LECTURE_SECONDS = 1.2
# Garde-fou mémoire si le navigateur cesse de consommer les images. Ne pas
# jeter arbitrairement des images H.264 : les suivantes peuvent en dépendre.
FILE_IMAGES_MAX = 900
FILE_IMAGES_MAX_OCTETS = 32 * 1024 * 1024
TAMPON_ENREGISTREMENT_MAX_OCTETS = 8 * 1024 * 1024
if DISPONIBLE:
# Chromium récent utilise aussi 35..63 quand 96..127 est rempli.
# aiortc 1.9 (et certaines versions récentes) ne les considère pas
# dynamiques : sa réponse conserve alors le PT local au lieu de celui
# de l'offre, et le navigateur jette les RTP reçus (zéro image).
# RFC 5761 §4 autorise cette plage, mais interdit 64..95 (RTCP).
_rtp.DYNAMIC_PAYLOAD_TYPES = frozenset(_rtp.DYNAMIC_PAYLOAD_TYPES) | frozenset(range(35, 64))
class _PisteH264(MediaStreamTrack):
"""Piste video sans decodage ni reencodage : lit le flux TCP local
de blinkpy octet par octet, demultiplexe nous-memes (pas de PyAV,
blink_ts_demux.py), regroupe les NAL units par unite d'acces
(AUD, type 9, quand la camera en emet comme les Mini d'origine ;
sinon premiere tranche d'une nouvelle image, cf. _traiter_nal),
les fait passer telles quelles en av.Packet.
recv() peut renvoyer un av.Packet plutot qu'un av.VideoFrame
(documente dans la docstring de MediaStreamTrack.recv elle-meme) :
RTCRtpSender (_next_encoded_frame, rtcrtpsender.py) detecte alors
que ce n'est pas une Frame et appelle encoder.pack() au lieu de
encoder.encode(), qui se contente de decouper le bitstream deja
compresse en paquets RTP. Zero appel a un encodeur. Equivalent
WebRTC du -c:v copy qu'utilise deja /live-mse.
Horodatage : le vrai PTS encode par la camera (extrait de l'entete
PES par blink_ts_demux, deja cadence a 90 kHz - meme horloge que la
video RTP, aucune conversion a faire), pas l'heure d'arrivee. Un
horodatage improvise a la reception a ete essaye puis retire
(2026-09-03) : le reseau livre par rafales, pas a cadence reguliere,
et le recepteur perd alors l'information de rythme dont il a besoin
pour lisser l'affichage - lecture saccadee, constate en reel. Repli
sur l'horloge locale seulement si jamais un PTS venait a manquer
(ne devrait pas arriver : la camera en donne un par image).
recv() cadence aussi sa sortie sur ce PTS, retenue derriere
TAMPON_LECTURE_SECONDS (cf. plus haut) plutot que de renvoyer une
image des qu'elle est demultiplexee : necessaire pour absorber le
trou reseau periodique mesure sur les cameras a pile."""
kind = "video"
def __init__(
self, reader: asyncio.StreamReader, ffmpeg: Optional[str] = None,
chemin_enregistrement: Optional[Callable[[], Path]] = None,
enregistrement_actif: Optional[Callable[[], bool]] = None,
journal: Optional[Callable[[str], None]] = None,
) -> None:
super().__init__()
self._reader = reader
self._demux = blink_ts_demux.DemuxeurTSVideo()
self._t0 = time.monotonic()
self._t_ancrage: Optional[float] = None
self._pts_ancrage: Optional[int] = None
self._unite_courante: list = []
self._unite_courante_octets = 0
self._unite_a_vcl = False
self._pts_courant = None
self._file: asyncio.Queue = asyncio.Queue(maxsize=FILE_IMAGES_MAX)
self._file_octets = 0
self.sps_pps = b""
self.sps_pps_pret = asyncio.Event()
# Posés par negocier() après le SPS/PPS : toute fin du flux doit
# fermer aussi la connexion, car track.stop() ne suffit pas à
# faire transiter connectionState.
self.pc = None
self._demander_fermeture = None
# Enregistrement disque, optionnel et pilote par bouton : voir
# _synchroniser_enregistrement. chemin_enregistrement est une
# fonction (pas un chemin deja resolu) : chaque demarrage doit
# recalculer un horodatage frais, pas reprendre celui du debut
# de session.
self._ffmpeg = ffmpeg
self._chemin_enregistrement = chemin_enregistrement
self._enregistrement_actif = enregistrement_actif
# No-op par defaut : negocier() sans camera identifiee (tests,
# appelants futurs) ne doit jamais planter faute de journal.
self._journal = journal or (lambda _message: None)
self._recorder_process = None
self._recorder_stdin = None
self._recorder_demande = False
self._recorder_demarrage = None
self._recorder_attente = []
self._recorder_attente_octets = 0
self._fermee = False
self._tache = asyncio.ensure_future(self._lire())
async def _lire(self) -> None:
fin_normale = False
try:
while not self._fermee:
# None (pas de plafond) tant que le SPS/PPS initial n'est
# pas arrive : PREMIERE_IMAGE_MAX_SECONDS (40s, negocier())
# accorde deja cette patience-la, plus longue qu'ici a
# dessein (reveil d'une camera sur batterie) - un plafond
# plus court ici y couperait court en silence, avant meme
# que ce delai plus genereux n'ait sa chance.
delai = (
SILENCE_FLUX_MAX_SECONDS
if self.sps_pps_pret.is_set() else None
)
try:
morceau = await asyncio.wait_for(
self._reader.read(65536), timeout=delai
)
except asyncio.TimeoutError:
self._journal(
f"flux Blink silencieux depuis "
f"{SILENCE_FLUX_MAX_SECONDS}s, fermeture"
)
fin_normale = True
break
if not morceau:
fin_normale = True
break
for pts, nal in self._demux.alimenter(morceau):
self._traiter_nal(pts, nal)
except Exception as error:
self._journal(msg("flux_interrompu",
erreur=f"{type(error).__name__}: {error}"))
finally:
# Toutes les fins du flux (EOF, reset, silence, annulation)
# doivent fermer aussi WebRTC ; arrêter la piste seule ne
# change pas connectionState et ne rend pas le module Blink.
try:
if fin_normale:
# Le démultiplexeur garde toujours sa toute dernière
# NAL en tampon, faute d'un prochain start code pour
# en confirmer la fin (qui ne viendra plus) ; sans ce
# flush, elle - et l'unité d'accès en cours qui
# l'attendait, jamais close par un nouvel AUD - étaient
# perdues (bug réel : dernière image d'un direct
# jamais transmise, ni à l'écran ni à l'enregistrement).
for pts, nal in self._demux.finaliser():
self._traiter_nal(pts, nal)
self._flush_unite_courante()
self._arreter_enregistrement()
finally:
# vider=False (fin normale) laisse recv() consommer les
# images déjà en file avant de tomber sur le sentinel de
# fin - sans ça, la fin de flux effaçait tout ce qui
# restait en attente, jamais transmis (bug réel : sur une
# source de 60 images, seules celles déjà lues par le
# navigateur au moment de l'EOF survivaient). Un arrêt
# forcé (fermer(), plus bas) garde le vidage immédiat :
# l'appelant ne va de toute façon plus lire recv().
self._terminer_file(vider=not fin_normale)
# Une fin normale laisse recv() transmettre les images
# déjà reçues avant de fermer la connexion.
if not fin_normale and self._demander_fermeture is not None:
self._demander_fermeture()
def _traiter_nal(self, pts, nal: bytes) -> None:
# Début d'unité d'accès selon ITU-T H.264 §7.4.1.2.3, pas
# seulement l'AUD : les Blink Mini d'origine en émettent un par
# image, les Mini 2 / Mini 2K+ apparemment pas (issue #18). Sans
# AUD, rien n'était jamais transmis au navigateur - MSE, qui
# délègue ce découpage à ffmpeg, marchait sur ces mêmes caméras.
type_ = blink_ts_demux.type_nal(nal)
if type_ == 9 or (
self._unite_a_vcl and blink_ts_demux.debut_unite_acces(nal)
):
self._flush_unite_courante()
if not self._unite_courante:
self._pts_courant = pts
if type_ in (1, 5):
self._unite_a_vcl = True
self._unite_courante_octets += len(nal)
if self._unite_courante_octets > FILE_IMAGES_MAX_OCTETS:
raise RuntimeError(msg("nal_sans_fin"))
self._unite_courante.append(nal)
if type_ in (7, 8) and not self.sps_pps_pret.is_set():
self.sps_pps += nal
if type_ == 8: # PPS suit toujours SPS : les
self.sps_pps_pret.set() # deux sont la
def _flush_unite_courante(self) -> None:
if not self._unite_courante:
return
unite = b"".join(self._unite_courante)
self._mettre_unite(self._pts_courant, unite)
self._synchroniser_enregistrement(unite)
self._unite_courante = []
self._unite_courante_octets = 0
self._unite_a_vcl = False
def _mettre_unite(self, pts, unite: bytes) -> None:
if (self._file.full()
or self._file_octets + len(unite) > FILE_IMAGES_MAX_OCTETS):
raise RuntimeError(msg("navigateur_ne_consomme_plus"))
self._file.put_nowait((pts, unite))
self._file_octets += len(unite)
def _terminer_file(self, *, vider: bool = True) -> None:
# Libère les octets et réveille recv(), même si la lecture a été
# annulée avant son tout premier tour de boucle.
if vider:
while not self._file.empty():
self._file.get_nowait()
self._file_octets = 0
try:
self._file.put_nowait(None)
except asyncio.QueueFull:
# Ne peut arriver que si FILE_IMAGES_MAX (900) images
# attendent déjà sans qu'aucune n'ait jamais été consommée -
# _mettre_unite() refuse déjà d'en ajouter une 901e avant
# d'en arriver là. best-effort : recv() finira par relire le
# flux vide sans sentinelle plutôt que de planter le
# nettoyage (fermeture du pc, libération du module Blink).
pass
async def recv(self):
item = await self._file.get()
if item is None:
self.stop()
if self._demander_fermeture is not None:
self._demander_fermeture()
raise MediaStreamError
pts, donnees = item
self._file_octets -= len(donnees)
if pts is None:
pts = int((time.monotonic() - self._t0) * 90000)
# Ancre posee sur la toute premiere image, puis chaque image
# attend son tour selon l'ecart de PTS depuis cette ancre - pas
# selon l'ordre d'arrivee. Un reseau qui rattrape un trou en
# livrant plusieurs images d'un coup ne les fait donc pas
# ressortir toutes en meme temps : elles restent espacees comme
# a la prise de vue.
if self._t_ancrage is None:
self._t_ancrage = time.monotonic()
self._pts_ancrage = pts
cible = (
self._t_ancrage + TAMPON_LECTURE_SECONDS
+ (pts - self._pts_ancrage) / 90000
)
attente = cible - time.monotonic()
if attente > 0:
await asyncio.sleep(attente)
paquet = av.Packet(donnees)
paquet.pts = pts
paquet.time_base = Fraction(1, 90000)
return paquet
def _synchroniser_enregistrement(self, unite: bytes) -> None:
"""Démarre/arrête le sous-processus d'enregistrement selon l'état
voulu par le bouton de la page (enregistrement_actif), puis lui
transmet cette unité d'accès si un enregistrement est en cours.
Consulté à chaque unité d'accès complète (plusieurs fois par
seconde) : un clic sur le bouton met donc au plus une fraction
de seconde à se voir ici, sans file d'attente ni événement
dédié - même mécanique que le drapeau ENREGISTREMENT_DIRECT_
ACTIF côté MSE (serve.py), lu à chaque bloc envoyé au navigateur."""
if self._enregistrement_actif is None or self._chemin_enregistrement is None:
return
desire = self._enregistrement_actif()
if (desire and not self._recorder_demande
and self.sps_pps_pret.is_set()
and (self._recorder_demarrage is None
or self._recorder_demarrage.done())):
# Un MP4 commencé par une image dépendante n'est pas
# décodable. Attendre le prochain IDR, sans réencoder.
depart = 0
while True:
depart = unite.find(b"\x00\x00\x01", depart)
if depart < 0:
return
depart += 3
if depart < len(unite) and unite[depart] & 0x1F == 5:
break
self._recorder_demande = True
self._recorder_attente = [unite]
self._recorder_attente_octets = len(unite)
self._recorder_demarrage = asyncio.create_task(
self._demarrer_enregistrement()
)
return
elif not desire and self._recorder_demande:
self._arreter_enregistrement()
return # deja arrete : rien a transmettre pour cette unite
if self._recorder_demande and self._recorder_stdin is None:
# create_subprocess_exec rend la main à la boucle : toutes
# les unités reçues dans cet intervalle doivent suivre l'IDR.
if (self._recorder_attente_octets + len(unite)
> TAMPON_ENREGISTREMENT_MAX_OCTETS):
self._journal(msg("enregistrement_trop_lent"))
self._arreter_enregistrement()
return
self._recorder_attente.append(unite)
self._recorder_attente_octets += len(unite)
return
self._ecrire_enregistrement(unite)
async def _demarrer_enregistrement(self) -> None:
"""Lance un ffmpeg dedie qui remuxe (sans reencoder) le flux
Annexe B deja demultiplexe ici vers un MP4 fragmente - meme
format que /live-mse (frag_keyframe+empty_moov, serve.py), pour
la meme raison : lisible meme coupe net (fermeture d'onglet,
crash), sans dependre d'une sortie propre pour ecrire un moov
final valide.
Sous-processus separe de la RTCPeerConnection a dessein : si
l'enregistrement ralentit (disque lent...), il ne doit jamais
ralentir _lire() ni donc le direct affiche au navigateur.
Les unités arrivées pendant le lancement sont conservées dans
_recorder_attente, depuis un IDR et jusqu'à l'adoption du stdin.
"""
try:
chemin = self._chemin_enregistrement()
runtime.creer_sous_sorties(chemin.parent)
processus = await asyncio.create_subprocess_exec(
self._ffmpeg, "-hide_banner", "-loglevel", "error",
"-f", "h264", "-i", "pipe:0",
"-c", "copy", "-f", "mp4",
"-movflags", "frag_keyframe+empty_moov+default_base_moof",
str(chemin),
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
creationflags=runtime.SANS_FENETRE,
)
except Exception as error:
# Permet un nouvel essai au prochain passage a "desire" :
# sans ceci, un seul echec (ffmpeg introuvable...) bloquerait
# tout enregistrement pour le reste de la session.
self._journal(msg("echec_demarrage_enregistrement",
erreur=f"{type(error).__name__}: {error}"))
self._recorder_demande = False
self._recorder_attente.clear()
self._recorder_attente_octets = 0
return
if not self._recorder_demande or self.readyState == "ended":
# Arrete (bouton, ou piste fermee) pendant que ce sous-
# processus demarrait : personne n'en veut plus, ne pas
# l'adopter - juste le laisser sortir proprement.
try:
processus.stdin.close()
except Exception:
pass
asyncio.ensure_future(_attendre_fin_enregistrement(processus))
return
self._recorder_process = processus
self._recorder_stdin = processus.stdin
self._ecrire_enregistrement(self.sps_pps)
attente = self._recorder_attente
self._recorder_attente = []
self._recorder_attente_octets = 0
for unite in attente:
self._ecrire_enregistrement(unite)
def _ecrire_enregistrement(self, donnees: bytes) -> None:
# write() ne bloque pas. Borner son tampon protège la mémoire
# si ffmpeg/disque cesse de consommer, sans ralentir le direct.
stdin = self._recorder_stdin
if stdin is None:
return
try:
if (stdin.transport.get_write_buffer_size() + len(donnees)
> TAMPON_ENREGISTREMENT_MAX_OCTETS):
raise BufferError("tampon disque plein")
stdin.write(donnees)
except Exception as error:
# Une seule ligne meme si _lire() rappelle _ecrire_enregistrement
# plusieurs fois par seconde : stdin passe a None des la
# premiere erreur, donc ce bloc ne s'execute plus ensuite.
self._journal(
f"enregistrement disque interrompu (ecriture), "
f"{type(error).__name__}: {error}"
)
self._arreter_enregistrement()
def _arreter_enregistrement(self) -> None:
self._recorder_demande = False
self._recorder_attente.clear()
self._recorder_attente_octets = 0
stdin = self._recorder_stdin
self._recorder_stdin = None
if stdin is not None:
try:
stdin.close()
except Exception:
pass
processus = self._recorder_process
self._recorder_process = None
if processus is not None:
asyncio.ensure_future(_attendre_fin_enregistrement(processus))
def fermer(self) -> None:
# Python 3.8 : wait_for peut absorber cancel() si read() vient
# de finir. Le drapeau interdit alors une nouvelle lecture.
self._fermee = True
self.stop()
self._tache.cancel()
self._terminer_file()
try:
self._arreter_enregistrement()
except Exception as error:
# Appelée depuis le nettoyage (negocier(), plus bas), avant
# `await on_close()` : une exception non
# rattrapée ici y interromprait la coroutine avant on_close,
# qui libère MODULE_SLOT - direct.log en a montré un cas
# réel (2026-09-04, "négocié" jamais suivi de "session
# rendue normalement", verrou resté pris indéfiniment).
# sous pythonw (autostart), sans console, une telle
# exception ne serait de toute façon jamais vue nulle part
# sans cette ligne : c'est exactement le cas que ce
# journal doit désormais rendre visible.
self._journal(msg("echec_arret_enregistrement",
erreur=f"{type(error).__name__}: {error}"))
async def _attendre_fin_enregistrement(processus) -> None:
"""Laisse ffmpeg vider son tampon et sortir de lui-meme une fois
stdin ferme (fermer(), ci-dessus) ; au-dela, meme plafond (5s) que
le nettoyage ffmpeg de send_live_mse (serve.py)."""
try:
await asyncio.wait_for(processus.wait(), timeout=5)
except Exception:
try:
processus.kill()
await asyncio.wait_for(processus.wait(), timeout=5)
except Exception:
pass
def _profile_level_id(sps_pps_annexb: bytes) -> Optional[str]:
"""profile_idc/level_idc (ex. "640028") a partir d'un SPS Annexe B.
Ces 3 octets suivent toujours l'entete NAL en tete de SPS (ITU-T
H.264, 7.3.2.1.1)."""
i = sps_pps_annexb.find(b"\x00\x00\x01")
if i == -1:
return None
i += 3
if len(sps_pps_annexb) < i + 4 or (sps_pps_annexb[i] & 0x1F) != 7:
return None
return sps_pps_annexb[i + 1 : i + 4].hex()
_PROFILS_ENREGISTRES: set = set()
def _profil_reception_h264(offer_sdp: str, profil_source: str) -> str:
"""Préfère le profil source ; accepte le décodeur High 4:4:4
logiciel de Chromium pour un flux High, sans convertir les images.
Il s'agit des capacités du récepteur, jamais d'annoncer Baseline
pour une source High. Le SPS et les octets vidéo restent inchangés.
ITU-T H.264.1 §6.5.7 impose ce décodage du sous-ensemble High.
"""
source, _ = parse_h264_profile_level_id(profil_source)
profils = set()
for media in SessionDescription.parse(offer_sdp).media:
if media.kind != "video" or media.port == 0 or media.direction in ("sendonly", "inactive"):
continue
for codec in media.rtp.codecs:
if (codec.mimeType.lower() != "video/h264"
or str(codec.parameters.get("packetization-mode", "0")) != "1"):
continue
try:
profil, _ = parse_h264_profile_level_id(
codec.parameters.get("profile-level-id", "42e01f")
)
profils.add(profil)
except ValueError:
continue
if source in profils:
return profil_source
if (source in (H264Profile.PROFILE_HIGH, H264Profile.PROFILE_CONSTRAINED_HIGH)
and H264Profile.PROFILE_PREDICTIVE_HIGH_444 in profils):
return "f400" + profil_source[4:]
return profil_source # aiortc produira l'erreur de codecs incompatibles.
def _enregistrer_profil_h264(profil: str) -> None:
"""Ajoute ce profile-level-id a aiortc.codecs.CODECS["video"].
get_capabilities() (aiortc/codecs/__init__.py) lit directement
cette liste, remplie une fois par init_codecs() avec seulement du
Baseline. setCodecPreferences() valide son argument contre cette
meme liste ("Codec is not in capabilities" sinon)."""
if profil in _PROFILS_ENREGISTRES:
return
_PROFILS_ENREGISTRES.add(profil)
pt = 110 + 2 * len(_PROFILS_ENREGISTRES)
CODECS["video"] += [
RTCRtpCodecParameters(
mimeType="video/H264", clockRate=90000, payloadType=pt,
rtcpFeedback=[
RTCRtcpFeedback(type="nack"),
RTCRtcpFeedback(type="nack", parameter="pli"),
RTCRtcpFeedback(type="goog-remb"),
],
parameters={
"level-asymmetry-allowed": "1",
"packetization-mode": "1",
"profile-level-id": profil,
},
),
RTCRtpCodecParameters(
mimeType="video/rtx", clockRate=90000, payloadType=pt + 1,
parameters={"apt": pt},
),
]
LIBELLES = {
"fr": {
"flux_interrompu": "lecture du flux interrompue, {erreur}",
"nal_sans_fin": "Unité d'accès H.264 sans fin détectable",
"navigateur_ne_consomme_plus": "Le navigateur ne consomme plus le flux vidéo",
"enregistrement_trop_lent": "enregistrement arrêté : lancement trop lent",
"echec_arret_enregistrement": "échec d'arrêt de l'enregistrement pendant la fermeture, {erreur}",
"aiortc_absent": "aiortc n'est pas installé (pip install -r requirements.txt)",
"flux_arrete_avant_image": "Le flux Blink s'est arrêté avant la première image.",
"aucune_image": "La caméra n'a envoyé aucune image. Hors de portée du module, endormie, ou déjà occupée par une autre session.",
"h264_negocie": "H.264 source {source}, décodeur négocié {reception}, sans réencodage",
"flux_arrete_negociation": "Le flux Blink s'est arrêté pendant la négociation.",
"echec_demarrage_enregistrement": "échec de démarrage de l'enregistrement disque, {erreur}",
"filet_declenche": "filet de {secondes}s déclenché, connectionState restait à {etat}",
},
"en": {
"flux_interrompu": "stream reading interrupted, {erreur}",
"nal_sans_fin": "H.264 access unit with no detectable end",
"navigateur_ne_consomme_plus": "The browser no longer consumes the video stream",
"enregistrement_trop_lent": "recording stopped: too slow to start",
"echec_arret_enregistrement": "could not stop the recording while closing, {erreur}",
"aiortc_absent": "aiortc is not installed (pip install -r requirements.txt)",
"flux_arrete_avant_image": "The Blink stream stopped before the first picture.",
"aucune_image": "The camera sent no picture. Out of the module's range, asleep, or already busy with another session.",
"h264_negocie": "H.264 source {source}, negotiated decoder {reception}, no re-encoding",
"flux_arrete_negociation": "The Blink stream stopped during negotiation.",
"echec_demarrage_enregistrement": "could not start the disk recording, {erreur}",
"filet_declenche": "{secondes}s safety net triggered, connectionState was still {etat}",
},
}
def msg(cle: str, **valeurs) -> str:
return runtime.traduire(LIBELLES, cle, **valeurs)
async def fermer_connexion(pc) -> None:
"""Ferme WebRTC et attend la libération du flux Blink associé.
aiortc émet ses événements de fermeture sans attendre leurs coroutines.
pc.close() seul ne garantit donc pas que on_close() a terminé.
"""
demander = getattr(pc, "_blink_demander_fermeture", None)
if demander is None:
await pc.close()
else:
await asyncio.shield(demander())
async def negocier(
stream_url: str, offer_sdp: str, offer_type: str,
on_close: Callable[[], Awaitable[None]],
ffmpeg: Optional[str] = None,
chemin_enregistrement: Optional[Callable[[], Path]] = None,
enregistrement_actif: Optional[Callable[[], bool]] = None,
journal: Optional[Callable[[str], None]] = None,
):
"""Ouvre le flux Blink (sans decodage), negocie WebRTC, renvoie
(pc, sdp_reponse, type_reponse).
`on_close` est attendu (await) quand la connexion se termine (etat
"closed" ou "failed") : c'est le seul signal de fin disponible ici, il
n'y a pas de requete HTTP a garder ouverte comme en MSE - fermer
l'onglet ferme la RTCPeerConnection cote navigateur, ce qui remonte
jusqu'ici via connectionstatechange. Appele au plus une fois.
`chemin_enregistrement`/`enregistrement_actif`, avec `ffmpeg` : si
fournis, une copie du direct est écrite à ce chemin tant que ce dernier
répond vrai - piloté par le bouton de la page (voir
_PisteH264._synchroniser_enregistrement)."""
if not DISPONIBLE:
raise RuntimeError(msg("aiortc_absent"))
journal = journal or (lambda _message: None)
host, port_str = stream_url.replace("tcp://", "").split(":")
writer = None
track = None
pc = None
tache_plafond = None
tache_fermeture = None
async def _fermer_ressources() -> None:
# Cette tâche unique est protégée de l'annulation de negocier() et
# des demandes HTTP. Elle possède le nettoyage jusqu'à sa fin.
if tache_plafond is not None:
tache_plafond.cancel()
try:
if track is not None:
try:
track.fermer()
except Exception as error:
journal(f"fermeture piste : {type(error).__name__}: {error}")
with suppress(asyncio.CancelledError, Exception):
await track._tache
if writer is not None:
try:
writer.close()
await asyncio.wait_for(writer.wait_closed(), timeout=5)
except Exception as error:
journal(f"fermeture TCP : {type(error).__name__}: {error}")
if pc is not None:
try:
await asyncio.wait_for(pc.close(), timeout=FERMETURE_MAX_SECONDS)
except Exception as error:
journal(f"fermeture WebRTC : {type(error).__name__}: {error}")
finally:
await on_close()
def _demander_fermeture():
nonlocal tache_fermeture
if tache_fermeture is None:
tache_fermeture = asyncio.create_task(_fermer_ressources())
# Les fins spontanées (EOF/état failed) n'ont pas d'appelant
# attendant cette tâche : consommer et journaliser leur erreur.
def _fin(tache):
if not tache.cancelled() and tache.exception() is not None:
error = tache.exception()
journal(f"nettoyage WebRTC : {type(error).__name__}: {error}")
tache_fermeture.add_done_callback(_fin)
return tache_fermeture
try:
reader, writer = await asyncio.open_connection(host, int(port_str))
track = _PisteH264(
reader, ffmpeg=ffmpeg, chemin_enregistrement=chemin_enregistrement,
enregistrement_actif=enregistrement_actif, journal=journal,
)
# Le temps que met la camera a se reveiller et a livrer son
# SPS/PPS - PREMIERE_IMAGE_MAX_SECONDS (40s, cf. plus haut), pas
# NEGOCIATION_MAX_SECONDS (15s, pour la suite, purement locale).
try:
attente_sps = asyncio.create_task(track.sps_pps_pret.wait())
try:
terminees, _ = await asyncio.wait(
(attente_sps, track._tache),
timeout=PREMIERE_IMAGE_MAX_SECONDS,
return_when=asyncio.FIRST_COMPLETED,
)
if not terminees:
raise asyncio.TimeoutError
if track._tache.done():
raise RuntimeError(msg("flux_arrete_avant_image"))
finally:
attente_sps.cancel()
with suppress(asyncio.CancelledError):
await attente_sps
except asyncio.TimeoutError:
# Sans ceci, l'utilisateur ne voyait qu'un "TimeoutError: " brut
# (constate en reel, 2026-09-03) - RuntimeError s'affiche tel
# quel cote page (send_offer_webrtc, serve.py), meme traitement
# que les messages MSE deja clairs pour ce cas.
raise RuntimeError(msg("aucune_image")) from None
# iceServers=[] : sans ceci, aiortc retombe sur son defaut
# (stun:stun.l.google.com:19302, RTCIceGatherer.getDefaultIceServers)
# et setLocalDescription() n'y renvoie qu'une fois la collecte de
# candidats terminee - un blocage reseau/pare-feu dessus bloque tout
# indefiniment (constate en reel, cf. BACKLOG.md). Navigateur et
# serveur sur le meme reseau local : aucun NAT a traverser, un STUN
# n'a de toute facon rien a apporter ici.
pc = RTCPeerConnection(configuration=RTCConfiguration(iceServers=[]))
track.pc = pc
track._demander_fermeture = _demander_fermeture
pc._blink_demander_fermeture = _demander_fermeture
@pc.on("connectionstatechange")
def _sur_changement_etat() -> None:
journal(f"connectionstatechange -> {pc.connectionState}")
if pc.connectionState in ("closed", "failed"):
# Ne pas attendre pc.close() depuis son propre événement :
# aiortc peut encore être en train de fermer les transports.
_demander_fermeture()
sender = pc.addTrack(track)
profil = _profile_level_id(track.sps_pps)
journal(f"H.264 source {profil or 'SPS illisible'}")
if profil:
profil_reception = _profil_reception_h264(offer_sdp, profil)
if profil_reception != profil:
journal(msg("h264_negocie", source=profil, reception=profil_reception))
profil = profil_reception
_enregistrer_profil_h264(profil)
transceiver = next(
t for t in pc.getTransceivers() if t.sender is sender
)
transceiver.setCodecPreferences([
RTCRtpCodecCapability(
mimeType="video/H264", clockRate=90000,
parameters={
"level-asymmetry-allowed": "1",
"packetization-mode": "1",
"profile-level-id": profil,
},
)
])
# Filet de securite : meme sans STUN, rien ne garantit qu'une
# negociation aboutisse toujours vite (reseau capricieux, bug
# aiortc/aioice non identifie...). Sans plafond ici, un blocage
# laisse MODULE_SLOT/le verrou disque tenus indefiniment - deja
# observe en reel (cf. BACKLOG.md) : aucun connectionstatechange
# n'arrive jamais si la negociation elle-meme ne se termine jamais.
async def _echanger_sdp():
await pc.setRemoteDescription(
RTCSessionDescription(sdp=offer_sdp, type=offer_type)
)
answer = await pc.createAnswer()
await pc.setLocalDescription(answer)
await asyncio.wait_for(
_echanger_sdp(), timeout=NEGOCIATION_MAX_SECONDS
)
if tache_fermeture is not None:
raise RuntimeError(msg("flux_arrete_negociation"))
except (Exception, asyncio.CancelledError):
# CancelledError n'hérite pas d'Exception. L'expiration du budget
# extérieur doit aussi fermer le TCP et rendre le module.
with suppress(Exception):
await asyncio.shield(_demander_fermeture())
raise
# Deuxieme filet, une fois la negociation reussie : une session qui ne
# se ferme jamais proprement (onglet tue plutot que ferme, reseau qui
# tombe sans notification...) ne doit pas non plus tenir la ressource
# indefiniment - meme principe que LIVE_MAX_SECONDS cote MSE (serve.py).
async def _plafond_duree() -> None:
await asyncio.sleep(SESSION_MAX_SECONDS)
if pc.connectionState not in ("closed", "failed"):
journal(msg("filet_declenche", secondes=SESSION_MAX_SECONDS,
etat=repr(pc.connectionState)))
await fermer_connexion(pc)
tache_plafond = asyncio.create_task(_plafond_duree())
return pc, pc.localDescription.sdp, pc.localDescription.type