jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-27 15:17:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix AVError and allow more missing DTS packets in stream (#42277)

Browse Source 
* Fix AVError and allow more missing DTS in stream init

* Avoid recreating container demux iterator

* Relax missing dts requirement

* Fix spelling error

* Fix error message count

* Add timeout to av.open read

* Increase STREAM_TIMEOUT to 30
This commit is contained in:
uvjustin 2020-10-25 10:55:12 +08:00 committed by GitHub
parent 09bd449698
commit f8d87bff16
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 39 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/stream/const.py
View File

@ -20,3 +20,6 @@ MIN_SEGMENT_DURATION = 1.5 # Each segment is at least this many seconds
PACKETS_TO_WAIT_FOR_AUDIO = 20 # Some streams have an audio stream with no audio PACKETS_TO_WAIT_FOR_AUDIO = 20 # Some streams have an audio stream with no audio
MAX_TIMESTAMP_GAP = 10000 # seconds - anything from 10 to 50000 is probably reasonable MAX_TIMESTAMP_GAP = 10000 # seconds - anything from 10 to 50000 is probably reasonable
MAX_MISSING_DTS = 6 # Number of packets missing DTS to allow
STREAM_TIMEOUT = 30 # Timeout for reading stream

59
homeassistant/components/stream/worker.py
View File

@ -6,7 +6,13 @@ import time
import av import av
from .const import MAX_TIMESTAMP_GAP, MIN_SEGMENT_DURATION, PACKETS_TO_WAIT_FOR_AUDIO from .const import (
MAX_MISSING_DTS,
MAX_TIMESTAMP_GAP,
MIN_SEGMENT_DURATION,
PACKETS_TO_WAIT_FOR_AUDIO,
STREAM_TIMEOUT,
)
from .core import Segment, StreamBuffer from .core import Segment, StreamBuffer
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -62,7 +68,7 @@ def stream_worker(hass, stream, quit_event):
def _stream_worker_internal(hass, stream, quit_event): def _stream_worker_internal(hass, stream, quit_event):
"""Handle consuming streams.""" """Handle consuming streams."""
container = av.open(stream.source, options=stream.options) container = av.open(stream.source, options=stream.options, timeout=STREAM_TIMEOUT)
try: try:
video_stream = container.streams.video[0] video_stream = container.streams.video[0]
except (KeyError, IndexError): except (KeyError, IndexError):
@ -81,13 +87,15 @@ def _stream_worker_internal(hass, stream, quit_event):
if audio_stream and audio_stream.profile is None: if audio_stream and audio_stream.profile is None:
audio_stream = None audio_stream = None
# Iterator for demuxing
container_packets = None
# The presentation timestamps of the first packet in each stream we receive # The presentation timestamps of the first packet in each stream we receive
# Use to adjust before muxing or outputting, but we don't adjust internally # Use to adjust before muxing or outputting, but we don't adjust internally
first_pts = {} first_pts = {}
# The decoder timestamps of the latest packet in each stream we processed # The decoder timestamps of the latest packet in each stream we processed
last_dts = None last_dts = None
# Keep track of consecutive packets without a dts to detect end of stream. # Keep track of consecutive packets without a dts to detect end of stream.
last_packet_was_without_dts = False missing_dts = 0
# Holds the buffers for each stream provider # Holds the buffers for each stream provider
outputs = None outputs = None
# Keep track of the number of segments we've processed # Keep track of the number of segments we've processed
@ -102,8 +110,8 @@ def _stream_worker_internal(hass, stream, quit_event):
# 2 - seeking can be problematic https://trac.ffmpeg.org/ticket/7815 # 2 - seeking can be problematic https://trac.ffmpeg.org/ticket/7815
def peek_first_pts(): def peek_first_pts():
nonlocal first_pts, audio_stream nonlocal first_pts, audio_stream, container_packets
missing_dts = False missing_dts = 0
def empty_stream_dict(): def empty_stream_dict():
return { return {
@ -112,17 +120,20 @@ def _stream_worker_internal(hass, stream, quit_event):
} }
try: try:
container_packets = container.demux((video_stream, audio_stream))
first_packet = empty_stream_dict() first_packet = empty_stream_dict()
first_pts = empty_stream_dict() first_pts = empty_stream_dict()
# Get to first video keyframe # Get to first video keyframe
while first_packet[video_stream] is None: while first_packet[video_stream] is None:
packet = next(container.demux()) packet = next(container_packets)
if ( if (
packet.dts is None packet.dts is None
): # Allow single packet with no dts, raise error on second ): # Allow MAX_MISSING_DTS packets with no dts, raise error on the next one
if missing_dts: if missing_dts >= MAX_MISSING_DTS:
raise av.AVError raise StopIteration(
missing_dts = True f"Invalid data - got {MAX_MISSING_DTS+1} packets with missing DTS while initializing"
)
missing_dts += 1
continue continue
if packet.stream == video_stream and packet.is_keyframe: if packet.stream == video_stream and packet.is_keyframe:
first_packet[video_stream] = packet first_packet[video_stream] = packet
@ -131,13 +142,15 @@ def _stream_worker_internal(hass, stream, quit_event):
while any( while any(
[pts is None for pts in {**first_packet, **first_pts}.values()] [pts is None for pts in {**first_packet, **first_pts}.values()]
) and (len(initial_packets) < PACKETS_TO_WAIT_FOR_AUDIO): ) and (len(initial_packets) < PACKETS_TO_WAIT_FOR_AUDIO):
packet = next(container.demux((video_stream, audio_stream))) packet = next(container_packets)
if ( if (
packet.dts is None packet.dts is None
): # Allow single packet with no dts, raise error on second ): # Allow MAX_MISSING_DTS packet with no dts, raise error on the next one
if missing_dts: if missing_dts >= MAX_MISSING_DTS:
raise av.AVError raise StopIteration(
missing_dts = True f"Invalid data - got {MAX_MISSING_DTS+1} packets with missing DTS while initializing"
)
missing_dts += 1
continue continue
if ( if (
first_packet[packet.stream] is None first_packet[packet.stream] is None
@ -223,16 +236,16 @@ def _stream_worker_internal(hass, stream, quit_event):
if len(initial_packets) > 0: if len(initial_packets) > 0:
packet = initial_packets.popleft() packet = initial_packets.popleft()
else: else:
packet = next(container.demux((video_stream, audio_stream))) packet = next(container_packets)
if packet.dts is None: if packet.dts is None:
_LOGGER.error("Stream packet without dts detected, skipping...") # Allow MAX_MISSING_DTS consecutive packets without dts. Terminate the stream on the next one.
# Allow a single packet without dts before terminating the stream. if missing_dts >= MAX_MISSING_DTS:
if last_packet_was_without_dts: raise StopIteration(
# If we get a "flushing" packet, the stream is done f"No dts in {MAX_MISSING_DTS+1} consecutive packets"
raise StopIteration("No dts in consecutive packets") )
last_packet_was_without_dts = True missing_dts += 1
continue continue
last_packet_was_without_dts = False missing_dts = 0
except (av.AVError, StopIteration) as ex: except (av.AVError, StopIteration) as ex:
_LOGGER.error("Error demuxing stream: %s", str(ex)) _LOGGER.error("Error demuxing stream: %s", str(ex))
finalize_stream() finalize_stream()