amine@403: import math amine@192: import os amine@411: from pathlib import Path amine@88: from random import random amine@192: from tempfile import TemporaryDirectory amine@426: from unittest import mock amine@426: from unittest.mock import patch amine@403: amine@405: import numpy as np amine@400: import pytest amine@403: amine@416: from auditok import ( amine@416: AudioParameterError, amine@416: AudioRegion, amine@416: load, amine@416: make_silence, amine@416: split, amine@416: split_and_join_with_silence, amine@416: ) amine@323: from auditok.core import ( amine@323: _duration_to_nb_windows, amine@323: _make_audio_region, amine@323: _read_chunks_online, amine@323: _read_offline, amine@323: ) amine@315: from auditok.io import get_audio_source amine@405: from auditok.signal import to_array amine@403: from auditok.util import AudioReader amine@86: amine@426: mock._magics.add("__round__") amine@426: amine@86: amine@299: def _make_random_length_regions( amine@299: byte_seq, sampling_rate, sample_width, channels amine@299: ): amine@88: regions = [] amine@88: for b in byte_seq: amine@88: duration = round(random() * 10, 6) amine@95: data = b * int(duration * sampling_rate) * sample_width * channels amine@244: region = AudioRegion(data, sampling_rate, sample_width, channels) amine@88: regions.append(region) amine@88: return regions amine@88: amine@88: amine@400: @pytest.mark.parametrize( amine@400: "skip, max_read, channels", amine@400: [ amine@405: (0, -1, 1), # no_skip_read_all amine@405: (0, -1, 2), # no_skip_read_all_stereo amine@405: (2, -1, 1), # skip_2_read_all amine@405: (2, None, 1), # skip_2_read_all_None amine@405: (2, 3, 1), # skip_2_read_3 amine@405: (2, 3.5, 2), # skip_2_read_3_5_stereo amine@405: (2.4, 3.5, 2), # skip_2_4_read_3_5_stereo amine@400: ], amine@400: ids=[ amine@400: "no_skip_read_all", amine@400: "no_skip_read_all_stereo", amine@400: "skip_2_read_all", amine@400: "skip_2_read_all_None", amine@400: "skip_2_read_3", amine@400: "skip_2_read_3_5_stereo", amine@400: "skip_2_4_read_3_5_stereo", amine@400: ], amine@400: ) amine@400: def test_load(skip, max_read, channels): amine@400: sampling_rate = 10 amine@400: sample_width = 2 amine@400: filename = "tests/data/test_split_10HZ_{}.raw" amine@400: filename = filename.format("mono" if channels == 1 else "stereo") amine@400: region = load( amine@400: filename, amine@400: skip=skip, amine@400: max_read=max_read, amine@400: sr=sampling_rate, amine@400: sw=sample_width, amine@400: ch=channels, amine@371: ) amine@400: with open(filename, "rb") as fp: amine@400: fp.read(round(skip * sampling_rate * sample_width * channels)) amine@400: if max_read is None or max_read < 0: amine@400: to_read = -1 amine@400: else: amine@400: to_read = round(max_read * sampling_rate * sample_width * channels) amine@400: expected = fp.read(to_read) amine@400: assert bytes(region) == expected amine@400: amine@400: amine@400: @pytest.mark.parametrize( amine@415: "duration, sampling_rate, sample_width, channels", amine@415: [ amine@415: (1.05, 16000, 1, 1), # mono_16K_1byte amine@415: (1.5, 16000, 2, 1), # mono_16K_2byte amine@415: (1.0001, 44100, 2, 2), # stereo_44100_2byte amine@415: (1.000005, 48000, 2, 3), # 3channel_48K_2byte amine@415: (1.0001, 48000, 4, 4), # 4channel_48K_4byte amine@415: (0, 48000, 4, 4), # 4channel_4K_4byte_0sec amine@415: ], amine@415: ids=[ amine@415: "mono_16K_1byte", amine@415: "mono_16K_2byte", amine@415: "stereo_44100_2byte", amine@415: "3channel_48000_2byte", amine@415: "4channel_48K_4byte", amine@415: "4channel_4K_4byte_0sec", amine@415: ], amine@415: ) amine@415: def test_make_silence(duration, sampling_rate, sample_width, channels): amine@415: silence = make_silence(duration, sampling_rate, sample_width, channels) amine@415: size = round(duration * sampling_rate) * sample_width * channels amine@415: expected_data = b"\0" * size amine@415: expected_duration = size / (sampling_rate * sample_width * channels) amine@415: assert silence.duration == expected_duration amine@415: assert silence.data == expected_data amine@415: amine@415: amine@415: @pytest.mark.parametrize( amine@416: "duration", amine@416: [ amine@416: (0,), # zero_second amine@416: (1,), # one_second amine@416: (1.0001,), # 1.0001_second amine@416: ], amine@416: ids=[ amine@416: "zero_second", amine@416: "one_second", amine@416: "1.0001_second", amine@416: ], amine@416: ) amine@416: def test_split_and_join_with_silence(duration): amine@416: duration = 1.0 amine@416: sampling_rate = 10 amine@416: sample_width = 2 amine@416: channels = 1 amine@416: amine@416: regions = split( amine@416: input="tests/data/test_split_10HZ_mono.raw", amine@416: min_dur=0.2, amine@416: max_dur=5, amine@416: max_silence=0.2, amine@416: drop_trailing_silence=False, amine@416: strict_min_dur=False, amine@416: analysis_window=0.1, amine@416: sr=sampling_rate, amine@416: sw=sample_width, amine@416: ch=channels, amine@416: eth=50, amine@416: ) amine@416: amine@416: size = round(duration * sampling_rate) * sample_width * channels amine@416: join_data = b"\0" * size amine@416: expected_data = join_data.join(region.data for region in regions) amine@416: expected_region = AudioRegion( amine@416: expected_data, sampling_rate, sample_width, channels amine@416: ) amine@416: amine@416: region_with_silence = split_and_join_with_silence( amine@416: input="tests/data/test_split_10HZ_mono.raw", amine@416: silence_duration=duration, amine@416: min_dur=0.2, amine@416: max_dur=5, amine@416: max_silence=0.2, amine@416: drop_trailing_silence=False, amine@416: strict_min_dur=False, amine@416: analysis_window=0.1, amine@416: sr=sampling_rate, amine@416: sw=sample_width, amine@416: ch=channels, amine@416: eth=50, amine@416: ) amine@416: assert region_with_silence == expected_region amine@416: amine@416: amine@416: @pytest.mark.parametrize( amine@400: "duration, analysis_window, round_fn, expected, kwargs", amine@400: [ amine@405: (0, 1, None, 0, None), # zero_duration amine@405: (0.3, 0.1, round, 3, None), # multiple amine@405: (0.35, 0.1, math.ceil, 4, None), # not_multiple_ceil amine@405: (0.35, 0.1, math.floor, 3, None), # not_multiple_floor amine@405: (0.05, 0.1, round, 0, None), # small_duration amine@405: (0.05, 0.1, math.ceil, 1, None), # small_duration_ceil amine@405: (0.3, 0.1, math.floor, 3, {"epsilon": 1e-6}), # with_round_error amine@405: (-0.5, 0.1, math.ceil, ValueError, None), # negative_duration amine@405: (0.5, -0.1, math.ceil, ValueError, None), # negative_analysis_window amine@400: ], amine@400: ids=[ amine@400: "zero_duration", amine@400: "multiple", amine@400: "not_multiple_ceil", amine@400: "not_multiple_floor", amine@400: "small_duration", amine@400: "small_duration_ceil", amine@400: "with_round_error", amine@400: "negative_duration", amine@400: "negative_analysis_window", amine@400: ], amine@400: ) amine@400: def test_duration_to_nb_windows( amine@400: duration, analysis_window, round_fn, expected, kwargs amine@400: ): amine@400: if expected == ValueError: amine@400: with pytest.raises(ValueError): amine@400: _duration_to_nb_windows(duration, analysis_window, round_fn) amine@400: else: amine@400: if kwargs is None: amine@400: kwargs = {} amine@400: result = _duration_to_nb_windows( amine@400: duration, analysis_window, round_fn, **kwargs amine@371: ) amine@400: assert result == expected amine@371: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "channels, skip, max_read", amine@400: [ amine@405: (1, 0, None), # mono_skip_0_max_read_None amine@405: (1, 3, None), # mono_skip_3_max_read_None amine@405: (1, 2, -1), # mono_skip_2_max_read_negative amine@405: (1, 2, 3), # mono_skip_2_max_read_3 amine@405: (2, 0, None), # stereo_skip_0_max_read_None amine@405: (2, 3, None), # stereo_skip_3_max_read_None amine@405: (2, 2, -1), # stereo_skip_2_max_read_negative amine@405: (2, 2, 3), # stereo_skip_2_max_read_3 amine@400: ], amine@400: ids=[ amine@400: "mono_skip_0_max_read_None", amine@400: "mono_skip_3_max_read_None", amine@400: "mono_skip_2_max_read_negative", amine@400: "mono_skip_2_max_read_3", amine@400: "stereo_skip_0_max_read_None", amine@400: "stereo_skip_3_max_read_None", amine@400: "stereo_skip_2_max_read_negative", amine@400: "stereo_skip_2_max_read_3", amine@400: ], amine@400: ) amine@400: def test_read_offline(channels, skip, max_read): amine@400: sampling_rate = 10 amine@400: sample_width = 2 amine@400: mono_or_stereo = "mono" if channels == 1 else "stereo" amine@400: filename = "tests/data/test_split_10HZ_{}.raw".format(mono_or_stereo) amine@400: with open(filename, "rb") as fp: amine@400: data = fp.read() amine@400: onset = round(skip * sampling_rate * sample_width * channels) amine@400: if max_read in (-1, None): amine@400: offset = len(data) + 1 amine@400: else: amine@400: offset = onset + round( amine@400: max_read * sampling_rate * sample_width * channels amine@400: ) amine@400: expected_data = data[onset:offset] amine@400: read_data, *audio_params = _read_offline( amine@400: filename, amine@400: skip=skip, amine@400: max_read=max_read, amine@400: sr=sampling_rate, amine@400: sw=sample_width, amine@400: ch=channels, amine@215: ) amine@400: assert read_data == expected_data amine@400: assert tuple(audio_params) == (sampling_rate, sample_width, channels) amine@215: amine@323: amine@400: @pytest.mark.parametrize( amine@405: ( amine@405: "min_dur, max_dur, max_silence, drop_trailing_silence, " amine@405: + "strict_min_dur, kwargs, expected" amine@405: ), amine@400: [ amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.2, amine@405: False, amine@405: False, amine@405: {"eth": 50}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # simple amine@400: ( amine@400: 0.3, amine@400: 2, amine@400: 0.2, amine@400: False, amine@400: False, amine@400: {"eth": 50}, amine@400: [(2, 16), (17, 31), (34, 54), (54, 74), (74, 76)], amine@405: ), # short_max_dur amine@405: (3, 5, 0.2, False, False, {"eth": 50}, [(34, 76)]), # long_min_dur amine@405: (0.2, 80, 10, False, False, {"eth": 50}, [(2, 76)]), # long_max_silence amine@400: ( amine@400: 0.2, amine@400: 5, amine@400: 0.0, amine@400: False, amine@400: False, amine@400: {"eth": 50}, amine@400: [(2, 14), (17, 24), (26, 29), (34, 76)], amine@405: ), # zero_max_silence amine@400: ( amine@299: 0.2, amine@299: 5, amine@299: 0.2, amine@299: False, amine@299: False, amine@207: {"energy_threshold": 40}, amine@207: [(0, 50), (50, 76)], amine@405: ), # low_energy_threshold amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.2, amine@405: False, amine@405: False, amine@405: {"energy_threshold": 60}, amine@405: [], amine@405: ), # high_energy_threshold amine@405: ( amine@405: 0.2, amine@405: 10, amine@405: 0.5, amine@405: True, amine@405: False, amine@405: {"eth": 50}, amine@405: [(2, 76)], amine@405: ), # trim_leading_and_trailing_silence amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.2, amine@405: True, amine@405: False, amine@405: {"eth": 50}, amine@405: [(2, 14), (17, 29), (34, 76)], amine@405: ), # drop_trailing_silence amine@405: ( amine@405: 1.5, amine@405: 5, amine@405: 0.2, amine@405: True, amine@405: False, amine@405: {"eth": 50}, amine@405: [(34, 76)], amine@405: ), # drop_trailing_silence_2 amine@400: ( amine@207: 0.3, amine@207: 2, amine@207: 0.2, amine@207: False, amine@207: True, amine@207: {"eth": 50}, amine@207: [(2, 16), (17, 31), (34, 54), (54, 74)], amine@405: ), # strict_min_dur amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "short_max_dur", amine@400: "long_min_dur", amine@400: "long_max_silence", amine@400: "zero_max_silence", amine@400: "low_energy_threshold", amine@400: "high_energy_threshold", amine@400: "trim_leading_and_trailing_silence", amine@400: "drop_trailing_silence", amine@400: "drop_trailing_silence_2", amine@400: "strict_min_dur", amine@400: ], amine@400: ) amine@400: def test_split_params( amine@400: min_dur, amine@400: max_dur, amine@400: max_silence, amine@400: drop_trailing_silence, amine@400: strict_min_dur, amine@400: kwargs, amine@400: expected, amine@400: ): amine@400: with open("tests/data/test_split_10HZ_mono.raw", "rb") as fp: amine@400: data = fp.read() amine@400: amine@400: regions = split( amine@400: data, amine@207: min_dur, amine@207: max_dur, amine@207: max_silence, amine@207: drop_trailing_silence, amine@207: strict_min_dur, amine@400: analysis_window=0.1, amine@400: sr=10, amine@400: sw=2, amine@400: ch=1, amine@400: **kwargs amine@400: ) amine@207: amine@400: region = AudioRegion(data, 10, 2, 1) amine@400: regions_ar = region.split( amine@400: min_dur, amine@400: max_dur, amine@400: max_silence, amine@400: drop_trailing_silence, amine@400: strict_min_dur, amine@400: analysis_window=0.1, amine@400: **kwargs amine@400: ) amine@255: amine@400: regions = list(regions) amine@400: regions_ar = list(regions_ar) amine@400: err_msg = "Wrong number of regions after split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions)) amine@400: assert len(regions) == len(expected), err_msg amine@400: err_msg = "Wrong number of regions after AudioRegion.split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions_ar)) amine@400: assert len(regions_ar) == len(expected), err_msg amine@255: amine@400: sample_width = 2 amine@426: for reg, reg_ar, exp in zip(regions, regions_ar, expected): amine@400: onset, offset = exp amine@400: exp_data = data[onset * sample_width : offset * sample_width] amine@400: assert bytes(reg) == exp_data amine@400: assert reg == reg_ar amine@207: amine@299: amine@400: @pytest.mark.parametrize( amine@400: "channels, kwargs, expected", amine@400: [ amine@405: (2, {}, [(2, 32), (34, 76)]), # stereo_all_default amine@405: (1, {"max_read": 5}, [(2, 16), (17, 31), (34, 50)]), # mono_max_read amine@405: ( amine@405: 1, amine@405: {"mr": 5}, amine@405: [(2, 16), (17, 31), (34, 50)], amine@405: ), # mono_max_read_short_name amine@405: ( amine@405: 1, amine@405: {"eth": 50, "use_channel": 0}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # mono_use_channel_1 amine@405: (1, {"eth": 50, "uc": 1}, [(2, 16), (17, 31), (34, 76)]), # mono_uc_1 amine@405: ( amine@405: 1, amine@405: {"eth": 50, "use_channel": None}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # mono_use_channel_None amine@405: ( amine@405: 2, amine@405: {"eth": 50, "use_channel": 0}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # stereo_use_channel_1 amine@405: ( amine@405: 2, amine@405: {"eth": 50}, amine@405: [(2, 32), (34, 76)], amine@405: ), # stereo_use_channel_no_use_channel_given amine@405: ( amine@405: 2, amine@405: {"eth": 50, "use_channel": -2}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # stereo_use_channel_minus_2 amine@405: (2, {"eth": 50, "uc": 1}, [(10, 32), (36, 76)]), # stereo_uc_2 amine@405: (2, {"eth": 50, "uc": -1}, [(10, 32), (36, 76)]), # stereo_uc_minus_1 amine@405: ( amine@405: 1, amine@405: {"eth": 50, "uc": "mix"}, amine@405: [(2, 16), (17, 31), (34, 76)], amine@405: ), # mono_uc_mix amine@405: ( amine@405: 2, amine@405: {"energy_threshold": 53.5, "use_channel": "mix"}, amine@405: [(54, 76)], amine@405: ), # stereo_use_channel_mix amine@405: (2, {"eth": 52, "uc": "mix"}, [(17, 26), (54, 76)]), # stereo_uc_mix amine@405: ( amine@405: 2, amine@405: {"uc": "mix"}, amine@405: [(10, 16), (17, 31), (36, 76)], amine@405: ), # stereo_uc_mix_default_eth amine@400: ], amine@400: ids=[ amine@400: "stereo_all_default", amine@400: "mono_max_read", amine@400: "mono_max_read_short_name", amine@400: "mono_use_channel_1", amine@400: "mono_uc_1", amine@400: "mono_use_channel_None", amine@400: "stereo_use_channel_1", amine@400: "stereo_use_channel_no_use_channel_given", amine@400: "stereo_use_channel_minus_2", amine@400: "stereo_uc_2", amine@400: "stereo_uc_minus_1", amine@400: "mono_uc_mix", amine@400: "stereo_use_channel_mix", amine@400: "stereo_uc_mix", amine@400: "stereo_uc_mix_default_eth", amine@400: ], amine@400: ) amine@400: def test_split_kwargs(channels, kwargs, expected): amine@400: amine@400: mono_or_stereo = "mono" if channels == 1 else "stereo" amine@400: filename = "tests/data/test_split_10HZ_{}.raw".format(mono_or_stereo) amine@400: with open(filename, "rb") as fp: amine@400: data = fp.read() amine@400: amine@400: regions = split( amine@400: data, amine@400: min_dur=0.2, amine@400: max_dur=5, amine@400: max_silence=0.2, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: analysis_window=0.1, amine@400: sr=10, amine@400: sw=2, amine@400: ch=channels, amine@400: **kwargs amine@211: ) amine@211: amine@400: region = AudioRegion(data, 10, 2, channels) amine@400: max_read = kwargs.get("max_read", kwargs.get("mr")) amine@400: if max_read is not None: amine@400: region = region.sec[:max_read] amine@400: kwargs.pop("max_read", None) amine@400: kwargs.pop("mr", None) amine@211: amine@400: regions_ar = region.split( amine@400: min_dur=0.2, amine@400: max_dur=5, amine@400: max_silence=0.2, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: analysis_window=0.1, amine@400: **kwargs amine@400: ) amine@255: amine@400: regions = list(regions) amine@400: regions_ar = list(regions_ar) amine@400: err_msg = "Wrong number of regions after split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions)) amine@400: assert len(regions) == len(expected), err_msg amine@400: err_msg = "Wrong number of regions after AudioRegion.split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions_ar)) amine@400: assert len(regions_ar) == len(expected), err_msg amine@306: amine@400: sample_width = 2 amine@400: sample_size_bytes = sample_width * channels amine@426: for reg, reg_ar, exp in zip( amine@426: regions, amine@426: regions_ar, amine@426: expected, amine@426: ): amine@400: onset, offset = exp amine@400: exp_data = data[onset * sample_size_bytes : offset * sample_size_bytes] amine@400: assert len(bytes(reg)) == len(exp_data) amine@400: assert reg == reg_ar amine@255: amine@255: amine@400: @pytest.mark.parametrize( amine@400: "min_dur, max_dur, max_silence, channels, kwargs, expected", amine@400: [ amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.2, amine@405: 1, amine@405: {"aw": 0.2}, amine@405: [(2, 30), (34, 76)], amine@405: ), # mono_aw_0_2_max_silence_0_2 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.3, amine@405: 1, amine@405: {"aw": 0.2}, amine@405: [(2, 30), (34, 76)], amine@405: ), # mono_aw_0_2_max_silence_0_3 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.4, amine@405: 1, amine@405: {"aw": 0.2}, amine@405: [(2, 32), (34, 76)], amine@405: ), # mono_aw_0_2_max_silence_0_4 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0, amine@405: 1, amine@405: {"aw": 0.2}, amine@405: [(2, 14), (16, 24), (26, 28), (34, 76)], amine@405: ), # mono_aw_0_2_max_silence_0 amine@405: (0.2, 5, 0.2, 1, {"aw": 0.2}, [(2, 30), (34, 76)]), # mono_aw_0_2 amine@405: ( amine@405: 0.3, amine@405: 5, amine@405: 0, amine@405: 1, amine@405: {"aw": 0.3}, amine@405: [(3, 12), (15, 24), (36, 76)], amine@405: ), # mono_aw_0_3_max_silence_0 amine@405: ( amine@405: 0.3, amine@405: 5, amine@405: 0.3, amine@405: 1, amine@405: {"aw": 0.3}, amine@405: [(3, 27), (36, 76)], amine@405: ), # mono_aw_0_3_max_silence_0_3 amine@405: ( amine@405: 0.3, amine@405: 5, amine@405: 0.5, amine@405: 1, amine@405: {"aw": 0.3}, amine@405: [(3, 27), (36, 76)], amine@405: ), # mono_aw_0_3_max_silence_0_5 amine@405: ( amine@405: 0.3, amine@405: 5, amine@405: 0.6, amine@405: 1, amine@405: {"aw": 0.3}, amine@405: [(3, 30), (36, 76)], amine@405: ), # mono_aw_0_3_max_silence_0_6 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0, amine@405: 1, amine@405: {"aw": 0.4}, amine@405: [(4, 12), (16, 24), (36, 76)], amine@405: ), # mono_aw_0_4_max_silence_0 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.3, amine@405: 1, amine@405: {"aw": 0.4}, amine@405: [(4, 12), (16, 24), (36, 76)], amine@405: ), # mono_aw_0_4_max_silence_0_3 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.4, amine@405: 1, amine@405: {"aw": 0.4}, amine@405: [(4, 28), (36, 76)], amine@405: ), # mono_aw_0_4_max_silence_0_4 amine@405: ( amine@405: 0.2, amine@405: 5, amine@405: 0.2, amine@405: 2, amine@405: {"analysis_window": 0.2}, amine@405: [(2, 32), (34, 76)], amine@405: ), # stereo_uc_None_analysis_window_0_2 amine@400: ( amine@316: 0.2, amine@316: 5, amine@316: 0.2, amine@316: 2, amine@316: {"uc": None, "analysis_window": 0.2}, amine@316: [(2, 32), (34, 76)], amine@405: ), # stereo_uc_any_analysis_window_0_2 amine@400: ( amine@316: 0.2, amine@316: 5, amine@316: 0.2, amine@316: 2, amine@316: {"use_channel": None, "analysis_window": 0.3}, amine@316: [(3, 30), (36, 76)], amine@405: ), # stereo_use_channel_None_aw_0_3_max_silence_0_2 amine@400: ( amine@316: 0.2, amine@316: 5, amine@316: 0.3, amine@316: 2, amine@316: {"use_channel": "any", "analysis_window": 0.3}, amine@316: [(3, 33), (36, 76)], amine@405: ), # stereo_use_channel_any_aw_0_3_max_silence_0_3 amine@400: ( amine@316: 0.2, amine@316: 5, amine@316: 0.2, amine@316: 2, amine@316: {"use_channel": None, "analysis_window": 0.4}, amine@316: [(4, 28), (36, 76)], amine@405: ), # stereo_use_channel_None_aw_0_4_max_silence_0_2 amine@400: ( amine@316: 0.2, amine@316: 5, amine@316: 0.4, amine@316: 2, amine@316: {"use_channel": "any", "analysis_window": 0.4}, amine@316: [(4, 32), (36, 76)], amine@405: ), # stereo_use_channel_any_aw_0_3_max_silence_0_4 amine@400: ( amine@241: 0.2, amine@241: 5, amine@241: 0.2, amine@241: 2, amine@241: {"uc": 0, "analysis_window": 0.2}, amine@241: [(2, 30), (34, 76)], amine@405: ), # stereo_uc_0_analysis_window_0_2 amine@400: ( amine@220: 0.2, amine@220: 5, amine@220: 0.2, amine@220: 2, amine@220: {"uc": 1, "analysis_window": 0.2}, amine@231: [(10, 32), (36, 76)], amine@405: ), # stereo_uc_1_analysis_window_0_2 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.1}, amine@233: [(10, 14), (17, 24), (26, 29), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_1_max_silence_0 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.1, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.1}, amine@233: [(10, 15), (17, 25), (26, 30), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_1_max_silence_0_1 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.2, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.1}, amine@233: [(10, 16), (17, 31), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_1_max_silence_0_2 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.3, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.1}, amine@233: [(10, 32), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_1_max_silence_0_3 amine@400: ( amine@233: 0.3, amine@233: 5, amine@233: 0, amine@233: 2, amine@316: {"uc": "avg", "analysis_window": 0.2}, amine@233: [(10, 14), (16, 24), (36, 76)], amine@405: ), # stereo_uc_avg_aw_0_2_max_silence_0_min_dur_0_3 amine@400: ( amine@233: 0.41, amine@233: 5, amine@233: 0, amine@233: 2, amine@316: {"uc": "average", "analysis_window": 0.2}, amine@233: [(16, 24), (36, 76)], amine@405: ), # stereo_uc_average_aw_0_2_max_silence_0_min_dur_0_41 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.1, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.2}, amine@233: [(10, 14), (16, 24), (26, 28), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_2_max_silence_0_1 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.2, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.2}, amine@233: [(10, 30), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_2_max_silence_0_2 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.4, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.2}, amine@233: [(10, 32), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_2_max_silence_0_4 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.5, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.2}, amine@233: [(10, 32), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_2_max_silence_0_5 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.2}, amine@233: [(10, 34), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_2_max_silence_0_6 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 24), (27, 30), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0 amine@400: ( amine@233: 0.4, amine@233: 5, amine@233: 0, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 24), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_min_dur_0_3 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 57), (57, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_6 amine@400: ( amine@233: 0.2, amine@233: 5.1, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 60), (60, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_1 amine@400: ( amine@233: 0.2, amine@233: 5.2, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 60), (60, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_2 amine@400: ( amine@233: 0.2, amine@233: 5.3, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 60), (60, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_3 amine@400: ( amine@233: 0.2, amine@233: 5.4, amine@233: 0.6, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.3}, amine@233: [(9, 63), (63, 76)], amine@405: ), # stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_4 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.4}, amine@233: [(16, 24), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_4_max_silence_0 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.3, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.4}, amine@233: [(16, 24), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_4_max_silence_0_3 amine@400: ( amine@233: 0.2, amine@233: 5, amine@233: 0.4, amine@233: 2, amine@233: {"uc": "mix", "analysis_window": 0.4}, amine@233: [(16, 28), (36, 76)], amine@405: ), # stereo_uc_mix_aw_0_4_max_silence_0_4 amine@400: ], amine@400: ids=[ amine@400: "mono_aw_0_2_max_silence_0_2", amine@400: "mono_aw_0_2_max_silence_0_3", amine@400: "mono_aw_0_2_max_silence_0_4", amine@400: "mono_aw_0_2_max_silence_0", amine@400: "mono_aw_0_2", amine@400: "mono_aw_0_3_max_silence_0", amine@400: "mono_aw_0_3_max_silence_0_3", amine@400: "mono_aw_0_3_max_silence_0_5", amine@400: "mono_aw_0_3_max_silence_0_6", amine@400: "mono_aw_0_4_max_silence_0", amine@400: "mono_aw_0_4_max_silence_0_3", amine@400: "mono_aw_0_4_max_silence_0_4", amine@400: "stereo_uc_None_analysis_window_0_2", amine@400: "stereo_uc_any_analysis_window_0_2", amine@400: "stereo_use_channel_None_aw_0_3_max_silence_0_2", amine@400: "stereo_use_channel_any_aw_0_3_max_silence_0_3", amine@400: "stereo_use_channel_None_aw_0_4_max_silence_0_2", amine@400: "stereo_use_channel_any_aw_0_3_max_silence_0_4", amine@400: "stereo_uc_0_analysis_window_0_2", amine@400: "stereo_uc_1_analysis_window_0_2", amine@400: "stereo_uc_mix_aw_0_1_max_silence_0", amine@400: "stereo_uc_mix_aw_0_1_max_silence_0_1", amine@400: "stereo_uc_mix_aw_0_1_max_silence_0_2", amine@400: "stereo_uc_mix_aw_0_1_max_silence_0_3", amine@400: "stereo_uc_avg_aw_0_2_max_silence_0_min_dur_0_3", amine@400: "stereo_uc_average_aw_0_2_max_silence_0_min_dur_0_41", amine@400: "stereo_uc_mix_aw_0_2_max_silence_0_1", amine@400: "stereo_uc_mix_aw_0_2_max_silence_0_2", amine@400: "stereo_uc_mix_aw_0_2_max_silence_0_4", amine@400: "stereo_uc_mix_aw_0_2_max_silence_0_5", amine@400: "stereo_uc_mix_aw_0_2_max_silence_0_6", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_min_dur_0_3", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_6", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_1", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_2", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_3", amine@400: "stereo_uc_mix_aw_0_3_max_silence_0_6_max_dur_5_4", amine@400: "stereo_uc_mix_aw_0_4_max_silence_0", amine@400: "stereo_uc_mix_aw_0_4_max_silence_0_3", amine@400: "stereo_uc_mix_aw_0_4_max_silence_0_4", amine@400: ], amine@400: ) amine@400: def test_split_analysis_window( amine@400: min_dur, max_dur, max_silence, channels, kwargs, expected amine@400: ): amine@400: amine@400: mono_or_stereo = "mono" if channels == 1 else "stereo" amine@400: filename = "tests/data/test_split_10HZ_{}.raw".format(mono_or_stereo) amine@400: with open(filename, "rb") as fp: amine@400: data = fp.read() amine@400: amine@400: regions = split( amine@400: data, amine@400: min_dur=min_dur, amine@400: max_dur=max_dur, amine@400: max_silence=max_silence, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: sr=10, amine@400: sw=2, amine@400: ch=channels, amine@400: eth=49.99, amine@400: **kwargs amine@220: ) amine@220: amine@400: region = AudioRegion(data, 10, 2, channels) amine@400: regions_ar = region.split( amine@400: min_dur=min_dur, amine@400: max_dur=max_dur, amine@400: max_silence=max_silence, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: eth=49.99, amine@400: **kwargs amine@400: ) amine@220: amine@400: regions = list(regions) amine@400: regions_ar = list(regions_ar) amine@400: err_msg = "Wrong number of regions after split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions)) amine@400: assert len(regions) == len(expected), err_msg amine@400: err_msg = "Wrong number of regions after AudioRegion.split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions_ar)) amine@400: assert len(regions_ar) == len(expected), err_msg amine@255: amine@400: sample_width = 2 amine@400: sample_size_bytes = sample_width * channels amine@426: for reg, reg_ar, exp in zip( amine@426: regions, amine@426: regions_ar, amine@426: expected, amine@426: ): amine@400: onset, offset = exp amine@400: exp_data = data[onset * sample_size_bytes : offset * sample_size_bytes] amine@400: assert bytes(reg) == exp_data amine@400: assert reg == reg_ar amine@255: amine@255: amine@400: def test_split_custom_validator(): amine@400: filename = "tests/data/test_split_10HZ_mono.raw" amine@400: with open(filename, "rb") as fp: amine@400: data = fp.read() amine@299: amine@400: regions = split( amine@400: data, amine@400: min_dur=0.2, amine@400: max_dur=5, amine@400: max_silence=0.2, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: sr=10, amine@400: sw=2, amine@400: ch=1, amine@400: analysis_window=0.1, amine@405: validator=lambda x: to_array(x, sample_width=2, channels=1)[0] >= 320, amine@400: ) amine@299: amine@400: region = AudioRegion(data, 10, 2, 1) amine@400: regions_ar = region.split( amine@400: min_dur=0.2, amine@400: max_dur=5, amine@400: max_silence=0.2, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: analysis_window=0.1, amine@405: validator=lambda x: to_array(x, sample_width=2, channels=1)[0] >= 320, amine@400: ) amine@299: amine@400: expected = [(2, 16), (17, 31), (34, 76)] amine@400: regions = list(regions) amine@400: regions_ar = list(regions_ar) amine@400: err_msg = "Wrong number of regions after split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions)) amine@400: assert len(regions) == len(expected), err_msg amine@400: err_msg = "Wrong number of regions after AudioRegion.split, expected: " amine@400: err_msg += "{}, found: {}".format(len(expected), len(regions_ar)) amine@400: assert len(regions_ar) == len(expected), err_msg amine@299: amine@400: sample_size_bytes = 2 amine@426: for reg, reg_ar, exp in zip( amine@426: regions, amine@426: regions_ar, amine@426: expected, amine@426: ): amine@400: onset, offset = exp amine@400: exp_data = data[onset * sample_size_bytes : offset * sample_size_bytes] amine@400: assert bytes(reg) == exp_data amine@400: assert reg == reg_ar amine@299: amine@220: amine@400: @pytest.mark.parametrize( amine@400: "input, kwargs", amine@400: [ amine@400: ( amine@212: "tests/data/test_split_10HZ_stereo.raw", amine@212: {"audio_format": "raw", "sr": 10, "sw": 2, "ch": 2}, amine@405: ), # filename_audio_format amine@400: ( amine@212: "tests/data/test_split_10HZ_stereo.raw", amine@212: {"fmt": "raw", "sr": 10, "sw": 2, "ch": 2}, amine@405: ), # filename_audio_format_short_name amine@405: ( amine@405: "tests/data/test_split_10HZ_stereo.raw", amine@405: {"sr": 10, "sw": 2, "ch": 2}, amine@405: ), # filename_no_audio_format amine@400: ( amine@212: "tests/data/test_split_10HZ_stereo.raw", amine@212: {"sampling_rate": 10, "sample_width": 2, "channels": 2}, amine@405: ), # filename_no_long_audio_params amine@400: ( amine@212: open("tests/data/test_split_10HZ_stereo.raw", "rb").read(), amine@212: {"sr": 10, "sw": 2, "ch": 2}, amine@405: ), # bytes_ amine@400: ( amine@403: AudioReader( amine@212: "tests/data/test_split_10HZ_stereo.raw", amine@212: sr=10, amine@212: sw=2, amine@212: ch=2, amine@212: block_dur=0.1, amine@212: ), amine@212: {}, amine@405: ), # audio_reader amine@400: ( amine@212: AudioRegion( amine@299: open("tests/data/test_split_10HZ_stereo.raw", "rb").read(), amine@299: 10, amine@299: 2, amine@299: 2, amine@212: ), amine@212: {}, amine@405: ), # audio_region amine@400: ( amine@212: get_audio_source( amine@212: "tests/data/test_split_10HZ_stereo.raw", sr=10, sw=2, ch=2 amine@212: ), amine@212: {}, amine@405: ), # audio_source amine@400: ], amine@400: ids=[ amine@400: "filename_audio_format", amine@400: "filename_audio_format_short_name", amine@400: "filename_no_audio_format", amine@400: "filename_no_long_audio_params", amine@400: "bytes_", amine@400: "audio_reader", amine@400: "audio_region", amine@400: "audio_source", amine@400: ], amine@400: ) amine@400: def test_split_input_type(input, kwargs): amine@400: amine@400: with open("tests/data/test_split_10HZ_stereo.raw", "rb") as fp: amine@400: data = fp.read() amine@400: amine@400: regions = split( amine@400: input, amine@400: min_dur=0.2, amine@400: max_dur=5, amine@400: max_silence=0.2, amine@400: drop_trailing_silence=False, amine@400: strict_min_dur=False, amine@400: analysis_window=0.1, amine@400: **kwargs amine@212: ) amine@400: regions = list(regions) amine@400: expected = [(2, 32), (34, 76)] amine@400: sample_width = 2 amine@400: err_msg = "Wrong number of regions after split, expected: " amine@400: err_msg += "{}, found: {}".format(expected, regions) amine@400: assert len(regions) == len(expected), err_msg amine@426: for reg, exp in zip( amine@426: regions, amine@426: expected, amine@426: ): amine@400: onset, offset = exp amine@400: exp_data = data[onset * sample_width * 2 : offset * sample_width * 2] amine@400: assert bytes(reg) == exp_data amine@212: amine@212: amine@400: @pytest.mark.parametrize( amine@400: "min_dur, max_dur, analysis_window", amine@400: [ amine@400: (0.5, 0.4, 0.1), amine@400: (0.44, 0.49, 0.1), amine@400: ], amine@400: ids=[ amine@400: "min_dur_greater_than_max_dur", amine@400: "durations_OK_but_wrong_number_of_analysis_windows", amine@400: ], amine@400: ) amine@400: def test_split_wrong_min_max_dur(min_dur, max_dur, analysis_window): amine@400: amine@400: with pytest.raises(ValueError) as val_err: amine@400: split( amine@400: b"0" * 16, amine@400: min_dur=min_dur, amine@400: max_dur=max_dur, amine@400: max_silence=0.2, amine@400: sr=16000, amine@400: sw=1, amine@400: ch=1, amine@400: analysis_window=analysis_window, amine@400: ) amine@400: amine@400: err_msg = "'min_dur' ({0} sec.) results in {1} analysis " amine@400: err_msg += "window(s) ({1} == ceil({0} / {2})) which is " amine@400: err_msg += "higher than the number of analysis window(s) for " amine@400: err_msg += "'max_dur' ({3} == floor({4} / {2}))" amine@400: amine@400: err_msg = err_msg.format( amine@400: min_dur, amine@400: math.ceil(min_dur / analysis_window), amine@400: analysis_window, amine@400: math.floor(max_dur / analysis_window), amine@400: max_dur, amine@400: ) amine@400: assert err_msg == str(val_err.value) amine@400: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "max_silence, max_dur, analysis_window", amine@400: [ amine@405: (0.5, 0.5, 0.1), # max_silence_equals_max_dur amine@405: (0.5, 0.4, 0.1), # max_silence_greater_than_max_dur amine@405: (0.44, 0.49, 0.1), # durations_OK_but_wrong_number_of_analysis_windows amine@400: ], amine@400: ids=[ amine@400: "max_silence_equals_max_dur", amine@400: "max_silence_greater_than_max_dur", amine@400: "durations_OK_but_wrong_number_of_analysis_windows", amine@400: ], amine@400: ) amine@400: def test_split_wrong_max_silence_max_dur(max_silence, max_dur, analysis_window): amine@400: amine@400: with pytest.raises(ValueError) as val_err: amine@400: split( amine@400: b"0" * 16, amine@400: min_dur=0.2, amine@400: max_dur=max_dur, amine@400: max_silence=max_silence, amine@400: sr=16000, amine@400: sw=1, amine@400: ch=1, amine@400: analysis_window=analysis_window, amine@400: ) amine@400: amine@400: err_msg = "'max_silence' ({0} sec.) results in {1} analysis " amine@400: err_msg += "window(s) ({1} == floor({0} / {2})) which is " amine@400: err_msg += "higher or equal to the number of analysis window(s) for " amine@400: err_msg += "'max_dur' ({3} == floor({4} / {2}))" amine@400: amine@400: err_msg = err_msg.format( amine@400: max_silence, amine@400: math.floor(max_silence / analysis_window), amine@400: analysis_window, amine@400: math.floor(max_dur / analysis_window), amine@400: max_dur, amine@400: ) amine@400: assert err_msg == str(val_err.value) amine@400: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "wrong_param", amine@400: [ amine@405: {"min_dur": -1}, # negative_min_dur amine@405: {"min_dur": 0}, # zero_min_dur amine@405: {"max_dur": -1}, # negative_max_dur amine@405: {"max_dur": 0}, # zero_max_dur amine@405: {"max_silence": -1}, # negative_max_silence amine@405: {"analysis_window": 0}, # zero_analysis_window amine@405: {"analysis_window": -1}, # negative_analysis_window amine@400: ], amine@400: ids=[ amine@400: "negative_min_dur", amine@400: "zero_min_dur", amine@400: "negative_max_dur", amine@400: "zero_max_dur", amine@400: "negative_max_silence", amine@400: "zero_analysis_window", amine@400: "negative_analysis_window", amine@400: ], amine@400: ) amine@400: def test_split_negative_temporal_params(wrong_param): amine@400: amine@400: params = { amine@400: "min_dur": 0.2, amine@400: "max_dur": 0.5, amine@400: "max_silence": 0.1, amine@400: "analysis_window": 0.1, amine@400: } amine@400: params.update(wrong_param) amine@400: with pytest.raises(ValueError) as val_err: amine@400: split(None, **params) amine@400: amine@400: name = set(wrong_param).pop() amine@400: value = wrong_param[name] amine@400: err_msg = "'{}' ({}) must be >{} 0".format( amine@400: name, value, "=" if name == "max_silence" else "" amine@400: ) amine@400: assert err_msg == str(val_err.value) amine@400: amine@400: amine@400: def test_split_too_small_analysis_window(): amine@400: with pytest.raises(ValueError) as val_err: amine@400: split(b"", sr=10, sw=1, ch=1, analysis_window=0.09) amine@403: err_msg = "Too small 'analysis_window' (0.09) for sampling rate (10)." amine@403: err_msg += " Analysis window should at least be 1/10 to cover one " amine@403: err_msg += "data sample" amine@400: assert err_msg == str(val_err.value) amine@400: amine@400: amine@400: def test_split_and_plot(): amine@400: amine@400: with open("tests/data/test_split_10HZ_mono.raw", "rb") as fp: amine@400: data = fp.read() amine@400: amine@400: region = AudioRegion(data, 10, 2, 1) amine@405: with patch("auditok.core.plot") as patch_fn: amine@400: regions = region.split_and_plot( amine@212: min_dur=0.2, amine@212: max_dur=5, amine@212: max_silence=0.2, amine@212: drop_trailing_silence=False, amine@212: strict_min_dur=False, amine@212: analysis_window=0.1, amine@400: sr=10, amine@400: sw=2, amine@400: ch=1, amine@400: eth=50, amine@212: ) amine@400: assert patch_fn.called amine@400: expected = [(2, 16), (17, 31), (34, 76)] amine@400: sample_width = 2 amine@400: expected_regions = [] amine@400: for onset, offset in expected: amine@400: onset *= sample_width amine@400: offset *= sample_width amine@400: expected_regions.append(AudioRegion(data[onset:offset], 10, 2, 1)) amine@400: assert regions == expected_regions amine@211: amine@223: amine@400: def test_split_exception(): amine@400: with open("tests/data/test_split_10HZ_mono.raw", "rb") as fp: amine@400: data = fp.read() amine@400: region = AudioRegion(data, 10, 2, 1) amine@223: amine@400: with pytest.raises(RuntimeWarning): amine@400: # max_read is not accepted when calling AudioRegion.split amine@400: region.split(max_read=2) amine@223: amine@223: amine@400: @pytest.mark.parametrize( amine@405: ( amine@405: "data, start, sampling_rate, sample_width, channels, expected_end, " amine@405: + "expected_duration_s, expected_duration_ms" amine@405: ), amine@400: [ amine@405: (b"\0" * 8000, 0, 8000, 1, 1, 1, 1, 1000), # simple amine@405: ( amine@405: b"\0" * 7992, amine@405: 0, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 0.999, amine@405: 0.999, amine@405: 999, amine@405: ), # one_ms_less_than_1_sec amine@405: ( amine@405: b"\0" * 7994, amine@405: 0, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 0.99925, amine@405: 0.99925, amine@405: 999, amine@405: ), # tree_quarter_ms_less_than_1_sec amine@405: ( amine@405: b"\0" * 7996, amine@405: 0, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 0.9995, amine@405: 0.9995, amine@405: 1000, amine@405: ), # half_ms_less_than_1_sec amine@405: ( amine@405: b"\0" * 7998, amine@405: 0, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 0.99975, amine@405: 0.99975, amine@405: 1000, amine@405: ), # quarter_ms_less_than_1_sec amine@405: (b"\0" * 8000 * 2, 0, 8000, 2, 1, 1, 1, 1000), # simple_sample_width_2 amine@405: (b"\0" * 8000 * 2, 0, 8000, 1, 2, 1, 1, 1000), # simple_stereo amine@405: (b"\0" * 8000 * 5, 0, 8000, 1, 5, 1, 1, 1000), # simple_multichannel amine@405: ( amine@405: b"\0" * 8000 * 2 * 5, amine@405: 0, amine@405: 8000, amine@405: 2, amine@405: 5, amine@405: 1, amine@405: 1, amine@405: 1000, amine@405: ), # simple_sample_width_2_multichannel amine@405: ( amine@405: b"\0" * 7992 * 2 * 5, amine@405: 0, amine@405: 8000, amine@405: 2, amine@405: 5, amine@405: 0.999, amine@405: 0.999, amine@405: 999, amine@405: ), # one_ms_less_than_1s_sw_2_multichannel amine@405: ( amine@405: b"\0" * 7994 * 2 * 5, amine@405: 0, amine@405: 8000, amine@405: 2, amine@405: 5, amine@405: 0.99925, amine@405: 0.99925, amine@405: 999, amine@405: ), # tree_qrt_ms_lt_1_s_sw_2_multichannel amine@405: ( amine@405: b"\0" * 7996 * 2 * 5, amine@405: 0, amine@405: 8000, amine@405: 2, amine@405: 5, amine@405: 0.9995, amine@405: 0.9995, amine@405: 1000, amine@405: ), # half_ms_lt_1s_sw_2_multichannel amine@405: ( amine@405: b"\0" * 7998 * 2 * 5, amine@405: 0, amine@405: 8000, amine@405: 2, amine@405: 5, amine@405: 0.99975, amine@405: 0.99975, amine@405: 1000, amine@405: ), # quarter_ms_lt_1s_sw_2_multichannel amine@405: ( amine@405: b"\0" * int(8000 * 1.33), amine@405: 2.7, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 4.03, amine@405: 1.33, amine@405: 1330, amine@405: ), # arbitrary_length_1 amine@405: ( amine@405: b"\0" * int(8000 * 0.476), amine@405: 11.568, amine@405: 8000, amine@405: 1, amine@405: 1, amine@405: 12.044, amine@405: 0.476, amine@405: 476, amine@405: ), # arbitrary_length_2 amine@400: ( amine@86: b"\0" * int(8000 * 1.711) * 2 * 3, amine@86: 9.415, amine@86: 8000, amine@86: 2, amine@86: 3, amine@86: 11.126, amine@86: 1.711, amine@86: 1711, amine@405: ), # arbitrary_length_sw_2_multichannel amine@400: ( amine@86: b"\0" * int(3172 * 1.318), amine@86: 17.236, amine@86: 3172, amine@86: 1, amine@86: 1, amine@86: 17.236 + int(3172 * 1.318) / 3172, amine@86: int(3172 * 1.318) / 3172, amine@86: 1318, amine@405: ), # arbitrary_sampling_rate amine@400: ( amine@86: b"\0" * int(11317 * 0.716) * 2 * 3, amine@86: 18.811, amine@86: 11317, amine@86: 2, amine@86: 3, amine@86: 18.811 + int(11317 * 0.716) / 11317, amine@86: int(11317 * 0.716) / 11317, amine@86: 716, amine@405: ), # arbitrary_sr_sw_2_multichannel amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "one_ms_less_than_1_sec", amine@400: "tree_quarter_ms_less_than_1_sec", amine@400: "half_ms_less_than_1_sec", amine@400: "quarter_ms_less_than_1_sec", amine@400: "simple_sample_width_2", amine@400: "simple_stereo", amine@400: "simple_multichannel", amine@400: "simple_sample_width_2_multichannel", amine@400: "one_ms_less_than_1s_sw_2_multichannel", amine@400: "tree_qrt_ms_lt_1_s_sw_2_multichannel", amine@400: "half_ms_lt_1s_sw_2_multichannel", amine@400: "quarter_ms_lt_1s_sw_2_multichannel", amine@400: "arbitrary_length_1", amine@400: "arbitrary_length_2", amine@400: "arbitrary_length_sw_2_multichannel", amine@405: "arbitrary_sampling_rate", amine@400: "arbitrary_sr_sw_2_multichannel", amine@400: ], amine@400: ) amine@400: def test_creation( amine@400: data, amine@400: start, amine@400: sampling_rate, amine@400: sample_width, amine@400: channels, amine@400: expected_end, amine@400: expected_duration_s, amine@400: expected_duration_ms, amine@400: ): amine@411: region = AudioRegion(data, sampling_rate, sample_width, channels, start) amine@400: assert region.sampling_rate == sampling_rate amine@400: assert region.sr == sampling_rate amine@400: assert region.sample_width == sample_width amine@400: assert region.sw == sample_width amine@400: assert region.channels == channels amine@400: assert region.ch == channels amine@400: assert region.meta.start == start amine@400: assert region.meta.end == expected_end amine@400: assert region.duration == expected_duration_s amine@400: assert len(region.ms) == expected_duration_ms amine@400: assert bytes(region) == data amine@400: amine@400: amine@400: def test_creation_invalid_data_exception(): amine@400: with pytest.raises(AudioParameterError) as audio_param_err: amine@400: _ = AudioRegion( amine@400: data=b"ABCDEFGHI", sampling_rate=8, sample_width=2, channels=1 amine@400: ) amine@400: assert str(audio_param_err.value) == ( amine@400: "The length of audio data must be an integer " amine@400: "multiple of `sample_width * channels`" amine@86: ) amine@88: amine@97: amine@400: @pytest.mark.parametrize( amine@400: "skip, max_read, channels", amine@400: [ amine@405: (0, -1, 1), # no_skip_read_all amine@405: (0, -1, 2), # no_skip_read_all_stereo amine@405: (2, -1, 1), # skip_2_read_all amine@405: (2, None, 1), # skip_2_read_all_None amine@405: (2, 3, 1), # skip_2_read_3 amine@405: (2, 3.5, 2), # skip_2_read_3_5_stereo amine@405: (2.4, 3.5, 2), # skip_2_4_read_3_5_stereo amine@400: ], amine@400: ids=[ amine@400: "no_skip_read_all", amine@400: "no_skip_read_all_stereo", amine@400: "skip_2_read_all", amine@400: "skip_2_read_all_None", amine@400: "skip_2_read_3", amine@400: "skip_2_read_3_5_stereo", amine@400: "skip_2_4_read_3_5_stereo", amine@400: ], amine@400: ) amine@400: def test_load_AudioRegion(skip, max_read, channels): amine@400: sampling_rate = 10 amine@400: sample_width = 2 amine@400: filename = "tests/data/test_split_10HZ_{}.raw" amine@400: filename = filename.format("mono" if channels == 1 else "stereo") amine@400: region = AudioRegion.load( amine@400: filename, amine@400: skip=skip, amine@400: max_read=max_read, amine@400: sr=sampling_rate, amine@400: sw=sample_width, amine@400: ch=channels, amine@308: ) amine@400: with open(filename, "rb") as fp: amine@400: fp.read(round(skip * sampling_rate * sample_width * channels)) amine@400: if max_read is None or max_read < 0: amine@400: to_read = -1 amine@400: else: amine@400: to_read = round(max_read * sampling_rate * sample_width * channels) amine@400: expected = fp.read(to_read) amine@400: assert bytes(region) == expected amine@308: amine@308: amine@400: def test_load_from_microphone(): amine@400: with patch("auditok.io.PyAudioSource") as patch_pyaudio_source: amine@400: with patch("auditok.core.AudioReader.read") as patch_reader: amine@400: patch_reader.return_value = None amine@400: with patch( amine@400: "auditok.core.AudioRegion.__init__" amine@400: ) as patch_AudioRegion: amine@400: patch_AudioRegion.return_value = None amine@400: AudioRegion.load(None, skip=0, max_read=5, sr=16000, sw=2, ch=1) amine@400: assert patch_pyaudio_source.called amine@400: assert patch_reader.called amine@400: assert patch_AudioRegion.called amine@307: amine@308: amine@400: @pytest.mark.parametrize( amine@400: "max_read", amine@400: [ amine@405: None, # None amine@405: -1, # negative amine@400: ], amine@400: ids=[ amine@405: "None", amine@400: "negative", amine@400: ], amine@400: ) amine@400: def test_load_from_microphone_without_max_read_exception(max_read): amine@400: with pytest.raises(ValueError) as val_err: amine@400: AudioRegion.load(None, max_read=max_read, sr=16000, sw=2, ch=1) amine@400: assert str(val_err.value) == ( amine@400: "'max_read' should not be None when reading from microphone" amine@400: ) amine@400: amine@400: amine@400: def test_load_from_microphone_with_nonzero_skip_exception(): amine@400: with pytest.raises(ValueError) as val_err: amine@400: AudioRegion.load(None, skip=1, max_read=5, sr=16000, sw=2, ch=1) amine@400: assert str(val_err.value) == ( amine@400: "'skip' should be 0 when reading from microphone" amine@400: ) amine@400: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "format, start, expected", amine@400: [ amine@405: ("output.wav", 1.230, "output.wav"), # simple amine@405: ("output_{meta.start:g}.wav", 1.230, "output_1.23.wav"), # start amine@405: ("output_{meta.start}.wav", 1.233712, "output_1.233712.wav"), # start_2 amine@405: ( amine@405: "output_{meta.start:.2f}.wav", amine@405: 1.2300001, amine@405: "output_1.23.wav", amine@405: ), # start_3 amine@405: ( amine@405: "output_{meta.start:.3f}.wav", amine@405: 1.233712, amine@405: "output_1.234.wav", amine@405: ), # start_4 amine@405: ( amine@405: "output_{meta.start:.8f}.wav", amine@405: 1.233712, amine@405: "output_1.23371200.wav", amine@405: ), # start_5 amine@400: ( amine@244: "output_{meta.start}_{meta.end}_{duration}.wav", amine@192: 1.455, amine@192: "output_1.455_2.455_1.0.wav", amine@405: ), # start_end_duration amine@400: ( amine@244: "output_{meta.start}_{meta.end}_{duration}.wav", amine@192: 1.455321, amine@192: "output_1.455321_2.455321_1.0.wav", amine@405: ), # start_end_duration_2 amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "start", amine@400: "start_2", amine@400: "start_3", amine@400: "start_4", amine@400: "start_5", amine@400: "start_end_duration", amine@400: "start_end_duration_2", amine@400: ], amine@400: ) amine@400: def test_save(format, start, expected): amine@400: with TemporaryDirectory() as tmpdir: amine@411: region = AudioRegion(b"0" * 160, 160, 1, 1, start) amine@400: format = os.path.join(tmpdir, format) amine@400: filename = region.save(format)[len(tmpdir) + 1 :] amine@400: assert filename == expected amine@192: amine@193: amine@400: def test_save_file_exists_exception(): amine@400: with TemporaryDirectory() as tmpdir: amine@400: filename = os.path.join(tmpdir, "output.wav") amine@400: open(filename, "w").close() amine@400: region = AudioRegion(b"0" * 160, 160, 1, 1) amine@400: with pytest.raises(FileExistsError): amine@400: region.save(filename, exists_ok=False) amine@400: amine@411: with pytest.raises(FileExistsError): amine@411: region.save(Path(filename), exists_ok=False) amine@411: amine@400: amine@400: @pytest.mark.parametrize( amine@414: "sampling_rate, sample_width, channels", amine@414: [ amine@414: (16000, 1, 1), # mono_16K_1byte amine@414: (16000, 2, 1), # mono_16K_2byte amine@414: (44100, 2, 2), # stereo_44100_2byte amine@414: (44100, 2, 3), # 3channel_44100_2byte amine@414: ], amine@414: ids=[ amine@414: "mono_16K_1byte", amine@414: "mono_16K_2byte", amine@414: "stereo_44100_2byte", amine@414: "3channel_44100_2byte", amine@414: ], amine@414: ) amine@414: def test_join(sampling_rate, sample_width, channels): amine@414: duration = 1 amine@414: size = int(duration * sampling_rate * sample_width * channels) amine@414: glue_data = b"\0" * size amine@414: regions_data = [ amine@414: b"\1" * int(size * 1.5), amine@414: b"\2" * int(size * 0.5), amine@414: b"\3" * int(size * 0.75), amine@414: ] amine@414: amine@414: glue_region = AudioRegion(glue_data, sampling_rate, sample_width, channels) amine@414: regions = [ amine@414: AudioRegion(data, sampling_rate, sample_width, channels) amine@414: for data in regions_data amine@414: ] amine@414: joined = glue_region.join(regions) amine@414: assert joined.data == glue_data.join(regions_data) amine@414: assert joined.duration == duration * 2 + 1.5 + 0.5 + 0.75 amine@414: amine@414: amine@414: @pytest.mark.parametrize( amine@414: "sampling_rate, sample_width, channels", amine@414: [ amine@414: (32000, 1, 1), # different_sampling_rate amine@414: (16000, 2, 1), # different_sample_width amine@414: (16000, 1, 2), # different_channels amine@414: ], amine@414: ids=[ amine@414: "different_sampling_rate", amine@414: "different_sample_width", amine@414: "different_channels", amine@414: ], amine@414: ) amine@414: def test_join_exception(sampling_rate, sample_width, channels): amine@414: amine@414: glue_sampling_rate = 16000 amine@414: glue_sample_width = 1 amine@414: glue_channels = 1 amine@414: amine@414: duration = 1 amine@414: size = int( amine@414: duration * glue_sampling_rate * glue_sample_width * glue_channels amine@414: ) amine@414: glue_data = b"\0" * size amine@414: glue_region = AudioRegion( amine@414: glue_data, glue_sampling_rate, glue_sample_width, glue_channels amine@414: ) amine@414: amine@414: size = int(duration * sampling_rate * sample_width * channels) amine@414: regions_data = [ amine@414: b"\1" * int(size * 1.5), amine@414: b"\2" * int(size * 0.5), amine@414: b"\3" * int(size * 0.75), amine@414: ] amine@414: regions = [ amine@414: AudioRegion(data, sampling_rate, sample_width, channels) amine@414: for data in regions_data amine@414: ] amine@414: amine@414: with pytest.raises(AudioParameterError): amine@414: glue_region.join(regions) amine@414: amine@414: amine@414: @pytest.mark.parametrize( amine@400: "region, slice_, expected_data", amine@400: [ amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@194: slice(0, 500), amine@405: b"a" * 80, # first_half amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(500, None), amine@405: b"b" * 80, # second_half amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(-500, None), amine@405: b"b" * 80, # second_half_negative amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(200, 750), amine@405: b"a" * 48 + b"b" * 40, # middle amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(-800, -250), amine@405: b"a" * 48 + b"b" * 40, # middle_negative amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 160 + b"b" * 160, 160, 2, 1), amine@244: slice(200, 750), amine@405: b"a" * 96 + b"b" * 80, # middle_sw2 amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 160 + b"b" * 160, 160, 1, 2), amine@244: slice(200, 750), amine@405: b"a" * 96 + b"b" * 80, # middle_ch2 amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 320 + b"b" * 320, 160, 2, 2), amine@244: slice(200, 750), amine@405: b"a" * 192 + b"b" * 160, # middle_sw2_ch2 amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(1, None), amine@405: b"a" * (4000 - 8) + b"b" * 4000, # but_first_sample amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(-999, None), amine@405: b"a" * (4000 - 8) + b"b" * 4000, # but_first_sample_negative amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(0, 999), amine@405: b"a" * 4000 + b"b" * (4000 - 8), # but_last_sample amine@244: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(0, -1), amine@405: b"a" * 4000 + b"b" * (4000 - 8), # but_last_sample_negative amine@244: ), amine@405: ( amine@405: AudioRegion(b"a" * 160, 160, 1, 1), amine@405: slice(-5000, None), amine@405: b"a" * 160, # big_negative_start amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 160, 160, 1, 1), amine@405: slice(None, -1500), amine@405: b"", # big_negative_stop amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(0, 0), amine@405: b"", # empty amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(200, 100), amine@405: b"", # empty_start_stop_reversed amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(2000, 3000), amine@405: b"", # empty_big_positive_start amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(-100, -200), amine@405: b"", # empty_negative_reversed amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(0, -2000), amine@405: b"", # empty_big_negative_stop amine@405: ), amine@400: ( amine@244: AudioRegion(b"a" * 124 + b"b" * 376, 1234, 1, 1), amine@244: slice(100, 200), amine@405: b"a" + b"b" * 123, # arbitrary_sampling_rate amine@244: ), amine@400: ], amine@400: ids=[ amine@400: "first_half", amine@400: "second_half", amine@400: "second_half_negative", amine@400: "middle", amine@400: "middle_negative", amine@400: "middle_sw2", amine@400: "middle_ch2", amine@400: "middle_sw2_ch2", amine@400: "but_first_sample", amine@400: "but_first_sample_negative", amine@400: "but_last_sample", amine@400: "but_last_sample_negative", amine@400: "big_negative_start", amine@400: "big_negative_stop", amine@400: "empty", amine@400: "empty_start_stop_reversed", amine@400: "empty_big_positive_start", amine@400: "empty_negative_reversed", amine@400: "empty_big_negative_stop", amine@400: "arbitrary_sampling_rate", amine@400: ], amine@400: ) amine@400: def test_region_temporal_slicing(region, slice_, expected_data): amine@400: sub_region = region.millis[slice_] amine@400: assert bytes(sub_region) == expected_data amine@400: start_sec = slice_.start / 1000 if slice_.start is not None else None amine@400: stop_sec = slice_.stop / 1000 if slice_.stop is not None else None amine@400: sub_region = region.sec[start_sec:stop_sec] amine@400: assert bytes(sub_region) == expected_data amine@244: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "region, slice_, time_shift, expected_data", amine@400: [ amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(0, 80), amine@194: 0, amine@405: b"a" * 80, # first_half amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(80, None), amine@194: 0.5, amine@405: b"b" * 80, # second_half amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(-80, None), amine@194: 0.5, amine@405: b"b" * 80, # second_half_negative amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(160 // 5, 160 // 4 * 3), amine@194: 0.2, amine@405: b"a" * 48 + b"b" * 40, # middle amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(-160 // 5 * 4, -160 // 4), amine@194: 0.2, amine@405: b"a" * 48 + b"b" * 40, # middle_negative amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 160 + b"b" * 160, 160, 2, 1), amine@244: slice(160 // 5, 160 // 4 * 3), amine@194: 0.2, amine@405: b"a" * 96 + b"b" * 80, # middle_sw2 amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 160 + b"b" * 160, 160, 1, 2), amine@244: slice(160 // 5, 160 // 4 * 3), amine@194: 0.2, amine@405: b"a" * 96 + b"b" * 80, # middle_ch2 amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 320 + b"b" * 320, 160, 2, 2), amine@244: slice(160 // 5, 160 // 4 * 3), amine@194: 0.2, amine@405: b"a" * 192 + b"b" * 160, # middle_sw2_ch2 amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@194: slice(1, None), amine@244: 1 / 8000, amine@405: b"a" * (4000 - 1) + b"b" * 4000, # but_first_sample amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(-7999, None), amine@244: 1 / 8000, amine@405: b"a" * (4000 - 1) + b"b" * 4000, # but_first_sample_negative amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@244: slice(0, 7999), amine@194: 0, amine@405: b"a" * 4000 + b"b" * (4000 - 1), # but_last_sample amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 4000 + b"b" * 4000, 8000, 1, 1), amine@194: slice(0, -1), amine@194: 0, amine@405: b"a" * 4000 + b"b" * (4000 - 1), # but_last_sample_negative amine@194: ), amine@405: ( amine@405: AudioRegion(b"a" * 160, 160, 1, 1), amine@405: slice(-1600, None), amine@405: 0, amine@405: b"a" * 160, # big_negative_start amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 160, 160, 1, 1), amine@405: slice(None, -1600), amine@405: 0, amine@405: b"", # big_negative_stop amine@405: ), amine@405: ( amine@405: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@405: slice(0, 0), amine@405: 0, amine@405: b"", # empty amine@405: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(80, 40), amine@244: 0.5, amine@405: b"", # empty_start_stop_reversed amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(1600, 3000), amine@244: 10, amine@405: b"", # empty_big_positive_start amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@244: slice(-16, -32), amine@194: 0.9, amine@405: b"", # empty_negative_reversed amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 80 + b"b" * 80, 160, 1, 1), amine@194: slice(0, -2000), amine@194: 0, amine@405: b"", # empty_big_negative_stop amine@194: ), amine@400: ( amine@244: AudioRegion(b"a" * 124 + b"b" * 376, 1235, 1, 1), amine@231: slice(100, 200), amine@231: 100 / 1235, amine@405: b"a" * 24 + b"b" * 76, # arbitrary_sampling_rate amine@231: ), amine@400: ( amine@244: AudioRegion(b"a" * 124 + b"b" * 376, 1235, 2, 2), amine@231: slice(25, 50), amine@231: 25 / 1235, amine@405: b"a" * 24 + b"b" * 76, # arbitrary_sampling_rate_middle_sw2_ch2 amine@231: ), amine@400: ], amine@400: ids=[ amine@400: "first_half", amine@400: "second_half", amine@400: "second_half_negative", amine@400: "middle", amine@400: "middle_negative", amine@400: "middle_sw2", amine@400: "middle_ch2", amine@400: "middle_sw2_ch2", amine@400: "but_first_sample", amine@400: "but_first_sample_negative", amine@400: "but_last_sample", amine@400: "but_last_sample_negative", amine@400: "big_negative_start", amine@400: "big_negative_stop", amine@400: "empty", amine@400: "empty_start_stop_reversed", amine@400: "empty_big_positive_start", amine@400: "empty_negative_reversed", amine@400: "empty_big_negative_stop", amine@400: "arbitrary_sampling_rate", amine@400: "arbitrary_sampling_rate_middle_sw2_ch2", amine@400: ], amine@400: ) amine@400: def test_region_sample_slicing(region, slice_, time_shift, expected_data): amine@400: sub_region = region[slice_] amine@400: assert bytes(sub_region) == expected_data amine@400: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "sampling_rate, sample_width, channels", amine@400: [ amine@405: (8000, 1, 1), # simple amine@405: (8000, 2, 2), # stereo_sw_2 amine@405: (5413, 2, 3), # arbitrary_sr_multichannel amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "stereo_sw_2", amine@400: "arbitrary_sr_multichannel", amine@400: ], amine@400: ) amine@400: def test_concatenation(sampling_rate, sample_width, channels): amine@400: amine@400: region_1, region_2 = _make_random_length_regions( amine@400: [b"a", b"b"], sampling_rate, sample_width, channels amine@231: ) amine@400: expected_duration = region_1.duration + region_2.duration amine@400: expected_data = bytes(region_1) + bytes(region_2) amine@400: concat_region = region_1 + region_2 amine@400: assert concat_region.duration == pytest.approx(expected_duration, abs=1e-6) amine@400: assert bytes(concat_region) == expected_data amine@231: amine@400: amine@400: @pytest.mark.parametrize( amine@400: "sampling_rate, sample_width, channels", amine@400: [ amine@405: (8000, 1, 1), # simple amine@405: (8000, 2, 2), # stereo_sw_2 amine@405: (5413, 2, 3), # arbitrary_sr_multichannel amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "stereo_sw_2", amine@400: "arbitrary_sr_multichannel", amine@400: ], amine@400: ) amine@400: def test_concatenation_many(sampling_rate, sample_width, channels): amine@400: amine@400: regions = _make_random_length_regions( amine@400: [b"a", b"b", b"c"], sampling_rate, sample_width, channels amine@88: ) amine@400: expected_duration = sum(r.duration for r in regions) amine@400: expected_data = b"".join(bytes(r) for r in regions) amine@400: concat_region = sum(regions) amine@88: amine@400: assert concat_region.duration == pytest.approx(expected_duration, abs=1e-6) amine@400: assert bytes(concat_region) == expected_data amine@88: amine@400: amine@400: def test_concatenation_different_sampling_rate_error(): amine@400: region_1 = AudioRegion(b"a" * 100, 8000, 1, 1) amine@400: region_2 = AudioRegion(b"b" * 100, 3000, 1, 1) amine@400: amine@414: with pytest.raises(AudioParameterError) as val_err: amine@400: region_1 + region_2 amine@400: assert str(val_err.value) == ( amine@400: "Can only concatenate AudioRegions of the same " amine@405: "sampling rate (8000 != 3000)" # different_sampling_rate amine@88: ) amine@88: amine@88: amine@400: def test_concatenation_different_sample_width_error(): amine@400: region_1 = AudioRegion(b"a" * 100, 8000, 2, 1) amine@400: region_2 = AudioRegion(b"b" * 100, 8000, 4, 1) amine@88: amine@414: with pytest.raises(AudioParameterError) as val_err: amine@400: region_1 + region_2 amine@400: assert str(val_err.value) == ( amine@405: "Can only concatenate AudioRegions of the same sample width (2 != 4)" amine@400: ) amine@88: amine@88: amine@400: def test_concatenation_different_number_of_channels_error(): amine@400: region_1 = AudioRegion(b"a" * 100, 8000, 1, 1) amine@400: region_2 = AudioRegion(b"b" * 100, 8000, 1, 2) amine@88: amine@414: with pytest.raises(AudioParameterError) as val_err: amine@400: region_1 + region_2 amine@400: assert str(val_err.value) == ( amine@400: "Can only concatenate AudioRegions of the same " amine@405: "number of channels (1 != 2)" # different_number_of_channels amine@400: ) amine@88: amine@88: amine@400: @pytest.mark.parametrize( amine@400: "duration, expected_duration, expected_len, expected_len_ms", amine@400: [ amine@405: (0.01, 0.03, 240, 30), # simple amine@405: (0.00575, 0.01725, 138, 17), # rounded_len_floor amine@405: (0.00625, 0.01875, 150, 19), # rounded_len_ceil amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "rounded_len_floor", amine@400: "rounded_len_ceil", amine@400: ], amine@400: ) amine@400: def test_multiplication( amine@400: duration, expected_duration, expected_len, expected_len_ms amine@400: ): amine@400: sw = 2 amine@400: data = b"0" * int(duration * 8000 * sw) amine@400: region = AudioRegion(data, 8000, sw, 1) amine@400: m_region = 1 * region * 3 amine@400: assert bytes(m_region) == data * 3 amine@400: assert m_region.sr == 8000 amine@400: assert m_region.sw == 2 amine@400: assert m_region.ch == 1 amine@400: assert m_region.duration == expected_duration amine@400: assert len(m_region) == expected_len amine@400: assert m_region.len == expected_len amine@400: assert m_region.s.len == expected_duration amine@400: assert len(m_region.ms) == expected_len_ms amine@400: assert m_region.ms.len == expected_len_ms amine@88: amine@196: amine@400: @pytest.mark.parametrize( amine@400: "factor, _type", amine@400: [ amine@405: ("x", str), # string amine@405: (1.4, float), # float amine@400: ], amine@400: ids=[ amine@405: "string", amine@405: "float", amine@400: ], amine@400: ) amine@400: def test_multiplication_non_int(factor, _type): amine@400: with pytest.raises(TypeError) as type_err: amine@400: AudioRegion(b"0" * 80, 8000, 1, 1) * factor amine@405: err_msg = "Can't multiply AudioRegion by a non-int of type '{}'" amine@405: assert err_msg.format(_type) == str(type_err.value) amine@197: amine@254: amine@400: @pytest.mark.parametrize( amine@400: "data", amine@400: [ amine@405: [b"a" * 80, b"b" * 80], # simple amine@405: [b"a" * 31, b"b" * 31, b"c" * 30], # extra_samples_1 amine@405: [b"a" * 31, b"b" * 30, b"c" * 30], # extra_samples_2 amine@405: [b"a" * 11, b"b" * 11, b"c" * 10, b"c" * 10], # extra_samples_3 amine@400: ], amine@400: ids=[ amine@400: "simple", amine@400: "extra_samples_1", amine@400: "extra_samples_2", amine@400: "extra_samples_3", amine@400: ], amine@400: ) amine@400: def test_truediv(data): amine@254: amine@400: region = AudioRegion(b"".join(data), 80, 1, 1) amine@252: amine@400: sub_regions = region / len(data) amine@426: for data_i, region in zip( amine@426: data, amine@426: sub_regions, amine@426: ): amine@400: assert len(data_i) == len(bytes(region)) amine@254: amine@254: amine@400: @pytest.mark.parametrize( amine@405: "data, sample_width, channels, expected", amine@400: [ amine@405: (b"a" * 10, 1, 1, [97] * 10), # mono_sw_1 amine@405: (b"a" * 10, 2, 1, [24929] * 5), # mono_sw_2 amine@405: (b"a" * 8, 4, 1, [1633771873] * 2), # mono_sw_4 amine@405: (b"ab" * 5, 1, 2, [[97] * 5, [98] * 5]), # stereo_sw_1 amine@400: ], amine@400: ids=[ amine@400: "mono_sw_1", amine@400: "mono_sw_2", amine@400: "mono_sw_4", amine@400: "stereo_sw_1", amine@400: ], amine@400: ) amine@405: def test_samples(data, sample_width, channels, expected): amine@337: amine@400: region = AudioRegion(data, 10, sample_width, channels) amine@405: expected = np.array(expected) amine@405: assert (region.samples == expected).all() amine@405: assert (region.numpy() == expected).all() amine@405: assert (np.array(region) == expected).all()