import asyncio import errno import fractions import logging import threading import time from typing import Any, Optional, Union, cast import av import av.container from av import AudioFrame, VideoFrame from av.audio import AudioStream from av.frame import Frame from av.packet import Packet from av.video.stream import VideoStream from ..mediastreams import AUDIO_PTIME, MediaStreamError, MediaStreamTrack logger = logging.getLogger(__name__) REAL_TIME_FORMATS = [ "alsa", "android_camera", "avfoundation", "bktr", "decklink", "dshow", "fbdev", "gdigrab", "iec61883", "jack", "kmsgrab", "openal", "oss", "pulse", "sndio", "rtsp", "v4l2", "vfwcap", "x11grab", ] _AudioOrVideoStream = Union[AudioStream, VideoStream] async def blackhole_consume(track: MediaStreamTrack) -> None: while True: try: await track.recv() except MediaStreamError: return class MediaBlackhole: """ A media sink that consumes and discards all media. """ def __init__(self) -> None: self.__tracks: dict[MediaStreamTrack, Optional[asyncio.Future]] = {} def addTrack(self, track: MediaStreamTrack) -> None: """ Add a track whose media should be discarded. :param track: A :class:`aiortc.MediaStreamTrack`. """ if track not in self.__tracks: self.__tracks[track] = None async def start(self) -> None: """ Start discarding media. """ for track, task in self.__tracks.items(): if task is None: self.__tracks[track] = asyncio.ensure_future(blackhole_consume(track)) async def stop(self) -> None: """ Stop discarding media. """ for task in self.__tracks.values(): if task is not None: task.cancel() self.__tracks = {} def player_worker_decode( loop: asyncio.AbstractEventLoop, container: av.container.InputContainer, streams: list[_AudioOrVideoStream], audio_track: "PlayerStreamTrack", video_track: "PlayerStreamTrack", quit_event: threading.Event, throttle_playback: bool, loop_playback: bool, ) -> None: audio_sample_rate = 48000 audio_samples = 0 audio_time_base = fractions.Fraction(1, audio_sample_rate) audio_resampler = av.AudioResampler( format="s16", layout="stereo", rate=audio_sample_rate, frame_size=int(audio_sample_rate * AUDIO_PTIME), ) video_first_pts = None frame_time = None start_time = time.time() while not quit_event.is_set(): try: frame = next(container.decode(*streams)) except Exception as exc: if isinstance(exc, av.FFmpegError) and exc.errno == errno.EAGAIN: time.sleep(0.01) continue if isinstance(exc, StopIteration) and loop_playback: container.seek(0) continue if audio_track: asyncio.run_coroutine_threadsafe(audio_track._queue.put(None), loop) if video_track: asyncio.run_coroutine_threadsafe(video_track._queue.put(None), loop) break # read up to 1 second ahead if throttle_playback: elapsed_time = time.time() - start_time if frame_time and frame_time > elapsed_time + 1: time.sleep(0.1) if isinstance(frame, AudioFrame) and audio_track: for frame in audio_resampler.resample(frame): # fix timestamps frame.pts = audio_samples frame.time_base = audio_time_base audio_samples += frame.samples frame_time = frame.time asyncio.run_coroutine_threadsafe(audio_track._queue.put(frame), loop) elif isinstance(frame, VideoFrame) and video_track: if frame.pts is None: # pragma: no cover logger.warning( "MediaPlayer(%s) Skipping video frame with no pts", container.name ) continue # video from a webcam doesn't start at pts 0, cancel out offset if video_first_pts is None: video_first_pts = frame.pts frame.pts -= video_first_pts frame_time = frame.time asyncio.run_coroutine_threadsafe(video_track._queue.put(frame), loop) def player_worker_demux( loop: asyncio.AbstractEventLoop, container: av.container.InputContainer, streams: list[_AudioOrVideoStream], audio_track: "PlayerStreamTrack", video_track: "PlayerStreamTrack", quit_event: threading.Event, throttle_playback: bool, loop_playback: bool, ) -> None: video_first_pts = None frame_time = None start_time = time.time() while not quit_event.is_set(): try: packet = next(container.demux(*streams)) if not packet.size: raise StopIteration except Exception as exc: if isinstance(exc, av.FFmpegError) and exc.errno == errno.EAGAIN: time.sleep(0.01) continue if isinstance(exc, StopIteration) and loop_playback: container.seek(0) continue if audio_track: asyncio.run_coroutine_threadsafe(audio_track._queue.put(None), loop) if video_track: asyncio.run_coroutine_threadsafe(video_track._queue.put(None), loop) break # read up to 1 second ahead if throttle_playback: elapsed_time = time.time() - start_time if frame_time and frame_time > elapsed_time + 1: time.sleep(0.1) track = None if isinstance(packet.stream, AudioStream) and audio_track: track = audio_track elif isinstance(packet.stream, VideoStream) and video_track: if packet.pts is None: # pragma: no cover logger.warning( "MediaPlayer(%s) Skipping video packet with no pts", container.name ) continue track = video_track # video from a webcam doesn't start at pts 0, cancel out offset if video_first_pts is None: video_first_pts = packet.pts packet.pts -= video_first_pts if ( track is not None and packet.pts is not None and packet.time_base is not None ): frame_time = int(packet.pts * packet.time_base) asyncio.run_coroutine_threadsafe(track._queue.put(packet), loop) class PlayerStreamTrack(MediaStreamTrack): def __init__(self, player: "MediaPlayer", kind: str) -> None: super().__init__() self.kind = kind self._player: Optional[MediaPlayer] = player self._queue: asyncio.Queue[Union[Frame, Packet]] = asyncio.Queue() self._start: Optional[float] = None async def recv(self) -> Union[Frame, Packet]: if self.readyState != "live": raise MediaStreamError self._player._start(self) data = await self._queue.get() if data is None: self.stop() raise MediaStreamError if isinstance(data, Frame): data_time = data.time elif isinstance(data, Packet): data_time = float(data.pts * data.time_base) # control playback rate if ( self._player is not None and self._player._throttle_playback and data_time is not None ): if self._start is None: self._start = time.time() - data_time else: wait = self._start + data_time - time.time() await asyncio.sleep(wait) return data def stop(self) -> None: super().stop() if self._player is not None: self._player._stop(self) self._player = None class MediaPlayer: """ A media source that reads audio and/or video from a file. Examples: .. code-block:: python # Open a video file. player = MediaPlayer('/path/to/some.mp4') # Open an HTTP stream. player = MediaPlayer( 'http://download.tsi.telecom-paristech.fr/' 'gpac/dataset/dash/uhd/mux_sources/hevcds_720p30_2M.mp4') # Open webcam on Linux. player = MediaPlayer('/dev/video0', format='v4l2', options={ 'video_size': '640x480' }) # Open webcam on OS X. player = MediaPlayer('default:none', format='avfoundation', options={ 'video_size': '640x480' }) # Open webcam on Windows. player = MediaPlayer('video=Integrated Camera', format='dshow', options={ 'video_size': '640x480' }) :param file: The path to a file, or a file-like object. :param format: The format to use, defaults to autodect. :param options: Additional options to pass to FFmpeg. :param timeout: Open/read timeout to pass to FFmpeg. :param loop: Whether to repeat playback indefinitely (requires a seekable file). """ def __init__( self, file: Any, format: Optional[str] = None, options: Optional[dict[str, str]] = None, timeout: Optional[int] = None, loop: bool = False, decode: bool = True, ) -> None: self.__container = av.open( file=file, format=format, mode="r", options=options, timeout=timeout ) self.__thread: Optional[threading.Thread] = None self.__thread_quit: Optional[threading.Event] = None # examine streams self.__started: set[PlayerStreamTrack] = set() self.__streams: list[_AudioOrVideoStream] = [] self.__decode = decode self.__audio: Optional[PlayerStreamTrack] = None self.__video: Optional[PlayerStreamTrack] = None for stream in self.__container.streams: if stream.type == "audio" and not self.__audio: if self.__decode: self.__audio = PlayerStreamTrack(self, kind="audio") self.__streams.append(stream) elif stream.codec_context.name in ["opus", "pcm_alaw", "pcm_mulaw"]: self.__audio = PlayerStreamTrack(self, kind="audio") self.__streams.append(stream) elif stream.type == "video" and not self.__video: if self.__decode: self.__video = PlayerStreamTrack(self, kind="video") self.__streams.append(stream) elif stream.codec_context.name in ["h264", "vp8"]: self.__video = PlayerStreamTrack(self, kind="video") self.__streams.append(stream) # check whether we need to throttle playback container_format = set(self.__container.format.name.split(",")) self._throttle_playback = not container_format.intersection(REAL_TIME_FORMATS) # check whether the looping is supported assert not loop or self.__container.duration is not None, ( "The `loop` argument requires a seekable file" ) self._loop_playback = loop @property def audio(self) -> Optional[MediaStreamTrack]: """ A :class:`aiortc.MediaStreamTrack` instance if the file contains audio. """ return self.__audio @property def video(self) -> Optional[MediaStreamTrack]: """ A :class:`aiortc.MediaStreamTrack` instance if the file contains video. """ return self.__video def _start(self, track: PlayerStreamTrack) -> None: self.__started.add(track) if self.__thread is None: self.__log_debug("Starting worker thread") self.__thread_quit = threading.Event() self.__thread = threading.Thread( name="media-player", target=player_worker_decode if self.__decode else player_worker_demux, args=( asyncio.get_event_loop(), self.__container, self.__streams, self.__audio, self.__video, self.__thread_quit, self._throttle_playback, self._loop_playback, ), ) self.__thread.start() def _stop(self, track: PlayerStreamTrack) -> None: self.__started.discard(track) if not self.__started and self.__thread is not None: self.__log_debug("Stopping worker thread") self.__thread_quit.set() self.__thread.join() self.__thread = None if not self.__started and self.__container is not None: self.__container.close() self.__container = None def __log_debug(self, msg: str, *args: object) -> None: logger.debug(f"MediaPlayer(%s) {msg}", self.__container.name, *args) class MediaRecorderContext: def __init__(self, stream: _AudioOrVideoStream) -> None: self.started = False self.stream = stream self.task: Optional[asyncio.Task[None]] = None class MediaRecorder: """ A media sink that writes audio and/or video to a file. Examples: .. code-block:: python # Write to a video file. player = MediaRecorder('/path/to/file.mp4') # Write to a set of images. player = MediaRecorder('/path/to/file-%3d.png') :param file: The path to a file, or a file-like object. :param format: The format to use, defaults to autodect. :param options: Additional options to pass to FFmpeg. """ def __init__( self, file: Any, format: Optional[str] = None, options: Optional[dict[str, str]] = None, ) -> None: self.__container: Optional[av.container.OutputContainer] = av.open( file=file, format=format, mode="w", options=options ) self.__tracks: dict[MediaStreamTrack, MediaRecorderContext] = {} def addTrack(self, track: MediaStreamTrack) -> None: """ Add a track to be recorded. :param track: A :class:`aiortc.MediaStreamTrack`. """ stream: Union[AudioStream, VideoStream] if track.kind == "audio": if self.__container.format.name in ("wav", "alsa", "pulse"): codec_name = "pcm_s16le" elif self.__container.format.name == "mp3": codec_name = "mp3" elif self.__container.format.name in ("ogg", "opus", "webm"): codec_name = "libopus" else: codec_name = "aac" stream = cast(AudioStream, self.__container.add_stream(codec_name)) else: if self.__container.format.name == "image2": stream = self.__container.add_stream("png", rate=30) stream.pix_fmt = "rgb24" elif self.__container.format.name == "webm": stream = self.__container.add_stream("libvpx", rate=30) stream.pix_fmt = "yuv420p" else: stream = self.__container.add_stream("libx264", rate=30) stream.pix_fmt = "yuv420p" self.__tracks[track] = MediaRecorderContext(stream) async def start(self) -> None: """ Start recording. """ for track, context in self.__tracks.items(): if context.task is None: context.task = asyncio.ensure_future(self.__run_track(track, context)) async def stop(self) -> None: """ Stop recording. """ if self.__container is not None: for track, context in self.__tracks.items(): if context.task is not None: context.task.cancel() context.task = None for packet in context.stream.encode(None): self.__container.mux(packet) self.__tracks = {} self.__container.close() self.__container = None async def __run_track( self, track: MediaStreamTrack, context: MediaRecorderContext ) -> None: while True: try: frame = await track.recv() except MediaStreamError: return assert isinstance(frame, (AudioFrame, VideoFrame)), ( "Only audio or video frames can be recorded" ) if not context.started: # Adjust the output size to match the first frame. if isinstance(context.stream, VideoStream) and isinstance( frame, VideoFrame ): context.stream.width = frame.width context.stream.height = frame.height context.started = True for packet in context.stream.encode(frame): # type: ignore self.__container.mux(packet) class RelayStreamTrack(MediaStreamTrack): def __init__( self, relay: "MediaRelay", source: MediaStreamTrack, buffered: bool, ) -> None: super().__init__() self.kind = source.kind self._relay = relay self._source: Optional[MediaStreamTrack] = source self._buffered = buffered self._frame: Union[Frame, Packet, None] = None self._queue: Optional[asyncio.Queue[Union[Frame, Packet, None]]] = None self._new_frame_event: Optional[asyncio.Event] = None if self._buffered: self._queue = asyncio.Queue() else: self._new_frame_event = asyncio.Event() async def recv(self) -> Union[Frame, Packet]: if self.readyState != "live": raise MediaStreamError self._relay._start(self) if self._buffered: self._frame = await self._queue.get() else: await self._new_frame_event.wait() self._new_frame_event.clear() if self._frame is None: self.stop() raise MediaStreamError return self._frame def stop(self) -> None: super().stop() if self._relay is not None: self._relay._stop(self) self._relay = None self._source = None class MediaRelay: """ A media source that relays one or more tracks to multiple consumers. This is especially useful for live tracks such as webcams or media received over the network. """ def __init__(self) -> None: self.__proxies: dict[MediaStreamTrack, set[RelayStreamTrack]] = {} self.__tasks: dict[MediaStreamTrack, asyncio.Future[None]] = {} def subscribe( self, track: MediaStreamTrack, buffered: bool = True ) -> MediaStreamTrack: """ Create a proxy around the given `track` for a new consumer. :param track: Source :class:`MediaStreamTrack` which is relayed. :param buffered: Whether there need a buffer between the source track and relayed track. :rtype: :class: MediaStreamTrack """ proxy = RelayStreamTrack(self, track, buffered) self.__log_debug("Create proxy %s for source %s", id(proxy), id(track)) if track not in self.__proxies: self.__proxies[track] = set() return proxy def _start(self, proxy: RelayStreamTrack) -> None: track = proxy._source if track is not None and track in self.__proxies: # register proxy if proxy not in self.__proxies[track]: self.__log_debug("Start proxy %s", id(proxy)) self.__proxies[track].add(proxy) # start worker if track not in self.__tasks: self.__tasks[track] = asyncio.ensure_future(self.__run_track(track)) def _stop(self, proxy: RelayStreamTrack) -> None: track = proxy._source if track is not None and track in self.__proxies: # unregister proxy self.__log_debug("Stop proxy %s", id(proxy)) self.__proxies[track].discard(proxy) def __log_debug(self, msg: str, *args: object) -> None: logger.debug(f"MediaRelay(%s) {msg}", id(self), *args) async def __run_track(self, track: MediaStreamTrack) -> None: self.__log_debug("Start reading source %s" % id(track)) while True: try: frame = await track.recv() except MediaStreamError: frame = None for proxy in self.__proxies[track]: if proxy._buffered: proxy._queue.put_nowait(frame) else: proxy._frame = frame proxy._new_frame_event.set() if frame is None: break self.__log_debug("Stop reading source %s", id(track)) del self.__proxies[track] del self.__tasks[track]