19831033427

Committed 01 Dec 2025 05:08PM UTC coverage: 96.981% (+0.03%) from 96.955%

Build # 19831033427

Build Type

Pull #307

github

Committed by

web-flow

Commit Message

Merge 92d3f5588 into 27d210a34

Pull Request Pull Request #307: [DRAFT] Relative paths

Run Details

41 of 42 new or added lines in 6 files covered. (97.62%)

13 existing lines in 5 files now uncovered.

3951 of 4074 relevant lines covered (96.98%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.15

/src/osekit/core_api/ltas_data.py

"""LTASData is a special form of SpectroData.

The Sx values from a LTASData object are computed recursively.
LTAS should be preferred in cases where the audio is really long.
In that case, the corresponding number of time bins (scipy.ShortTimeFTT.p_nums) is
too long for the whole Sx matrix to be computed once.

The LTAS are rather computed recursively. If the number of temporal bins is higher
than a target p_num value, the audio is split in p_num parts.
A separate sft is computed on each of these bits and averaged so that the end Sx
presents p_num temporal windows.

This averaging is performed recursively:
if the audio data is such that after a first split, the p_nums for each part
still is higher than p_num, the parts are further split
and each part is replaced with an average of the stft performed within it.

"""

from __future__ import annotations

from typing import TYPE_CHECKING

import numpy as np
from scipy.signal import ShortTimeFFT

from osekit.core_api.spectro_data import SpectroData
from osekit.utils.multiprocess_utils import multiprocess

if TYPE_CHECKING:
    from pandas import Timestamp

    from osekit.core_api.audio_data import AudioData
    from osekit.core_api.spectro_item import SpectroItem


class LTASData(SpectroData):
    """LTASData is a special form of SpectroData.

    The Sx values from a LTASData object are computed recursively.
    LTAS should be preferred in cases where the audio is really long.
    In that case, the corresponding number of time bins (scipy.ShortTimeFTT.p_nums) is
    too long for the whole Sx matrix to be computed once.

    The LTAS are rather computed recursively. If the number of temporal bins is higher
    than a target p_num value, the audio is split in p_num parts.
    A separate sft is computed on each of these bits and averaged so that the end Sx
    presents p_num temporal windows.

    This averaging is performed recursively:
    if the audio data is such that after a first split, the p_nums for each part
    still is higher than p_num, the parts are further split
    and each part is replaced with an average of the stft performed within it.

    """

    def __init__(
        self,
        items: list[SpectroItem] | None = None,
        audio_data: AudioData = None,
        begin: Timestamp | None = None,
        end: Timestamp | None = None,
        fft: ShortTimeFFT | None = None,
        db_ref: float | None = None,
        v_lim: tuple[float, float] | None = None,
        colormap: str | None = None,
        nb_time_bins: int = 1920,
    ) -> None:
        """Initialize a SpectroData from a list of SpectroItems.

        Parameters
        ----------
        items: list[SpectroItem]
            List of the SpectroItem constituting the SpectroData.
        audio_data: AudioData
            The audio data from which to compute the spectrogram.
        begin: Timestamp | None
            Only effective if items is None.
            Set the begin of the empty data.
        end: Timestamp | None
            Only effective if items is None.
            Set the end of the empty data.
        fft: ShortTimeFFT
            The short time FFT used for computing the spectrogram.
        db_ref: float | None
            Reference value for computing sx values in decibel.
        v_lim: tuple[float,float]
            Lower and upper limits (in dB) of the colormap used
            for plotting the spectrogram.
        colormap: str
            Colormap to use for plotting the spectrogram.
        nb_time_bins: int
            The maximum number of time bins of the LTAS.
            Given the audio data and the fft parameters,
            if the resulting spectrogram has a number of windows p_num
            <= nb_time_bins, the LTAS is computed like a classic spectrogram.
            Otherwise, the audio data is split in nb_time_bins equal-duration
            audio data, and each bin of the LTAS consist in an average of the
            fft values obtained on each of these bins. The audio is split recursively
            until p_num <= nb_time_bins.

        """
        ltas_fft = LTASData.get_ltas_fft(fft)
        super().__init__(
            items=items,
            audio_data=audio_data,
            begin=begin,
            end=end,
            fft=ltas_fft,
            db_ref=db_ref,
            v_lim=v_lim,
            colormap=colormap,
        )
        self.nb_time_bins = nb_time_bins
        self.sx_dtype = float

    @property
    def shape(self) -> tuple[int, int]:
        """Shape of the LTAS data."""
        return self.fft.f_pts, self.nb_time_bins

    def mean_value_part(self, sub_spectro: LTASData) -> np.ndarray:
        return np.mean(
            sub_spectro.get_value(depth=1),
            axis=1,
        )

    def get_value(self, depth: int = 0) -> np.ndarray:
        """Return the Sx matrix of the LTAS.

        The Sx matrix contains the absolute square of the STFT.
        """
        if not self.is_empty:
            return self._get_value_from_items(self.items)
        if super().shape[1] <= self.nb_time_bins:
            return super().get_value()
        sub_spectros = [
            LTASData.from_spectro_data(
                SpectroData.from_audio_data(ad, self.fft),
                nb_time_bins=self.nb_time_bins,
            )
            for ad in self.audio_data.split(self.nb_time_bins, pass_normalization=False)
        ]

        if depth != 0:
            return np.vstack(
                [self.mean_value_part(sub_spectro) for sub_spectro in sub_spectros],
            ).T

        return np.vstack(
            list(multiprocess(self.mean_value_part, sub_spectros)),
        ).T

    @classmethod
    def from_spectro_data(
        cls,
        spectro_data: SpectroData,
        nb_time_bins: int,
    ) -> LTASData:
        """Initialize a LTASData from a SpectroData.

        Parameters
        ----------
        spectro_data: SpectroData
            The spectrogram to turn in a LTAS.
        nb_time_bins: int
            The maximum number of windows over which the audio will be split to perform
            a LTAS.

        Returns
        -------
        LTASData:
            The LTASData instance.

        """
        items = spectro_data.items
        audio_data = spectro_data.audio_data
        begin = spectro_data.begin
        end = spectro_data.end
        fft = spectro_data.fft
        db_ref = spectro_data.db_ref
        v_lim = spectro_data.v_lim
        colormap = spectro_data.colormap
        return cls(
            items=items,
            audio_data=audio_data,
            begin=begin,
            end=end,
            fft=fft,
            nb_time_bins=nb_time_bins,
            db_ref=db_ref,
            v_lim=v_lim,
            colormap=colormap,
        )

    @classmethod
    def from_audio_data(
        cls,
        data: AudioData,
        fft: ShortTimeFFT,
        v_lim: tuple[float, float] | None = None,
        colormap: str | None = None,
        nb_time_bins: int = 1920,
    ) -> SpectroData:
        """Instantiate a SpectroData object from a AudioData object.

        Parameters
        ----------
        data: AudioData
            Audio data from which the SpectroData should be computed.
        fft: ShortTimeFFT
            The ShortTimeFFT used to compute the spectrogram.
        v_lim: tuple[float,float]
            Lower and upper limits (in dB) of the colormap used
            for plotting the spectrogram.
        colormap: str
            Colormap to use for plotting the spectrogram.
        nb_time_bins: int
            The maximum number of windows over which the audio will be split to perform
            Defaulted to 1920.

        Returns
        -------
        LTASData:
            The SpectroData object.

        """
        return cls(
            audio_data=data,
            fft=fft,
            begin=data.begin,
            end=data.end,
            v_lim=v_lim,
            colormap=colormap,
            nb_time_bins=nb_time_bins,
        )

    def to_dict(self, embed_sft: bool = True) -> dict:
        """Serialize a LTASData to a dictionary.

        Parameters
        ----------
        embed_sft: bool
            If True, the SFT parameters will be included in the dictionary.
            In a case where multiple SpectroData that share a same SFT are serialized,
            SFT parameters shouldn't be included in the dictionary, as the window
            values might lead to large redundant data.
            Rather, the SFT parameters should be serialized in
            a SpectroDataset dictionary so that it can be only stored once
            for all SpectroData instances.

        Returns
        -------
        dict:
            The serialized dictionary representing the LTASData.

        """
        return super().to_dict() | {"nb_time_bins": self.nb_time_bins}

    @classmethod
    def from_dict(cls, dictionary: dict, sft: ShortTimeFFT | None = None) -> LTASData:
        """Deserialize a LTASDataset from a dictionary.

        Parameters
        ----------
        dictionary: dict
            The serialized dictionary representing the AudioData.
        sft: ShortTimeFFT | None
            The ShortTimeFFT used to compute the spectrogram.
            If not provided, the SFT parameters must be included in the dictionary.

        Returns
        -------
        LTASDataset
            The deserialized LTASDataset.

        """
        return cls.from_spectro_data(
            SpectroData.from_dict(dictionary),
            nb_time_bins=dictionary["nb_time_bins"],
        )

    @staticmethod
    def get_ltas_fft(fft: ShortTimeFFT) -> ShortTimeFFT:
        """Return a ShortTimeFFT object optimized for computing LTAS.

        The overlap of the fft is forced set to 0, as the value of consecutive
        windows will in the end be averaged.

        Parameters
        ----------
        fft: ShortTimeFFT
            The fft to optimize for LTAS computation.

        Returns
        -------
        ShortTimeFFT
            The optimized fft.

        """
        win = fft.win
        fs = fft.fs
        mfft = fft.mfft
        hop = win.shape[0]
        return ShortTimeFFT(win=win, hop=hop, fs=fs, mfft=mfft)

1	"""LTASData is a special form of SpectroData.
2
3	The Sx values from a LTASData object are computed recursively.
4	LTAS should be preferred in cases where the audio is really long.
5	In that case, the corresponding number of time bins (scipy.ShortTimeFTT.p_nums) is
6	too long for the whole Sx matrix to be computed once.
7
8	The LTAS are rather computed recursively. If the number of temporal bins is higher
9	than a target p_num value, the audio is split in p_num parts.
10	A separate sft is computed on each of these bits and averaged so that the end Sx
11	presents p_num temporal windows.
12
13	This averaging is performed recursively:
14	if the audio data is such that after a first split, the p_nums for each part
15	still is higher than p_num, the parts are further split
16	and each part is replaced with an average of the stft performed within it.
17
18	"""
19
20	from __future__ import annotations	1✔
21
22	from typing import TYPE_CHECKING	1✔
23
24	import numpy as np	1✔
25	from scipy.signal import ShortTimeFFT	1✔
26
27	from osekit.core_api.spectro_data import SpectroData	1✔
28	from osekit.utils.multiprocess_utils import multiprocess	1✔
29
30	if TYPE_CHECKING:
31	from pandas import Timestamp
32
33	from osekit.core_api.audio_data import AudioData
34	from osekit.core_api.spectro_item import SpectroItem
35
36
37	class LTASData(SpectroData):	1✔
38	"""LTASData is a special form of SpectroData.
39
40	The Sx values from a LTASData object are computed recursively.
41	LTAS should be preferred in cases where the audio is really long.
42	In that case, the corresponding number of time bins (scipy.ShortTimeFTT.p_nums) is
43	too long for the whole Sx matrix to be computed once.
44
45	The LTAS are rather computed recursively. If the number of temporal bins is higher
46	than a target p_num value, the audio is split in p_num parts.
47	A separate sft is computed on each of these bits and averaged so that the end Sx
48	presents p_num temporal windows.
49
50	This averaging is performed recursively:
51	if the audio data is such that after a first split, the p_nums for each part
52	still is higher than p_num, the parts are further split
53	and each part is replaced with an average of the stft performed within it.
54
55	"""
56
57	def __init__(	1✔
58	self,
59	items: list[SpectroItem] \| None = None,
60	audio_data: AudioData = None,
61	begin: Timestamp \| None = None,
62	end: Timestamp \| None = None,
63	fft: ShortTimeFFT \| None = None,
64	db_ref: float \| None = None,
65	v_lim: tuple[float, float] \| None = None,
66	colormap: str \| None = None,
67	nb_time_bins: int = 1920,
68	) -> None:
69	"""Initialize a SpectroData from a list of SpectroItems.
70
71	Parameters
72	----------
73	items: list[SpectroItem]
74	List of the SpectroItem constituting the SpectroData.
75	audio_data: AudioData
76	The audio data from which to compute the spectrogram.
77	begin: Timestamp \| None
78	Only effective if items is None.
79	Set the begin of the empty data.
80	end: Timestamp \| None
81	Only effective if items is None.
82	Set the end of the empty data.
83	fft: ShortTimeFFT
84	The short time FFT used for computing the spectrogram.
85	db_ref: float \| None
86	Reference value for computing sx values in decibel.
87	v_lim: tuple[float,float]
88	Lower and upper limits (in dB) of the colormap used
89	for plotting the spectrogram.
90	colormap: str
91	Colormap to use for plotting the spectrogram.
92	nb_time_bins: int
93	The maximum number of time bins of the LTAS.
94	Given the audio data and the fft parameters,
95	if the resulting spectrogram has a number of windows p_num
96	<= nb_time_bins, the LTAS is computed like a classic spectrogram.
97	Otherwise, the audio data is split in nb_time_bins equal-duration
98	audio data, and each bin of the LTAS consist in an average of the
99	fft values obtained on each of these bins. The audio is split recursively
100	until p_num <= nb_time_bins.
101
102	"""
103	ltas_fft = LTASData.get_ltas_fft(fft)	1✔
104	super().__init__(	1✔
105	items=items,
106	audio_data=audio_data,
107	begin=begin,
108	end=end,
109	fft=ltas_fft,
110	db_ref=db_ref,
111	v_lim=v_lim,
112	colormap=colormap,
113	)
114	self.nb_time_bins = nb_time_bins	1✔
115	self.sx_dtype = float	1✔
116
117	@property	1✔
118	def shape(self) -> tuple[int, int]:	1✔
119	"""Shape of the LTAS data."""
120	return self.fft.f_pts, self.nb_time_bins	1✔
121
122	def mean_value_part(self, sub_spectro: LTASData) -> np.ndarray:	1✔
123	return np.mean(	1✔
124	sub_spectro.get_value(depth=1),
125	axis=1,
126	)
127
128	def get_value(self, depth: int = 0) -> np.ndarray:	1✔
129	"""Return the Sx matrix of the LTAS.
130
131	The Sx matrix contains the absolute square of the STFT.
132	"""
133	if not self.is_empty:	1✔
UNCOV 134	return self._get_value_from_items(self.items)	×
135	if super().shape[1] <= self.nb_time_bins:	1✔
136	return super().get_value()	1✔
137	sub_spectros = [	1✔
138	LTASData.from_spectro_data(
139	SpectroData.from_audio_data(ad, self.fft),
140	nb_time_bins=self.nb_time_bins,
141	)
142	for ad in self.audio_data.split(self.nb_time_bins, pass_normalization=False)
143	]
144
145	if depth != 0:	1✔
146	return np.vstack(	1✔
147	[self.mean_value_part(sub_spectro) for sub_spectro in sub_spectros],
148	).T
149
150	return np.vstack(	1✔
151	list(multiprocess(self.mean_value_part, sub_spectros)),
152	).T
153
154	@classmethod	1✔
155	def from_spectro_data(	1✔
156	cls,
157	spectro_data: SpectroData,
158	nb_time_bins: int,
159	) -> LTASData:
160	"""Initialize a LTASData from a SpectroData.
161
162	Parameters
163	----------
164	spectro_data: SpectroData
165	The spectrogram to turn in a LTAS.
166	nb_time_bins: int
167	The maximum number of windows over which the audio will be split to perform
168	a LTAS.
169
170	Returns
171	-------
172	LTASData:
173	The LTASData instance.
174
175	"""
176	items = spectro_data.items	1✔
177	audio_data = spectro_data.audio_data	1✔
178	begin = spectro_data.begin	1✔
179	end = spectro_data.end	1✔
180	fft = spectro_data.fft	1✔
181	db_ref = spectro_data.db_ref	1✔
182	v_lim = spectro_data.v_lim	1✔
183	colormap = spectro_data.colormap	1✔
184	return cls(	1✔
185	items=items,
186	audio_data=audio_data,
187	begin=begin,
188	end=end,
189	fft=fft,
190	nb_time_bins=nb_time_bins,
191	db_ref=db_ref,
192	v_lim=v_lim,
193	colormap=colormap,
194	)
195
196	@classmethod	1✔
197	def from_audio_data(	1✔
198	cls,
199	data: AudioData,
200	fft: ShortTimeFFT,
201	v_lim: tuple[float, float] \| None = None,
202	colormap: str \| None = None,
203	nb_time_bins: int = 1920,
204	) -> SpectroData:
205	"""Instantiate a SpectroData object from a AudioData object.
206
207	Parameters
208	----------
209	data: AudioData
210	Audio data from which the SpectroData should be computed.
211	fft: ShortTimeFFT
212	The ShortTimeFFT used to compute the spectrogram.
213	v_lim: tuple[float,float]
214	Lower and upper limits (in dB) of the colormap used
215	for plotting the spectrogram.
216	colormap: str
217	Colormap to use for plotting the spectrogram.
218	nb_time_bins: int
219	The maximum number of windows over which the audio will be split to perform
220	Defaulted to 1920.
221
222	Returns
223	-------
224	LTASData:
225	The SpectroData object.
226
227	"""
UNCOV 228	return cls(	×
229	audio_data=data,
230	fft=fft,
231	begin=data.begin,
232	end=data.end,
233	v_lim=v_lim,
234	colormap=colormap,
235	nb_time_bins=nb_time_bins,
236	)
237
238	def to_dict(self, embed_sft: bool = True) -> dict:	1✔
239	"""Serialize a LTASData to a dictionary.
240
241	Parameters
242	----------
243	embed_sft: bool
244	If True, the SFT parameters will be included in the dictionary.
245	In a case where multiple SpectroData that share a same SFT are serialized,
246	SFT parameters shouldn't be included in the dictionary, as the window
247	values might lead to large redundant data.
248	Rather, the SFT parameters should be serialized in
249	a SpectroDataset dictionary so that it can be only stored once
250	for all SpectroData instances.
251
252	Returns
253	-------
254	dict:
255	The serialized dictionary representing the LTASData.
256
257	"""
258	return super().to_dict() \| {"nb_time_bins": self.nb_time_bins}	1✔
259
260	@classmethod	1✔
261	def from_dict(cls, dictionary: dict, sft: ShortTimeFFT \| None = None) -> LTASData:	1✔
262	"""Deserialize a LTASDataset from a dictionary.
263
264	Parameters
265	----------
266	dictionary: dict
267	The serialized dictionary representing the AudioData.
268	sft: ShortTimeFFT \| None
269	The ShortTimeFFT used to compute the spectrogram.
270	If not provided, the SFT parameters must be included in the dictionary.
271
272	Returns
273	-------
274	LTASDataset
275	The deserialized LTASDataset.
276
277	"""
278	return cls.from_spectro_data(	1✔
279	SpectroData.from_dict(dictionary),
280	nb_time_bins=dictionary["nb_time_bins"],
281	)
282
283	@staticmethod	1✔
284	def get_ltas_fft(fft: ShortTimeFFT) -> ShortTimeFFT:	1✔
285	"""Return a ShortTimeFFT object optimized for computing LTAS.
286
287	The overlap of the fft is forced set to 0, as the value of consecutive
288	windows will in the end be averaged.
289
290	Parameters
291	----------
292	fft: ShortTimeFFT
293	The fft to optimize for LTAS computation.
294
295	Returns
296	-------
297	ShortTimeFFT
298	The optimized fft.
299
300	"""
301	win = fft.win	1✔
302	fs = fft.fs	1✔
303	mfft = fft.mfft	1✔
304	hop = win.shape[0]	1✔
305	return ShortTimeFFT(win=win, hop=hop, fs=fs, mfft=mfft)	1✔

Project-OSmOSE / OSEkit / 19831033427

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous