From f960ec4bf43b07cf0f405461099f584a53a569bc Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Tue, 31 Oct 2023 01:41:13 +0100 Subject: [PATCH] [`sound`] Plugin refactor. - Added `input_format`/`output_format` options to both input and output audio streams. - Replaced the previous (confusing) occurrences of `ffmpeg_format` and `format`. - Added custom `dtype` option for `sound.play`. - Added `join` flag (default: false) to `sound.play` to wait for the playback to finish. --- platypush/plugins/sound/__init__.py | 28 ++++++++- platypush/plugins/sound/_converters/_base.py | 62 +++++-------------- .../plugins/sound/_converters/_from_raw.py | 35 ++++++++++- .../plugins/sound/_converters/_to_raw.py | 33 +++++++++- platypush/plugins/sound/_manager/_main.py | 3 + platypush/plugins/sound/_streams/_base.py | 9 +++ .../sound/_streams/_player/_resource.py | 1 + platypush/plugins/sound/_streams/_recorder.py | 2 +- 8 files changed, 115 insertions(+), 58 deletions(-) diff --git a/platypush/plugins/sound/__init__.py b/platypush/plugins/sound/__init__.py index d3fa12de08..5cfe958cab 100644 --- a/platypush/plugins/sound/__init__.py +++ b/platypush/plugins/sound/__init__.py @@ -86,8 +86,11 @@ class SoundPlugin(RunnablePlugin): sample_rate: Optional[int] = None, channels: int = 2, volume: float = 100, + dtype: Optional[str] = None, + format: Optional[str] = None, # pylint: disable=redefined-builtin stream_name: Optional[str] = None, stream_index: Optional[int] = None, + join: bool = False, ): """ Plays an audio file/URL (any audio format supported by ffmpeg works) or @@ -159,6 +162,12 @@ class SoundPlugin(RunnablePlugin): :param channels: Number of audio channels. Default: number of channels in the audio file in file mode, 1 if in synth mode :param volume: Playback volume, between 0 and 100. Default: 100. + :param dtype: Data type for the audio samples, if playing raw PCM audio + frames. Supported types: 'float64', 'float32', 'int32', 'int16', + 'int8', 'uint8'. + :param format: Output audio format, if you want to convert the audio to + another format before playing it. The list of available formats can + be retrieved through the ``ffmpeg -formats`` command. Default: None :param stream_index: If specified, play to an already active stream index (you can get them through :meth:`.query_streams`). Default: creates a new audio stream through PortAudio. @@ -167,6 +176,8 @@ class SoundPlugin(RunnablePlugin): name will be created. If not set, and ``stream_index`` is not set either, then a new stream will be created on the next available index and named ``platypush-stream-``. + :param join: If True, then the method will block until the playback is + completed. Default: False. """ dev = self._manager.get_device(device=device, type=StreamType.OUTPUT) @@ -193,7 +204,13 @@ class SoundPlugin(RunnablePlugin): stream_index, ) - self._manager.create_player( + player_kwargs = {} + if dtype: + player_kwargs['dtype'] = dtype + if format: + player_kwargs['format'] = format + + player = self._manager.create_player( device=dev.index, infile=resource, sound=sound, @@ -203,7 +220,12 @@ class SoundPlugin(RunnablePlugin): channels=channels, volume=volume, stream_name=stream_name, - ).start() + **player_kwargs, + ) + + player.start() + if join: + player.join() @action def stream_recording(self, *args, **kwargs): @@ -255,7 +277,7 @@ class SoundPlugin(RunnablePlugin): :param sample_rate: Recording sample rate (default: device default rate) :param dtype: Data type for the audio samples. Supported types: 'float64', 'float32', 'int32', 'int16', 'int8', 'uint8'. Default: - float32 + int16. :param blocksize: Audio block size (default: configured `input_blocksize` or 2048) :param play_audio: If True, then the recorded audio will be played in diff --git a/platypush/plugins/sound/_converters/_base.py b/platypush/plugins/sound/_converters/_base.py index 3c81b90849..17f57310c1 100644 --- a/platypush/plugins/sound/_converters/_base.py +++ b/platypush/plugins/sound/_converters/_base.py @@ -7,8 +7,9 @@ from threading import Event, RLock, Thread from typing import Any, Callable, Coroutine, Iterable, Optional from platypush.context import get_or_create_event_loop +from platypush.utils import is_debug_enabled -_dtype_to_ffmpeg_format = { +dtype_to_ffmpeg_format = { 'int8': 's8', 'uint8': 'u8', 'int16': 's16le', @@ -46,7 +47,8 @@ class AudioConverter(Thread, ABC): volume: float, dtype: str, chunk_size: int, - format: Optional[str] = None, # pylint: disable=redefined-builtin + input_format: Optional[str] = None, # pylint: disable=redefined-builtin + output_format: Optional[str] = None, # pylint: disable=redefined-builtin on_exit: Optional[Callable[[], Any]] = None, **kwargs, ): @@ -58,24 +60,20 @@ class AudioConverter(Thread, ABC): :param dtype: The (numpy) data type of the raw input/output audio. :param chunk_size: Number of bytes that will be read at once from the ffmpeg process. - :param format: Input/output audio format. + :param input_format: Input audio format. + :param output_format: Output audio format. :param on_exit: Function to call when the ffmpeg process exits. """ super().__init__(*args, **kwargs) - ffmpeg_format = _dtype_to_ffmpeg_format.get(dtype) - assert ffmpeg_format, ( - f'Unsupported data type: {dtype}. Supported data types: ' - f'{list(_dtype_to_ffmpeg_format.keys())}' - ) - self._ffmpeg_bin = ffmpeg_bin - self._ffmpeg_format = ffmpeg_format self._ffmpeg_task: Optional[Coroutine] = None self._sample_rate = sample_rate self._channels = channels self._chunk_size = chunk_size - self._format = format + self._input_format = input_format + self._output_format = output_format + self._dtype = dtype self._closed = False self._out_queue = Queue() self.ffmpeg = None @@ -120,7 +118,8 @@ class AudioConverter(Thread, ABC): """ Set of arguments common to all ffmpeg converter instances. """ - return ('-hide_banner', '-loglevel', 'warning', '-y') + log_level = 'debug' if is_debug_enabled() else 'warning' + return ('-hide_banner', '-loglevel', log_level, '-y') @property @abstractmethod @@ -150,20 +149,6 @@ class AudioConverter(Thread, ABC): return args + ('-channel_layout', 'stereo') return args - @property - def _raw_ffmpeg_args(self) -> Iterable[str]: - """ - Ffmpeg arguments for raw audio input/output given the current - configuration. - """ - return ( - '-f', - self._ffmpeg_format, - '-ar', - str(self._sample_rate), - *self._channel_layout_args, - ) - @property def _audio_volume_args(self) -> Iterable[str]: """ @@ -197,23 +182,6 @@ class AudioConverter(Thread, ABC): """ return PIPE - @property - def _compressed_ffmpeg_args(self) -> Iterable[str]: - """ - Ffmpeg arguments for the compressed audio given the current - configuration. - """ - if not self._format: - return () - - ffmpeg_args = self._format_to_ffmpeg_args.get(self._format) - assert ffmpeg_args, ( - f'Unsupported output format: {self._format}. Supported formats: ' - f'{list(self._format_to_ffmpeg_args.keys())}' - ) - - return ffmpeg_args - async def _audio_proxy(self, timeout: Optional[float] = None): """ Proxy the converted audio stream to the output queue for downstream @@ -236,11 +204,6 @@ class AudioConverter(Thread, ABC): self.logger.info('Running ffmpeg: %s', ' '.join(ffmpeg_args)) - try: - await asyncio.wait_for(self.ffmpeg.wait(), 0.1) - except asyncio.TimeoutError: - pass - while ( self._loop and self.ffmpeg @@ -298,6 +261,9 @@ class AudioConverter(Thread, ABC): self._loop.run_until_complete(self._ffmpeg_task) except RuntimeError as e: self.logger.warning(e) + except Exception as e: + self.logger.warning('Audio converter error: %s', e) + self.logger.exception(e) finally: self.stop() diff --git a/platypush/plugins/sound/_converters/_from_raw.py b/platypush/plugins/sound/_converters/_from_raw.py index f67f94de75..7ca58e73f1 100644 --- a/platypush/plugins/sound/_converters/_from_raw.py +++ b/platypush/plugins/sound/_converters/_from_raw.py @@ -1,6 +1,6 @@ from typing import Iterable -from ._base import AudioConverter +from ._base import AudioConverter, dtype_to_ffmpeg_format class RawInputAudioConverter(AudioConverter): @@ -8,13 +8,42 @@ class RawInputAudioConverter(AudioConverter): Converts raw audio input to a compressed media format. """ + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if self._dtype and not self._input_format: + ffmpeg_format = dtype_to_ffmpeg_format.get(self._dtype) + assert ffmpeg_format, ( + f'Unsupported data type: {self._dtype}. Supported data types: ' + f'{list(dtype_to_ffmpeg_format.keys())}' + ) + + self._input_format = ffmpeg_format + @property def _input_format_args(self) -> Iterable[str]: - return self._raw_ffmpeg_args + args = ( + '-ar', + str(self._sample_rate), + *self._channel_layout_args, + ) + + if self._input_format: + args = ('-f', self._input_format) + args + + return args @property def _output_format_args(self) -> Iterable[str]: - return self._compressed_ffmpeg_args + if not self._output_format: + return () + + ffmpeg_args = self._format_to_ffmpeg_args.get(self._output_format) + assert ffmpeg_args, ( + f'Unsupported output format: {self._output_format}. Supported formats: ' + f'{list(self._format_to_ffmpeg_args.keys())}' + ) + + return ffmpeg_args # vim:sw=4:ts=4:et: diff --git a/platypush/plugins/sound/_converters/_to_raw.py b/platypush/plugins/sound/_converters/_to_raw.py index 024256af12..89f1964c32 100644 --- a/platypush/plugins/sound/_converters/_to_raw.py +++ b/platypush/plugins/sound/_converters/_to_raw.py @@ -1,6 +1,6 @@ from typing import Iterable -from ._base import AudioConverter +from ._base import AudioConverter, dtype_to_ffmpeg_format class RawOutputAudioConverter(AudioConverter): @@ -8,13 +8,40 @@ class RawOutputAudioConverter(AudioConverter): Converts input audio to raw audio output. """ + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if self._dtype and not self._output_format: + ffmpeg_format = dtype_to_ffmpeg_format.get(self._dtype) + assert ffmpeg_format, ( + f'Unsupported data type: {self._dtype}. Supported data types: ' + f'{list(dtype_to_ffmpeg_format.keys())}' + ) + + self._output_format = ffmpeg_format + @property def _input_format_args(self) -> Iterable[str]: - return self._compressed_ffmpeg_args + if not self._input_format: + return () + + ffmpeg_args = self._format_to_ffmpeg_args.get(self._input_format) + if not ffmpeg_args: + return ('-f', self._input_format) + + return ffmpeg_args @property def _output_format_args(self) -> Iterable[str]: - return self._raw_ffmpeg_args + args = ( + '-ar', + str(self._sample_rate), + *self._channel_layout_args, + ) + + if self._output_format: + args = ('-f', self._output_format) + args + + return args class RawOutputAudioFromFileConverter(RawOutputAudioConverter): diff --git a/platypush/plugins/sound/_manager/_main.py b/platypush/plugins/sound/_manager/_main.py index 9278475888..9a1d96cfbb 100644 --- a/platypush/plugins/sound/_manager/_main.py +++ b/platypush/plugins/sound/_manager/_main.py @@ -61,6 +61,7 @@ class AudioManager: duration: Optional[float] = None, sample_rate: Optional[int] = None, dtype: str = 'int16', + format: Optional[str] = None, # pylint: disable=redefined-builtin blocksize: Optional[int] = None, latency: Union[float, str] = 'high', stream_name: Optional[str] = None, @@ -77,6 +78,7 @@ class AudioManager: :param duration: Duration of the stream in seconds. :param sample_rate: Sample rate of the stream. :param dtype: Data type of the stream. + :param format: Output format of the stream. :param blocksize: Block size of the stream. :param latency: Latency of the stream. :param stream_name: Name of the stream. @@ -93,6 +95,7 @@ class AudioManager: blocksize=blocksize or self.output_blocksize, latency=latency, channels=channels, + output_format=format, queue_size=self.queue_size, should_stop=self._should_stop, ) diff --git a/platypush/plugins/sound/_streams/_base.py b/platypush/plugins/sound/_streams/_base.py index 6c369fe544..d352681034 100644 --- a/platypush/plugins/sound/_streams/_base.py +++ b/platypush/plugins/sound/_streams/_base.py @@ -43,6 +43,8 @@ class AudioThread(Thread, ABC): infile: Optional[str] = None, outfile: Optional[str] = None, duration: Optional[float] = None, + input_format: Optional[str] = None, + output_format: Optional[str] = None, latency: Union[float, str] = 'high', redis_queue: Optional[str] = None, should_stop: Optional[Event] = None, @@ -66,6 +68,8 @@ class AudioThread(Thread, ABC): stream. :param outfile: Path to the output file. :param duration: Duration of the audio stream. + :param input_format: Input format override. + :param output_format: Output format override. :param latency: Latency to use. :param redis_queue: Redis queue to use. :param should_stop: Synchronize with upstream stop events. @@ -83,6 +87,8 @@ class AudioThread(Thread, ABC): self.volume = volume self.sample_rate = sample_rate self.dtype = dtype + self.input_format = input_format + self.output_format = output_format self.stream = stream self.duration = duration self.blocksize = blocksize * channels @@ -349,6 +355,9 @@ class AudioThread(Thread, ABC): self.logger.warning( 'Audio callback timeout for %s', self.__class__.__name__ ) + except Exception as e: + self.logger.warning('Unhandled sound on %s', self.__class__.__name__) + self.logger.exception(e) finally: self.notify_stop() diff --git a/platypush/plugins/sound/_streams/_player/_resource.py b/platypush/plugins/sound/_streams/_player/_resource.py index 5186d5e79a..43c4a92f3c 100644 --- a/platypush/plugins/sound/_streams/_player/_resource.py +++ b/platypush/plugins/sound/_streams/_player/_resource.py @@ -20,6 +20,7 @@ class AudioResourcePlayer(AudioPlayer): def _converter_args(self) -> dict: return { 'infile': self.infile, + 'output_format': self.output_format, **super()._converter_args, } diff --git a/platypush/plugins/sound/_streams/_recorder.py b/platypush/plugins/sound/_streams/_recorder.py index abec81da16..5d6ea71cc1 100644 --- a/platypush/plugins/sound/_streams/_recorder.py +++ b/platypush/plugins/sound/_streams/_recorder.py @@ -81,7 +81,7 @@ class AudioRecorder(AudioThread): @property def _converter_args(self) -> dict: return { - 'format': self.output_format, + 'output_format': self.output_format, **super()._converter_args, }