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

Minor stream typing improvements (#129691)

Browse Source
This commit is contained in:
Marc Mueller 2024-11-02 19:29:09 +01:00 committed by GitHub
parent 4f20977a8e
commit 0eea3176d6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,5 +1,9 @@
"""Constants for Stream component.""" """Constants for Stream component."""
from __future__ import annotations
from typing import Final
DOMAIN = "stream" DOMAIN = "stream"
ATTR_ENDPOINTS = "endpoints" ATTR_ENDPOINTS = "endpoints"
@ -11,8 +15,8 @@ RECORDER_PROVIDER = "recorder"
OUTPUT_FORMATS = [HLS_PROVIDER] OUTPUT_FORMATS = [HLS_PROVIDER]
SEGMENT_CONTAINER_FORMAT = "mp4" # format for segments SEGMENT_CONTAINER_FORMAT: Final = "mp4" # format for segments
RECORDER_CONTAINER_FORMAT = "mp4" # format for recorder output RECORDER_CONTAINER_FORMAT: Final = "mp4" # format for recorder output
AUDIO_CODECS = {"aac", "mp3"} AUDIO_CODECS = {"aac", "mp3"}
FORMAT_CONTENT_TYPE = {HLS_PROVIDER: "application/vnd.apple.mpegurl"} FORMAT_CONTENT_TYPE = {HLS_PROVIDER: "application/vnd.apple.mpegurl"}

2
homeassistant/components/stream/core.py
View File

@ -442,7 +442,7 @@ class KeyFrameConverter:
# pylint: disable-next=import-outside-toplevel # pylint: disable-next=import-outside-toplevel
from homeassistant.components.camera.img_util import TurboJPEGSingleton from homeassistant.components.camera.img_util import TurboJPEGSingleton
self._packet: Packet = None self._packet: Packet | None = None
self._event: asyncio.Event = asyncio.Event() self._event: asyncio.Event = asyncio.Event()
self._hass = hass self._hass = hass
self._image: bytes | None = None self._image: bytes | None = None

5
homeassistant/components/stream/recorder.py
View File

@ -9,6 +9,7 @@ import os
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import av import av
import av.container
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
@ -168,7 +169,9 @@ class RecorderOutput(StreamOutput):
os.remove(video_path + ".tmp") os.remove(video_path + ".tmp")
def finish_writing( def finish_writing(
segments: deque[Segment], output: av.OutputContainer, video_path: str segments: deque[Segment],
output: av.container.OutputContainer | None,
video_path: str,
) -> None: ) -> None:
"""Finish writing output.""" """Finish writing output."""
# Should only have 0 or 1 segments, but loop through just in case # Should only have 0 or 1 segments, but loop through just in case

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

@ -13,6 +13,10 @@ from threading import Event
from typing import Any, Self, cast from typing import Any, Self, cast
import av import av
import av.audio
import av.container
import av.stream
import av.video
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
@ -127,7 +131,7 @@ class StreamMuxer:
self, self,
hass: HomeAssistant, hass: HomeAssistant,
video_stream: av.video.VideoStream, video_stream: av.video.VideoStream,
audio_stream: av.audio.stream.AudioStream | None, audio_stream: av.audio.AudioStream | None,
audio_bsf: av.BitStreamFilter | None, audio_bsf: av.BitStreamFilter | None,
stream_state: StreamState, stream_state: StreamState,
stream_settings: StreamSettings, stream_settings: StreamSettings,
@ -138,11 +142,11 @@ class StreamMuxer:
self._memory_file: BytesIO = cast(BytesIO, None) self._memory_file: BytesIO = cast(BytesIO, None)
self._av_output: av.container.OutputContainer = None self._av_output: av.container.OutputContainer = None
self._input_video_stream: av.video.VideoStream = video_stream self._input_video_stream: av.video.VideoStream = video_stream
self._input_audio_stream: av.audio.stream.AudioStream | None = audio_stream self._input_audio_stream: av.audio.AudioStream | None = audio_stream
self._audio_bsf = audio_bsf self._audio_bsf = audio_bsf
self._audio_bsf_context: av.BitStreamFilterContext = None self._audio_bsf_context: av.BitStreamFilterContext = None
self._output_video_stream: av.video.VideoStream = None self._output_video_stream: av.video.VideoStream = None
self._output_audio_stream: av.audio.stream.AudioStream | None = None self._output_audio_stream: av.audio.AudioStream | None = None
self._segment: Segment | None = None self._segment: Segment | None = None
# the following 3 member variables are used for Part formation # the following 3 member variables are used for Part formation
self._memory_file_pos: int = cast(int, None) self._memory_file_pos: int = cast(int, None)
@ -157,11 +161,11 @@ class StreamMuxer:
memory_file: BytesIO, memory_file: BytesIO,
sequence: int, sequence: int,
input_vstream: av.video.VideoStream, input_vstream: av.video.VideoStream,
input_astream: av.audio.stream.AudioStream | None, input_astream: av.audio.AudioStream | None,
) -> tuple[ ) -> tuple[
av.container.OutputContainer, av.container.OutputContainer,
av.video.VideoStream, av.video.VideoStream,
av.audio.stream.AudioStream | None, av.audio.AudioStream | None,
]: ]:
"""Make a new av OutputContainer and add output streams.""" """Make a new av OutputContainer and add output streams."""
container_options: dict[str, str] = { container_options: dict[str, str] = {
@ -396,7 +400,7 @@ class StreamMuxer:
self._memory_file.close() self._memory_file.close()
class PeekIterator(Iterator): class PeekIterator(Iterator[av.Packet]):
"""An Iterator that may allow multiple passes. """An Iterator that may allow multiple passes.
This may be consumed like a normal Iterator, however also supports a This may be consumed like a normal Iterator, however also supports a