diff --git a/docs/source/_static/sample_audio/multichannel/multichannel_220925_223450.wav b/docs/source/_static/sample_audio/multichannel/multichannel_220925_223450.wav new file mode 100644 index 000000000..d7f763607 Binary files /dev/null and b/docs/source/_static/sample_audio/multichannel/multichannel_220925_223450.wav differ diff --git a/docs/source/example_multichannel.ipynb b/docs/source/example_multichannel.ipynb new file mode 100644 index 000000000..a378338f4 --- /dev/null +++ b/docs/source/example_multichannel.ipynb @@ -0,0 +1,276 @@ +{ + "cells": [ + { + "cell_type": "code", + "id": "initial_id", + "metadata": { + "collapsed": true, + "tags": [ + "remove-cell" + ] + }, + "source": [ + "# Executing this cell will:\n", + "\n", + "# Disable all TQDM outputs in stdout.\n", + "import os\n", + "\n", + "os.environ[\"DISABLE_TQDM\"] = \"True\"\n", + "\n", + "# Setup the python logger for the Public API\n", + "from osekit import setup_logging\n", + "\n", + "setup_logging() # Overwrites the default logger to" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "8c395a2079d86493", + "metadata": {}, + "source": [ + "# Working with multichannel audio files [^download]\n", + "\n", + "[^download]: This notebook can be downloaded as **{nb-download}`example_multichannel.ipynb`**." + ] + }, + { + "cell_type": "markdown", + "id": "90049102bdc38599", + "metadata": {}, + "source": [ + "# Basics: Core API\n", + "\n", + "## Parsing a multichannel file\n", + "\n", + "The `AudioFile.channels` property indicates the number of channels that a given `AudioFile` has.\n", + "\n", + "The `AudioData.channels` property is a **list of ints** that represents the targeted channels of the file:" + ] + }, + { + "cell_type": "code", + "id": "c1f1dcd7e3e90141", + "metadata": {}, + "source": [ + "from pathlib import Path\n", + "from osekit.core.audio_file import AudioFile\n", + "from osekit.core.audio_data import AudioData\n", + "\n", + "af = AudioFile(\n", + " path=Path(\"_static/sample_audio/multichannel/multichannel_220925_223450.wav\"),\n", + " strptime_format=r\"%y%m%d_%H%M%S\",\n", + ")\n", + "\n", + "print(f\"The audio file has {af.channels} channels.\")\n", + "\n", + "ad: AudioData = AudioData.from_files([af])\n", + "\n", + "print(f\"By default, all channels are targeted: {ad.channels}.\")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "fd8e3d095a534d5", + "metadata": {}, + "source": [ + "## Targeting specific channel(s)\n", + "\n", + "`AudioData.channels` can be set to target specific channel(s):" + ] + }, + { + "cell_type": "code", + "id": "ea1c469c76a14fe3", + "metadata": {}, + "source": [ + "import matplotlib.pyplot as plt\n", + "\n", + "ad.channels = [0, 2] # Removing channel 1 from targeted channels\n", + "ad.plot()\n", + "plt.show()" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "1b00f1600c66e3c6", + "metadata": {}, + "source": [ + "## Computing the spectrum of a specific channel\n", + "\n", + "`SpectroData` target **a specific** channel of a file:" + ] + }, + { + "cell_type": "code", + "id": "87f3839389f684c3", + "metadata": {}, + "source": [ + "from osekit.core.spectro_data import SpectroData\n", + "from scipy.signal import ShortTimeFFT, windows\n", + "\n", + "sd = SpectroData.from_audio_data(\n", + " data=ad,\n", + " fft=ShortTimeFFT(win=windows.hamming(1024), hop=128, fs=ad.sample_rate),\n", + ")\n", + "\n", + "sd.audio_channel = 2 # Targets the third channel of the file (which is the second channel of the AudioData)\n", + "sd.plot()\n", + "plt.show()" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "8bfbe3f450ad73cd", + "metadata": {}, + "source": "# Public API" + }, + { + "cell_type": "markdown", + "id": "e2d5321198880205", + "metadata": {}, + "source": [ + "## Build the Project\n", + "\n", + "First, we have to build the project from the raw audio files:" + ] + }, + { + "cell_type": "code", + "id": "3ab3cb447c59a857", + "metadata": {}, + "source": [ + "from pathlib import Path\n", + "from osekit.public.project import Project\n", + "\n", + "folder = Path(r\"_static/sample_audio/multichannel\")\n", + "strptime_format = r\"%y%m%d_%H%M%S\"\n", + "\n", + "project = Project(\n", + " folder=folder,\n", + " strptime_format=strptime_format,\n", + ")\n", + "\n", + "project.build()" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "2f5510c9c396ee2f", + "metadata": {}, + "source": [ + "## Declare the Transform\n", + "\n", + "Then we **declare** a `Transform` which would work on the audio (e.g. export spectrograms):" + ] + }, + { + "cell_type": "code", + "id": "d993991e8c23a2c0", + "metadata": {}, + "source": [ + "from osekit.public.transform import Transform, OutputType\n", + "from scipy.signal import ShortTimeFFT\n", + "from scipy.signal.windows import hamming\n", + "\n", + "transform = Transform(\n", + " output_type=OutputType.SPECTROGRAM,\n", + " fft=ShortTimeFFT(win=hamming(1024), hop=128, fs=project.origin_dataset.sample_rate),\n", + " name=\"one_spectrogram_per_channel\",\n", + ")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "7f58320386d3dc14", + "metadata": {}, + "source": "Now, we will use some **Core API** on top of the **Public API** to get one spectrogram per channel:" + }, + { + "cell_type": "code", + "id": "c6b54cf5f1ecb44b", + "metadata": {}, + "source": [ + "import copy\n", + "\n", + "# We get the transform SpectroData(s) -- here there is only one\n", + "sds = project.prepare_spectro(transform=transform)\n", + "\n", + "# We'll create one spectro data per channel:\n", + "sds_channels = []\n", + "for sd in sds.data:\n", + " for channel in (0, 2): # Targeting the specific channels here\n", + " sd_copy = copy.copy(sd)\n", + " sd_copy.audio_channel = channel\n", + " sd_copy.name += \"_channel_\" + str(channel)\n", + " sds_channels.append(sd_copy)\n", + "\n", + "# We'll then set the sds data as sd_channels:\n", + "sds.data = sds_channels\n", + "\n", + "# Let's check everything's ok:\n", + "for sd in sds.data:\n", + " print(f\"Spectrogram for channel {sd.audio_channel}\")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "a119e3339b02e775", + "metadata": {}, + "source": "We should now be able to run the transform on the edited `SpectroDataset`:" + }, + { + "cell_type": "code", + "id": "1948b260fcaf03ab", + "metadata": {}, + "source": "project.run(transform=transform, spectro_dataset=sds)", + "outputs": [], + "execution_count": null + }, + { + "cell_type": "code", + "id": "449dc442b4a5df75", + "metadata": {}, + "source": [ + "# Reset the project to get all files back to place.\n", + "project.reset()" + ], + "outputs": [], + "execution_count": null + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 2 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython2", + "version": "2.7.6" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/source/examples.rst b/docs/source/examples.rst index 37dbf83a1..37f06ac24 100644 --- a/docs/source/examples.rst +++ b/docs/source/examples.rst @@ -57,10 +57,17 @@ In the ``docs/source/_static/sample_audio/timestamped`` folder, files are just n =========== +.. topic:: :doc:`Work with multichannel audio files ` + + Work on multichannel audio, compute spectrums from specific channels... + +=========== + .. topic:: :doc:`Use APLOSE results ` Parse `APLOSE `_ results csv files in OSEkit, and use the detections to filter the audio or spectrograms from the project. + .. toctree :: :hidden: @@ -70,4 +77,5 @@ In the ``docs/source/_static/sample_audio/timestamped`` folder, files are just n example_multiple_spectrograms example_multiple_spectrograms_id example_ltas + example_multichannel example_aplose_result diff --git a/docs/source/multichannel.rst b/docs/source/multichannel.rst new file mode 100644 index 000000000..c0253c553 --- /dev/null +++ b/docs/source/multichannel.rst @@ -0,0 +1,46 @@ +.. _multichannel: + +Working with multichannel audio files +------------------------------------- + +The audio classes in **OSEkit** allow multichannel audio handling, e.g. the :meth:`osekit.core.audio_data.AudioData.get_value` method returns a ``samples x channels`` matrix. + +The :attr:`osekit.core.audio_file.AudioFile.channels` property depicts the number of channels of a given ``AudioFile``: + +.. code-block:: python + + from pathlib import Path + from osekit.core.audio_file import AudioFile + + af = AudioFile(...) + print(af.channels) + + >>> 3 + +The :attr:`osekit.core.audio_data.AudioData.channels` property relates to the list of channels concerned by this audio data: + +.. code-block:: python + + from osekit.core.audio_data import AudioData + + ad = AudioData.from_files(files=[af]) + ad.channels = [0,2] # We want to keep only channels 0 and 2 + print(ad.get_value().shape[1]) + + >>> 2 # One value array per channel + +Finally, a ``SpectroData`` that has a linked ``AudioData`` targets a **specific channel** of this ``AudioData``: + +.. code-block:: python + + from osekit.core.spectro_data import SpectroData + from scipy.signal import ShortTimeFFT + + sd = SpectroData.from_audio_data(data=ad, ...) + sd.audio_channel = 2 # The spectrum will be computed on the channel 2 of the file + +.. important:: + + The :attr:`osekit.core.spectro_data.SpectroData.audio_channel` value refers to the index of the channel of the **file**. + + If, as in the example above, the ``AudioFile`` has 3 channels ``[0,1,2]``, the ``AudioData`` targets channels ``[0,2]`` and the ``SpectroData`` targets the channel ``2``, the spectrum will be computed on the **third channel** of the file (with index ``2``). diff --git a/docs/source/usage.rst b/docs/source/usage.rst index 68aa8f974..c09d38354 100644 --- a/docs/source/usage.rst +++ b/docs/source/usage.rst @@ -28,3 +28,4 @@ The package combines two APIs: multiprocessing jobs aplose + multichannel diff --git a/src/osekit/core/audio_data.py b/src/osekit/core/audio_data.py index 49dbd1767..473a91393 100644 --- a/src/osekit/core/audio_data.py +++ b/src/osekit/core/audio_data.py @@ -49,6 +49,7 @@ def __init__( normalization: Normalization = Normalization.RAW, normalization_values: dict | None = None, butter: Butterworth | None = None, + channels: list[int] | None = None, ) -> None: """Initialize an ``AudioData`` from a list of ``AudioItems``. @@ -73,6 +74,8 @@ def __init__( The type of normalization to apply to the audio data. butter: Butterworth | None Butterworth filter to apply to the audio data. + channels: list[int] + Considered channels of the linked audio file(s). """ super().__init__(items=items, begin=begin, end=end, name=name) @@ -81,13 +84,12 @@ def __init__( self.normalization = normalization self.normalization_values = normalization_values self.butter = butter + self.channels = channels @property def nb_channels(self) -> int: """Number of channels of the audio data.""" - return max( - [1] + [item.nb_channels for item in self.items if type(item) is AudioItem], - ) + return len(self.channels) @property def shape(self) -> tuple[int, int]: @@ -131,6 +133,21 @@ def butter(self) -> Butterworth: def butter(self, value: Butterworth) -> None: self._butter = value + @property + def channels(self) -> list[int]: + """The Butterworth filter to apply to the audio data.""" + return self._channels + + @channels.setter + def channels(self, value: list[int] | None) -> None: + if value is None: + nb_channels_max = max( + [1] + + [item.nb_channels for item in self.items if type(item) is AudioItem], + ) + value = list(range(nb_channels_max)) + self._channels = value + @classmethod def _make_item( cls, @@ -186,9 +203,9 @@ def get_normalization_values(self) -> dict: """ values = np.array(self.get_filtered_value()) self.normalization_values = { - "mean": values.mean(), - "peak": values.max(), - "std": values.std(), + "mean": values.mean(axis=0), + "peak": values.max(axis=0), + "std": values.std(axis=0), } return self.normalization_values @@ -309,7 +326,7 @@ def stream(self, chunk_size: int = 8192) -> Generator[np.ndarray, None, None]: ) for chunk in item.stream(chunk_size=chunk_size): - y = chunk + y = chunk[:, self.channels] if item.sample_rate != self.sample_rate: y = resampler.resample_chunk(x=chunk) @@ -385,13 +402,18 @@ def plot( to the ``matplotlib.axes._axes.Axes.plot()`` method. """ - ax = ax if ax is not None else get_default_axes() + ax = ax if ax is not None else get_default_axes(nb_rows=self.nb_channels) values = self.get_value() if values is None else values time = pd.date_range(start=self.begin, end=self.end, periods=values.shape[0]) - ax.xaxis_date() - ax.plot(time, values, **kwargs) + if type(ax) is plt.Axes: + ax.xaxis_date() + ax.plot(time, values, **kwargs) + if type(ax) is np.ndarray: # Multichannel audio + for idx, axes in enumerate(ax): + axes.xaxis_date() + axes.plot(time, values[:, idx], **kwargs) def write( self, @@ -608,6 +630,7 @@ def to_dict(self) -> dict: "sample_rate": self.sample_rate, "normalization": self.normalization.value, "normalization_values": self.normalization_values, + "channels": self.channels, } ) @@ -663,6 +686,7 @@ def _from_base_dict( normalization=Normalization(dictionary["normalization"]), normalization_values=dictionary.get("normalization_values", None), butter=butter, + channels=dictionary.get("channels", None), ) @classmethod diff --git a/src/osekit/core/spectro_data.py b/src/osekit/core/spectro_data.py index d0aab90b5..a0cc480b9 100644 --- a/src/osekit/core/spectro_data.py +++ b/src/osekit/core/spectro_data.py @@ -49,7 +49,8 @@ class SpectroData(BaseData[SpectroItem, SpectroFile]): def __init__( self, items: list[SpectroItem] | None = None, - audio_data: AudioData = None, + audio_data: AudioData | None = None, + audio_channel: int = 0, begin: Timestamp | None = None, end: Timestamp | None = None, name: str | None = None, @@ -69,6 +70,8 @@ def __init__( List of the ``SpectroItem`` constituting the ``SpectroData``. audio_data: AudioData The ``AudioData`` from which to compute the spectrogram. + audio_channel: int + Channel of the linked ``AudioData`` concerned by this ``SpectroData``. begin: Timestamp | None Only effective if items is ``None``. Set the begin of the empty data. @@ -90,6 +93,7 @@ def __init__( """ super().__init__(items=items, begin=begin, end=end, name=name) self.audio_data = audio_data + self.audio_channel = audio_channel self.fft = fft self._sx_dtype = complex self._db_ref = db_ref @@ -131,6 +135,21 @@ def shape(self) -> tuple[int, ...]: int(self.fft.fs * self.duration.total_seconds()), ) + @property + def audio_channel(self) -> int: + """Channel of the linked ``AudioData`` concerned by this ``SpectroData``.""" + return self._audio_channel + + @audio_channel.setter + def audio_channel(self, channel: int) -> None: + if self.audio_data and channel not in self.audio_data.channels: + msg = ( + f"Can't target audio channel {channel}: AudioData only targets " + f"channels {self.audio_data.channels}." + ) + raise ValueError(msg) + self._audio_channel = channel + @property def nb_bytes(self) -> int: """Total bytes consumed by the spectro values.""" @@ -234,8 +253,8 @@ def get_value(self) -> np.ndarray: sx = self.fft.stft( x=self.audio_data.get_value_calibrated()[ :, - 0, - ], # Only consider the 1st channel + self.audio_data.channels.index(self.audio_channel), + ], padding="zeros", ) @@ -885,6 +904,7 @@ def from_audio_data( fft: ShortTimeFFT, v_lim: tuple[float, float] | None = None, colormap: str | None = None, + **kwargs: dict, ) -> SpectroData: """Instantiate a ``SpectroData`` object from a ``AudioData`` object. @@ -899,6 +919,8 @@ def from_audio_data( for plotting the spectrogram. colormap: str Colormap to use for plotting the spectrogram. + kwargs: dict + Additionnal kwargs to pass to the SpectroData constructor. Returns ------- @@ -913,6 +935,7 @@ def from_audio_data( end=data.end, v_lim=v_lim, colormap=colormap, + **kwargs, ) def to_dict(self, *, embed_sft: bool = True) -> dict: @@ -959,7 +982,11 @@ def to_dict(self, *, embed_sft: bool = True) -> dict: base_dict | audio_dict | sft_dict - | {"v_lim": self.v_lim, "colormap": self.colormap} + | { + "v_lim": self.v_lim, + "colormap": self.colormap, + "audio_channel": self.audio_channel, + } ) @classmethod @@ -1006,6 +1033,7 @@ def from_dict( sft, v_lim=v_lim, colormap=dictionary["colormap"], + audio_channel=dictionary.get("audio_channel", 0), ) if dictionary["files"]: diff --git a/src/osekit/utils/audio.py b/src/osekit/utils/audio.py index 38d2e6335..f667b5fa4 100644 --- a/src/osekit/utils/audio.py +++ b/src/osekit/utils/audio.py @@ -125,27 +125,37 @@ def normalize_raw(values: np.ndarray) -> np.ndarray: def normalize_dc_reject( values: np.ndarray, - dc_component: float | None = None, + dc_component: float | list[float] | None = None, ) -> np.ndarray: """Reject the DC component of the audio data.""" - return values - (values.mean() if dc_component is None else dc_component) + return values - (values.mean(axis=0) if dc_component is None else dc_component) -def normalize_peak(values: np.ndarray, peak: float | None = None) -> np.ndarray: +def normalize_peak( + values: np.ndarray, peak: float | list[float] | None = None +) -> np.ndarray: """Return values normalized so that the peak value is ``1.0``.""" - divisor = max(abs(values)) if peak is None else peak - return values / (divisor if divisor else 1) + divisor = abs(values).max(axis=0) if peak is None else peak + if np.isscalar(divisor): + divisor = divisor or 1 + else: + divisor[divisor == 0] = 1 + return values / divisor def normalize_zscore( values: np.ndarray, - mean: float | None = None, - std: float | None = None, + mean: float | list[float] | None = None, + std: float | list[float] | None = None, ) -> np.ndarray: """Return normalized zscore from the audio data.""" - mean = values.mean() if mean is None else mean - std = values.std() if std is None else std - return (values - mean) / (std if std else 1) + mean = values.mean(axis=0) if mean is None else mean + std = values.std(axis=0) if std is None else std + if np.isscalar(std): + std = std or 1 + else: + std[std == 0] = 1 + return (values - mean) / std class NormalizationValider(enum.EnumMeta): @@ -194,9 +204,9 @@ class Normalization(enum.Flag, metaclass=NormalizationValider): def normalize( values: np.ndarray, normalization: Normalization, - mean: float | None = None, - peak: float | None = None, - std: float | None = None, + mean: float | list[float] | None = None, + peak: float | list[float] | None = None, + std: float | list[float] | None = None, ) -> np.ndarray: """Normalize the audio data.""" if Normalization.DC_REJECT in normalization: diff --git a/src/osekit/utils/plot.py b/src/osekit/utils/plot.py index 1d91787f0..45de9999e 100644 --- a/src/osekit/utils/plot.py +++ b/src/osekit/utils/plot.py @@ -1,7 +1,8 @@ +import numpy as np from matplotlib import pyplot as plt -def get_default_axes() -> plt.Axes: +def get_default_axes(nb_rows: int = 1, nb_cols: int = 1) -> plt.Axes | np.ndarray: """Return a default-formatted ``Axes`` on a new figure. By default, OSEkit plots on wide, borderless figures. @@ -9,25 +10,31 @@ def get_default_axes() -> plt.Axes: Returns ------- - plt.Axes: + plt.Axes | np.ndarray: The default ``Axes`` on a new figure. + If nb_rows > 1, returns a np.ndarray of plt.Axes. """ # Legacy OSEkit behaviour. - _, ax = plt.subplots( - nrows=1, - ncols=1, + _, axs = plt.subplots( + nrows=nb_rows, + ncols=nb_cols, figsize=(1813 / 100, 512 / 100), dpi=100, ) - ax.get_xaxis().set_visible(False) - ax.get_yaxis().set_visible(False) - ax.set_frame_on(False) - ax.spines["right"].set_visible(False) - ax.spines["left"].set_visible(False) - ax.spines["bottom"].set_visible(False) - ax.spines["top"].set_visible(False) + # Skim through both 1D and 2D ax arrays + axs_array = axs if type(axs) is np.ndarray else [axs] + for outer in axs_array: + inner_axs_array = outer if type(outer) is np.ndarray else [outer] + for ax in inner_axs_array: + ax.get_xaxis().set_visible(False) + ax.get_yaxis().set_visible(False) + ax.set_frame_on(False) + ax.spines["right"].set_visible(False) + ax.spines["left"].set_visible(False) + ax.spines["bottom"].set_visible(False) + ax.spines["top"].set_visible(False) plt.axis("off") plt.subplots_adjust( top=1, @@ -37,4 +44,4 @@ def get_default_axes() -> plt.Axes: hspace=0, wspace=0, ) - return ax + return axs diff --git a/tests/helpers/audio.py b/tests/helpers/audio.py index 726126a81..cc5fa5538 100644 --- a/tests/helpers/audio.py +++ b/tests/helpers/audio.py @@ -1,9 +1,56 @@ import typing +from pathlib import Path import numpy as np -from pandas import Timestamp +from pandas import Timedelta, Timestamp from osekit.core.audio_data import AudioData +from osekit.core.audio_file import AudioFile + + +class MockedAudioFile(AudioFile): + def __init__( + self, + mocked_value: np.ndarray, + *args: typing.Any, + **kwargs: typing.Any, + ) -> None: + defaults = { + "begin": Timestamp("2000-01-01 00:00:00"), + "path": Path("foo"), + } + for key, value in defaults.items(): + if key not in kwargs: + kwargs.update(**{key: value}) + + if mocked_value.ndim == 1: + mocked_value = mocked_value[:, None] + + self.mocked_value = mocked_value + self.channels = self.mocked_value.shape[1] + self.sample_rate = kwargs.get("sample_rate", 48000) + self.begin = kwargs.pop("begin") + self.__dict__.update(kwargs) + self.end = self.begin + Timedelta( + seconds=mocked_value.shape[0] / self.sample_rate + ) + self.pointer = 0 + + def read(self, start: Timestamp, stop: Timestamp) -> np.ndarray: + start_sample, stop_sample = self.frames_indexes(start, stop) + pointer = self.pointer + self.seek(start_sample) + vs = self.stream(chunk_size=stop_sample - start_sample) + self.pointer = pointer + return vs + + def stream(self, chunk_size: int) -> np.ndarray: + values = self.mocked_value[self.pointer : self.pointer + chunk_size] + self.pointer += chunk_size + return values + + def seek(self, frame: int) -> None: + self.pointer = frame class MockedAudioData(AudioData): diff --git a/tests/test_audio.py b/tests/test_audio.py index 7d4423462..71659c8b9 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -36,7 +36,7 @@ normalize, ) from osekit.utils.plot import get_default_axes -from tests.helpers.audio import MockedAudioData +from tests.helpers.audio import MockedAudioData, MockedAudioFile def test_mocked_audio_data() -> None: @@ -62,6 +62,50 @@ def test_mocked_audio_data() -> None: ) +def test_mocked_audio_file() -> None: + mocked_value_mono = np.array([1.0, 2.0, 3.0]) + mocked_value_stereo = np.array([[1, 1], [2, 2], [3, 3]]) + + af_mono = MockedAudioFile( + mocked_value=mocked_value_mono, sample_rate=len(mocked_value_mono) + ) + + af_stereo = MockedAudioFile( + mocked_value=mocked_value_stereo, + ) + + assert af_mono.channels == 1 + assert af_stereo.channels == 2 + + # sample_rate equals len + assert af_mono.duration == Timedelta(seconds=1) + + # Mono should be 2D too for compatibility issues + assert np.array_equal( + af_mono.read(af_mono.begin, af_mono.end), mocked_value_mono[:, None] + ) + + # Full time stereo read + assert np.array_equal( + af_stereo.read(af_stereo.begin, af_stereo.end), mocked_value_stereo + ) + + # Specific times + period = Timedelta(seconds=1 / af_mono.sample_rate) + sample_time = af_mono.begin + 2 * period + assert np.array_equal( + af_mono.read(start=sample_time, stop=sample_time), mocked_value_mono[1:2, None] + ) + + # Stream + assert af_mono.pointer == 0 + assert np.array_equal(af_mono.stream(1), mocked_value_mono[0:1, None]) + assert af_mono.pointer == 1 + + af_mono.seek(frame=2) + assert np.array_equal(af_mono.stream(1), mocked_value_mono[2:3, None]) + + @pytest.mark.parametrize( "audio_files", [ @@ -274,36 +318,20 @@ def mocked_init(self: AudioFile, *args: None, **kwargs: None) -> None: assert af.stream(1024).shape == expected_shape -def test_multichannel_audio_file_read(monkeypatch: pytest.MonkeyPatch) -> None: - full_file = np.array([[1, 1, 1], [2, 2, 2], [3, 3, 3], [4, 4, 4], [5, 5, 5]]) +def test_multitchannel_audio_data() -> None: + af_data = np.array([[1, 2, 3] for _ in range(10)]) - def read_patch(*args: list, **kwargs: dict) -> np.ndarray: - start, stop = kwargs["start"], kwargs["stop"] - return full_file[start:stop, :] + af = MockedAudioFile(mocked_value=af_data, sample_rate=10) - monkeypatch.setattr(AudioFileManager, "read", read_patch) + ad: AudioData = AudioData.from_files([af]) - def init_patch(self: AudioFile, *args: list, **kwargs: dict) -> None: - self.begin = kwargs["begin"] - self.path = kwargs["path"] - self.end = kwargs["end"] - self.sample_rate = kwargs["sample_rate"] + # Default channels is all channels + assert ad.channels == [0, 1, 2] + assert np.array_equal(ad.get_value(), af_data) - monkeypatch.setattr(AudioFile, "__init__", init_patch) - - af = AudioFile( - begin=Timestamp("2005-10-18 00:00:00"), - end=Timestamp("2005-10-18 00:00:01"), - path=Path(r"foo"), - sample_rate=5, - ) - - assert np.array_equal(af.read(start=af.begin, stop=af.end), full_file) - - assert np.array_equal( - af.read(start=af.begin, stop=af.begin + Timedelta(seconds=3 / 5)), - full_file[:3, :], - ) + ad.channels = [1, 2] + assert ad.nb_channels == 2 + assert np.array_equal(ad.get_value(), np.array([[2, 3] for _ in range(10)])) @pytest.mark.parametrize( @@ -1863,6 +1891,39 @@ def test_split_data_normalization_pass() -> None: ) +def test_multichannel_data_normalization() -> None: + ad = MockedAudioData(mocked_value=np.array([[1, 2] for _ in range(10)])) + + normalization_values = ad.get_normalization_values() + + assert np.array_equal(normalization_values["mean"], [1.0, 2.0]) + assert np.array_equal(normalization_values["peak"], [1, 2]) + assert np.array_equal(normalization_values["std"], [0.0, 0.0]) + + # Normalization should be channel-wise: + ad.normalization = Normalization.DC_REJECT + assert not np.any(ad.get_value()) + + # Default normalization + for normalization in ( + Normalization.DC_REJECT, + Normalization.ZSCORE, + Normalization.PEAK, + ): + # New AudioData with empty normalization_values + ad2 = MockedAudioData(mocked_value=np.array([[1, 2] for _ in range(10)])) + + ad.normalization = normalization + ad2.normalization = normalization + + assert np.array_equal(ad.get_value(), ad2.get_value()) + + # Normalization deserialization + assert np.array_equal( + ad.normalization_values, AudioData.from_dict(ad.to_dict()).normalization_values + ) + + @pytest.mark.parametrize( ("audio_files", "start_frame", "stop_frame", "expected_begin", "expected_data"), [ @@ -2246,17 +2307,20 @@ def test_butter_audiodataset() -> None: assert ads.butter == butter2 -plot_calls = [] +plot_calls_args = [] +plot_calls_kwargs = [] @pytest.fixture(autouse=False) def patch_plot(monkeypatch: pytest.MonkeyPatch) -> Generator[None, Any, None]: def mock_plot(self: Axes, *args: Any, **kwargs: Any) -> None: - plot_calls.append((self, kwargs)) + plot_calls_args.append((self, args)) + plot_calls_kwargs.append((self, kwargs)) monkeypatch.setattr(plt.Axes, "plot", mock_plot) yield - plot_calls.clear() + plot_calls_args.clear() + plot_calls_kwargs.clear() def test_plot_on_default_axes(patch_plot: None) -> None: @@ -2264,19 +2328,55 @@ def test_plot_on_default_axes(patch_plot: None) -> None: default_axes = get_default_axes() ad.plot() - axes, _ = plot_calls.pop() + axes, _ = plot_calls_kwargs.pop() assert np.array_equal(axes.viewLim, default_axes.viewLim) assert np.array_equal(axes.dataLim, default_axes.dataLim) assert np.array_equal(axes.spines, default_axes.spines) +def test_plot_multichannel_audio_data(patch_plot: None) -> None: + af = MockedAudioFile( + mocked_value=np.array([[1, 2, 3], [1, 2, 3], [1, 2, 3], [1, 2, 3], [1, 2, 3]]) + ) + ad: AudioData = AudioData.from_files([af]) + + ad.plot() + assert len(plot_calls_kwargs) == af.channels + for idx, entry in enumerate(plot_calls_args): + values = entry[1][1] + assert np.array_equal(values, ad.get_value()[:, idx]) + + +@pytest.mark.parametrize( + ("nb_rows", "nb_cols", "expected_type", "expected_shape"), + [ + pytest.param(1, 1, plt.Axes, None, id="only_one_plot"), + pytest.param(1, 3, np.ndarray, (3,), id="multiple_cols"), + pytest.param(3, 1, np.ndarray, (3,), id="multiple_rows"), + pytest.param(3, 4, np.ndarray, (3, 4), id="2d_plot"), + ], +) +def test_default_axes_shape( + nb_rows: int, + nb_cols: int, + expected_type: type[Axes] | type[np.ndarray], + expected_shape: tuple | None, +) -> None: + axs = get_default_axes(nb_rows=nb_rows, nb_cols=nb_cols) + + assert type(axs) is expected_type + + if expected_shape: + assert axs.shape == expected_shape + + def test_plot_on_custom_axes(patch_plot: None) -> None: ad = MockedAudioData(mocked_value=[1, 2, 3]) _, custom_axes = plt.subplots() ad.plot(ax=custom_axes) - used_axes, _ = plot_calls.pop() + used_axes, _ = plot_calls_kwargs.pop() assert custom_axes is used_axes @@ -2285,7 +2385,7 @@ def test_plot_with_kwargs(patch_plot: None) -> None: ad = MockedAudioData(mocked_value=[1, 2, 3]) ad.plot(None, None, velvet="underground", sweet="jane") - _, kwargs = plot_calls.pop() + _, kwargs = plot_calls_kwargs.pop() assert np.array_equal(kwargs, {"velvet": "underground", "sweet": "jane"}) @@ -2308,7 +2408,7 @@ def mocked_get_value(*args: Any, **kwargs: Any) -> np.ndarray: assert get_value_calls[0] == 1 ad.plot(values=vs, the="voidz") - _, kwargs = plot_calls.pop() + _, kwargs = plot_calls_kwargs.pop() # Values are provided and shouldn't be fetched again assert get_value_calls[0] == 1 diff --git a/tests/test_serialization.py b/tests/test_serialization.py index 82460c97f..16a3c93dc 100644 --- a/tests/test_serialization.py +++ b/tests/test_serialization.py @@ -31,6 +31,7 @@ from osekit.core.spectro_dataset import SpectroDataset from osekit.core.spectro_file import SpectroFile from osekit.utils.audio import Normalization +from tests.helpers.audio import MockedAudioFile, MockedAudioData def test_audio_file_from_dict_depends_on_available_info( @@ -234,6 +235,30 @@ def test_audio_data_serialization( assert np.array_equal(ad.get_value(), AudioData.from_dict(ad.to_dict()).get_value()) +def test_audio_data_channels_serialization(monkeypatch: pytest.MonkeyPatch) -> None: + mocked_value = np.array( + [ + [1, 2, 3], + [1, 2, 3], + [1, 2, 3], + [1, 2, 3], + ] + ) + + af = MockedAudioFile( + mocked_value=mocked_value, + ) + + def mocked_make_file(**kwargs: dict) -> MockedAudioFile: + return MockedAudioFile(mocked_value=mocked_value, **kwargs) + + monkeypatch.setattr(AudioData, "_make_file", mocked_make_file) + + ad = AudioData.from_files([af], channels=[0, 2]) + + assert np.array_equal(AudioData.from_dict(ad.to_dict()).channels, [0, 2]) + + @pytest.mark.parametrize( ("audio_files", "data_duration", "sample_rate", "normalization", "name"), [ @@ -733,6 +758,22 @@ def test_spectro_data_serialization( assert SpectroData.from_dict(sd.to_dict(embed_sft=True)).files == sd.files +def test_spectro_data_serialization_channel() -> None: + ad = MockedAudioData( + mocked_value=np.array([[1, 2, 3] for _ in range(100)]), + sample_rate=100, + ) + + ad.channels = [0, 2] + + sd = SpectroData.from_audio_data( + ad, fft=ShortTimeFFT(win=hamming(48), hop=48, fs=100) + ) + sd.audio_channel = 2 + + assert SpectroData.from_dict(sd.to_dict()).audio_channel == sd.audio_channel + + @pytest.mark.parametrize( ( "audio_files", diff --git a/tests/test_spectro.py b/tests/test_spectro.py index 5626128a8..f0dc3f36e 100644 --- a/tests/test_spectro.py +++ b/tests/test_spectro.py @@ -32,7 +32,7 @@ from osekit.core.spectro_file import SpectroFile from osekit.core.spectro_item import SpectroItem from osekit.utils.audio import Normalization, generate_sample_audio -from tests.helpers.audio import MockedAudioData +from tests.helpers.audio import MockedAudioData, MockedAudioFile from tests.helpers.dummy import DummyFile @@ -1468,9 +1468,9 @@ def test_plot_timezone( called_ax = set() def mock_imshow( - self: plt.Axes, - sx: np.ndarray, - **kwargs: str, + self: plt.Axes, + sx: np.ndarray, + **kwargs: str, ) -> None: called_ax.add(self) @@ -1486,8 +1486,6 @@ def mock_imshow( assert spectro_data_axes.xaxis.units is None - - def test_spectro_default_v_lim(audio_files: pytest.fixture) -> None: files, _ = audio_files ad = AudioData.from_files(files) @@ -1943,3 +1941,58 @@ def mock_check_duplicate_data_names( sds.save_all(spectrum_folder=Path("bantam"), spectrogram_folder=Path("lyons")) assert check_calls[0] == 2 # noqa: PLR2004 + + +def test_spectro_data_with_multichannel_audio(monkeypatch: pytest.MonkeyPatch) -> None: + mocked_audio_value = np.array([[0.0, 1.0, 2.0] for _ in range(100)]) + af = MockedAudioFile(mocked_value=mocked_audio_value, sample_rate=100) + + ad: AudioData = AudioData.from_files(files=[af]) + + # By default, SpectroData targets channel 0 + sd = SpectroData.from_audio_data( + data=ad, fft=ShortTimeFFT(win=hamming(16), hop=16, fs=ad.sample_rate) + ) + + last_fetched_audio = [] + fft_stft = ShortTimeFFT.stft + + def mock_stft(*args, **kwargs) -> np.ndarray: + last_fetched_audio.clear() + last_fetched_audio[:] = kwargs["x"] + return fft_stft(self=sd.fft, *args, **kwargs) + + monkeypatch.setattr(sd.fft, "stft", mock_stft) + + sd.get_value() + assert np.array_equal(last_fetched_audio, [m[0] for m in mocked_audio_value]) + + # Set the audio_channel to the second channel + sd.audio_channel = 1 + sd.get_value() + assert np.array_equal(last_fetched_audio, [m[1] for m in mocked_audio_value]) + + # SpectroData.audio_channel is the index of the channel of the audio **file** + # (not the index of the channel in the AudioData.channels list) + sd.audio_data.channels = [1, 2] + sd.audio_channel = 1 + sd.get_value() + assert np.array_equal(last_fetched_audio, [m[1] for m in mocked_audio_value]) + + +def test_spectrodata_audio_channel_raises_if_not_in_audio_data_channels() -> None: + mocked_audio_value = np.array([[0.0, 1.0, 2.0] for _ in range(100)]) + af = MockedAudioFile(mocked_value=mocked_audio_value, sample_rate=100) + + ad: AudioData = AudioData.from_files(files=[af]) + + ad.channels = [0, 2] + + sd = SpectroData.from_audio_data( + data=ad, fft=ShortTimeFFT(win=hamming(48), hop=48, fs=ad.sample_rate) + ) + + with pytest.raises( + ValueError, match=r"channel 1: AudioData only targets channels \[0, 2\]" + ): + sd.audio_channel = 1