Additional stream typing improvements (#129695)

homeassistant/components/stream/core.py

@@ -9,7 +9,7 @@ from dataclasses import dataclass, field
 import datetime
 from enum import IntEnum
 import logging
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from aiohttp import web
 import numpy as np
@@ -27,7 +27,8 @@ from .const import (
 )
 
 if TYPE_CHECKING:
-    from av import CodecContext, Packet
+    from av import Packet
+    from av.video.codeccontext import VideoCodecContext
 
     from homeassistant.components.camera import DynamicStreamSettings
 
@@ -448,7 +449,7 @@ class KeyFrameConverter:
         self._image: bytes | None = None
         self._turbojpeg = TurboJPEGSingleton.instance()
         self._lock = asyncio.Lock()
-        self._codec_context: CodecContext | None = None
+        self._codec_context: VideoCodecContext | None = None
         self._stream_settings = stream_settings
         self._dynamic_stream_settings = dynamic_stream_settings
 
@@ -460,7 +461,7 @@ class KeyFrameConverter:
         self._packet = packet
         self._hass.loop.call_soon_threadsafe(self._event.set)
 
-    def create_codec_context(self, codec_context: CodecContext) -> None:
+    def create_codec_context(self, codec_context: VideoCodecContext) -> None:
         """Create a codec context to be used for decoding the keyframes.
 
         This is run by the worker thread and will only be called once per worker.
@@ -474,7 +475,9 @@ class KeyFrameConverter:
         # pylint: disable-next=import-outside-toplevel
         from av import CodecContext
 
-        self._codec_context = CodecContext.create(codec_context.name, "r")
+        self._codec_context = cast(
+            "VideoCodecContext", CodecContext.create(codec_context.name, "r")
+        )
         self._codec_context.extradata = codec_context.extradata
         self._codec_context.skip_frame = "NONKEY"
         self._codec_context.thread_type = "NONE"

homeassistant/components/stream/recorder.py

@@ -122,7 +122,7 @@ class RecorderOutput(StreamOutput):
             if not output_v:
                 output_v = output.add_stream(template=source_v)
                 context = output_v.codec_context
-                context.flags |= "GLOBAL_HEADER"
+                context.global_header = True
             if source_a and not output_a:
                 output_a = output.add_stream(template=source_a)
 

homeassistant/components/stream/worker.py

@@ -127,6 +127,16 @@ class StreamState:
 class StreamMuxer:
     """StreamMuxer re-packages video/audio packets for output."""
 
+    _segment_start_dts: int
+    _memory_file: BytesIO
+    _av_output: av.container.OutputContainer
+    _output_video_stream: av.video.VideoStream
+    _output_audio_stream: av.audio.AudioStream | None
+    _segment: Segment | None
+    # the following 2 member variables are used for Part formation
+    _memory_file_pos: int
+    _part_start_dts: int
+
     def __init__(
         self,
         hass: HomeAssistant,
@@ -138,19 +148,10 @@ class StreamMuxer:
     ) -> None:
         """Initialize StreamMuxer."""
         self._hass = hass
-        self._segment_start_dts: int = cast(int, None)
+        self._input_video_stream = video_stream
-        self._memory_file: BytesIO = cast(BytesIO, None)
+        self._input_audio_stream = audio_stream
-        self._av_output: av.container.OutputContainer = None
-        self._input_video_stream: av.video.VideoStream = video_stream
-        self._input_audio_stream: av.audio.AudioStream | None = audio_stream
         self._audio_bsf = audio_bsf
-        self._audio_bsf_context: av.BitStreamFilterContext = None
+        self._audio_bsf_context: av.BitStreamFilterContext | None = None
-        self._output_video_stream: av.video.VideoStream = None
-        self._output_audio_stream: av.audio.AudioStream | None = None
-        self._segment: Segment | None = None
-        # the following 3 member variables are used for Part formation
-        self._memory_file_pos: int = cast(int, None)
-        self._part_start_dts: int = cast(int, None)
         self._part_has_keyframe = False
         self._stream_settings = stream_settings
         self._stream_state = stream_state
@@ -256,7 +257,7 @@ class StreamMuxer:
             input_astream=self._input_audio_stream,
         )
         if self._output_video_stream.name == "hevc":
-            self._output_video_stream.codec_tag = "hvc1"
+            self._output_video_stream.codec_context.codec_tag = "hvc1"
 
     def mux_packet(self, packet: av.Packet) -> None:
         """Mux a packet to the appropriate output stream."""
@@ -562,7 +563,7 @@ def stream_worker(
 
     dts_validator = TimestampValidator(
         int(1 / video_stream.time_base),
-        1 / audio_stream.time_base if audio_stream else 1,
+        int(1 / audio_stream.time_base) if audio_stream else 1,
     )
     container_packets = PeekIterator(
         filter(dts_validator.is_valid, container.demux((video_stream, audio_stream)))