amine@269: import os
amine@403: from collections import namedtuple
amine@403: from tempfile import TemporaryDirectory
amine@403: from unittest.mock import patch
amine@403: 
amine@400: import pytest
amine@267: 
amine@267: from auditok.cmdline_util import (
amine@273:     _AUDITOK_LOGGER,
amine@403:     KeywordArguments,
amine@403:     initialize_workers,
amine@267:     make_kwargs,
amine@267:     make_logger,
amine@267: )
amine@421: from auditok.exceptions import ArgumentError
amine@271: from auditok.workers import (
amine@418:     AudioEventsJoinerWorker,
amine@403:     CommandLineWorker,
amine@403:     PlayerWorker,
amine@403:     PrintWorker,
amine@403:     RegionSaverWorker,
amine@271:     StreamSaverWorker,
amine@271: )
amine@267: 
amine@400: _ArgsNamespace = namedtuple(
amine@400:     "_ArgsNamespace",
amine@267:     [
amine@267:         "input",
amine@292:         "max_read",
amine@267:         "analysis_window",
amine@267:         "sampling_rate",
amine@267:         "sample_width",
amine@267:         "channels",
amine@267:         "use_channel",
amine@267:         "input_format",
amine@267:         "output_format",
amine@267:         "large_file",
amine@267:         "frame_per_buffer",
amine@267:         "input_device_index",
amine@267:         "save_stream",
amine@292:         "save_detections_as",
amine@418:         "join_detections",
amine@267:         "plot",
amine@267:         "save_image",
amine@267:         "min_duration",
amine@267:         "max_duration",
amine@267:         "max_silence",
amine@267:         "drop_trailing_silence",
amine@267:         "strict_min_duration",
amine@267:         "energy_threshold",
amine@267:         "echo",
amine@292:         "progress_bar",
amine@267:         "command",
amine@267:         "quiet",
amine@267:         "printf",
amine@267:         "time_format",
amine@267:         "timestamp_format",
amine@267:     ],
amine@267: )
amine@267: 
amine@267: 
amine@400: @pytest.mark.parametrize(
amine@418:     "save_stream, save_detections_as, join_detections, plot, save_image, use_channel, exp_use_channel, exp_record",  # noqa: B950
amine@400:     [
amine@418:         # no_record_no_join
amine@418:         ("stream.ogg", None, None, False, None, "mix", "mix", False),
amine@418:         # no_record_plot_join
amine@418:         ("stream.ogg", None, 1.0, True, None, None, None, False),
amine@400:         # no_record_save_image
amine@418:         ("stream.ogg", None, None, True, "image.png", None, None, False),
amine@400:         # record_plot
amine@418:         (None, None, None, True, None, None, None, True),
amine@400:         # record_save_image
amine@418:         (None, None, None, False, "image.png", None, None, True),
amine@400:         # int_use_channel
amine@418:         ("stream.ogg", None, None, False, None, "1", 1, False),
amine@400:         # save_detections_as
amine@418:         ("stream.ogg", "{id}.wav", None, False, None, None, None, False),
amine@400:     ],
amine@400:     ids=[
amine@418:         "no_record_no_join",
amine@400:         "no_record_plot",
amine@400:         "no_record_save_image",
amine@400:         "record_plot",
amine@400:         "record_save_image",
amine@400:         "int_use_channel",
amine@400:         "save_detections_as",
amine@400:     ],
amine@400: )
amine@400: def test_make_kwargs(
amine@400:     save_stream,
amine@400:     save_detections_as,
amine@418:     join_detections,
amine@400:     plot,
amine@400:     save_image,
amine@400:     use_channel,
amine@400:     exp_use_channel,
amine@400:     exp_record,
amine@400: ):
amine@400:     args = (
amine@400:         "file",
amine@400:         30,
amine@400:         0.01,
amine@400:         16000,
amine@400:         2,
amine@400:         2,
amine@400:         use_channel,
amine@400:         "raw",
amine@400:         "ogg",
amine@400:         True,
amine@400:         None,
amine@400:         1,
amine@267:         save_stream,
amine@292:         save_detections_as,
amine@418:         join_detections,
amine@267:         plot,
amine@267:         save_image,
amine@400:         0.2,
amine@400:         10,
amine@400:         0.3,
amine@400:         False,
amine@400:         False,
amine@400:         55,
amine@400:     )
amine@400:     misc = (
amine@400:         False,
amine@400:         False,
amine@400:         None,
amine@400:         True,
amine@400:         None,
amine@400:         "TIME_FORMAT",
amine@400:         "TIMESTAMP_FORMAT",
amine@400:     )
amine@400:     args_ns = _ArgsNamespace(*(args + misc))
amine@267: 
amine@400:     io_kwargs = {
amine@400:         "input": "file",
amine@400:         "max_read": 30,
amine@400:         "block_dur": 0.01,
amine@400:         "sampling_rate": 16000,
amine@400:         "sample_width": 2,
amine@400:         "channels": 2,
amine@400:         "use_channel": exp_use_channel,
amine@400:         "save_stream": save_stream,
amine@400:         "save_detections_as": save_detections_as,
amine@418:         "join_detections": join_detections,
amine@400:         "audio_format": "raw",
amine@400:         "export_format": "ogg",
amine@400:         "large_file": True,
amine@400:         "frames_per_buffer": None,
amine@400:         "input_device_index": 1,
amine@400:         "record": exp_record,
amine@400:     }
amine@267: 
amine@400:     split_kwargs = {
amine@400:         "min_dur": 0.2,
amine@400:         "max_dur": 10,
amine@400:         "max_silence": 0.3,
amine@400:         "drop_trailing_silence": False,
amine@400:         "strict_min_dur": False,
amine@400:         "energy_threshold": 55,
amine@400:     }
amine@267: 
amine@400:     miscellaneous = {
amine@400:         "echo": False,
amine@400:         "command": None,
amine@400:         "progress_bar": False,
amine@400:         "quiet": True,
amine@400:         "printf": None,
amine@400:         "time_format": "TIME_FORMAT",
amine@400:         "timestamp_format": "TIMESTAMP_FORMAT",
amine@400:     }
amine@267: 
amine@400:     expected = KeywordArguments(io_kwargs, split_kwargs, miscellaneous)
amine@400:     kwargs = make_kwargs(args_ns)
amine@400:     assert kwargs == expected
amine@267: 
amine@268: 
amine@421: def test_make_kwargs_error():
amine@421: 
amine@421:     args = (
amine@421:         "file",
amine@421:         30,
amine@421:         0.01,
amine@421:         16000,
amine@421:         2,
amine@421:         2,
amine@421:         1,
amine@421:         "raw",
amine@421:         "ogg",
amine@421:         True,
amine@421:         None,
amine@421:         1,
amine@421:         None,  # save_stream
amine@421:         None,
amine@421:         1.0,  # join_detections
amine@421:         None,
amine@421:         None,
amine@421:         0.2,
amine@421:         10,
amine@421:         0.3,
amine@421:         False,
amine@421:         False,
amine@421:         55,
amine@421:         False,
amine@421:         False,
amine@421:         None,
amine@421:         True,
amine@421:         None,
amine@421:         "TIME_FORMAT",
amine@421:         "TIMESTAMP_FORMAT",
amine@421:     )
amine@421: 
amine@421:     args_ns = _ArgsNamespace(*args)
amine@421:     expected_err_msg = "using --join-detections/-j requires "
amine@421:     expected_err_msg += "--save-stream/-O to be specified."
amine@421:     with pytest.raises(ArgumentError) as arg_err:
amine@421:         make_kwargs(args_ns)
amine@421:     assert str(arg_err.value) == expected_err_msg
amine@421: 
amine@421: 
amine@400: def test_make_logger_stderr_and_file(capsys):
amine@400:     with TemporaryDirectory() as tmpdir:
amine@400:         file = os.path.join(tmpdir, "file.log")
amine@400:         logger = make_logger(stderr=True, file=file)
amine@400:         assert logger.name == _AUDITOK_LOGGER
amine@400:         assert len(logger.handlers) == 2
amine@400:         assert logger.handlers[1].stream.name == file
amine@400:         logger.info("This is a debug message")
amine@400:         captured = capsys.readouterr()
amine@400:         assert "This is a debug message" in captured.err
amine@400: 
amine@400: 
amine@400: def test_make_logger_None():
amine@400:     logger = make_logger(stderr=False, file=None)
amine@400:     assert logger is None
amine@400: 
amine@400: 
amine@418: def test_initialize_workers_all_plus_full_stream_saver():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@269:         with TemporaryDirectory() as tmpdir:
amine@400:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@271:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:                 save_stream=export_filename,
amine@400:                 export_format="wave",
amine@400:                 save_detections_as="{id}.wav",
amine@418:                 join_detections=None,
amine@400:                 echo=True,
amine@400:                 progress_bar=False,
amine@400:                 command="some command",
amine@400:                 quiet=False,
amine@400:                 printf="abcd",
amine@400:                 time_format="%S",
amine@400:                 timestamp_format="%h:%M:%S",
amine@400:             )
amine@400:             reader.stop()
amine@400:             assert patched_player_for.called
amine@400:             assert isinstance(reader, StreamSaverWorker)
amine@400:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@400:                 [
amine@400:                     RegionSaverWorker,
amine@400:                     PlayerWorker,
amine@400:                     CommandLineWorker,
amine@400:                     PrintWorker,
amine@400:                 ],
amine@418:             ):
amine@418:                 assert isinstance(obs, cls)
amine@418: 
amine@418: 
amine@418: def test_initialize_workers_all_plus_audio_event_joiner():
amine@418:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@418:         with TemporaryDirectory() as tmpdir:
amine@418:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@418:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@418:                 save_stream=export_filename,
amine@418:                 export_format="wave",
amine@418:                 save_detections_as="{id}.wav",
amine@418:                 join_detections=1,
amine@418:                 echo=True,
amine@418:                 progress_bar=False,
amine@418:                 command="some command",
amine@418:                 quiet=False,
amine@418:                 printf="abcd",
amine@418:                 time_format="%S",
amine@418:                 timestamp_format="%h:%M:%S",
amine@418:             )
amine@418:             assert patched_player_for.called
amine@418:             assert not isinstance(reader, StreamSaverWorker)
amine@418:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@418:                 [
amine@418:                     AudioEventsJoinerWorker,
amine@418:                     RegionSaverWorker,
amine@418:                     PlayerWorker,
amine@418:                     CommandLineWorker,
amine@418:                     PrintWorker,
amine@418:                 ],
amine@400:             ):
amine@400:                 assert isinstance(obs, cls)
amine@400: 
amine@400: 
amine@400: def test_initialize_workers_no_RegionSaverWorker():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@400:         with TemporaryDirectory() as tmpdir:
amine@400:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@400:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:                 save_stream=export_filename,
amine@271:                 export_format="wave",
amine@271:                 save_detections_as=None,
amine@418:                 join_detections=None,
amine@271:                 echo=True,
amine@271:                 progress_bar=False,
amine@400:                 command="some command",
amine@400:                 quiet=False,
amine@400:                 printf="abcd",
amine@400:                 time_format="%S",
amine@400:                 timestamp_format="%h:%M:%S",
amine@400:             )
amine@400:             reader.stop()
amine@400:             assert patched_player_for.called
amine@400:             assert isinstance(reader, StreamSaverWorker)
amine@400:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@418:                 [PlayerWorker, CommandLineWorker, PrintWorker],
amine@400:             ):
amine@400:                 assert isinstance(obs, cls)
amine@400: 
amine@400: 
amine@400: def test_initialize_workers_no_PlayerWorker():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@400:         with TemporaryDirectory() as tmpdir:
amine@400:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@400:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:                 save_stream=export_filename,
amine@400:                 export_format="wave",
amine@400:                 save_detections_as="{id}.wav",
amine@418:                 join_detections=None,
amine@400:                 echo=False,
amine@400:                 progress_bar=False,
amine@400:                 command="some command",
amine@400:                 quiet=False,
amine@400:                 printf="abcd",
amine@400:                 time_format="%S",
amine@400:                 timestamp_format="%h:%M:%S",
amine@400:             )
amine@400:             reader.stop()
amine@400:             assert not patched_player_for.called
amine@400:             assert isinstance(reader, StreamSaverWorker)
amine@400:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@400:                 [RegionSaverWorker, CommandLineWorker, PrintWorker],
amine@400:             ):
amine@400:                 assert isinstance(obs, cls)
amine@400: 
amine@400: 
amine@400: def test_initialize_workers_no_CommandLineWorker():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@400:         with TemporaryDirectory() as tmpdir:
amine@400:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@400:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:                 save_stream=export_filename,
amine@400:                 export_format="wave",
amine@400:                 save_detections_as="{id}.wav",
amine@418:                 join_detections=None,
amine@400:                 echo=True,
amine@400:                 progress_bar=False,
amine@271:                 command=None,
amine@400:                 quiet=False,
amine@400:                 printf="abcd",
amine@400:                 time_format="%S",
amine@400:                 timestamp_format="%h:%M:%S",
amine@400:             )
amine@400:             reader.stop()
amine@400:             assert patched_player_for.called
amine@400:             assert isinstance(reader, StreamSaverWorker)
amine@400:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@418:                 [RegionSaverWorker, PlayerWorker, PrintWorker],
amine@400:             ):
amine@400:                 assert isinstance(obs, cls)
amine@400: 
amine@400: 
amine@400: def test_initialize_workers_no_PrintWorker():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@400:         with TemporaryDirectory() as tmpdir:
amine@400:             export_filename = os.path.join(tmpdir, "output.wav")
amine@418:             reader, tokenizer_worker = initialize_workers(
amine@400:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:                 save_stream=export_filename,
amine@400:                 export_format="wave",
amine@400:                 save_detections_as="{id}.wav",
amine@418:                 join_detections=None,
amine@400:                 echo=True,
amine@400:                 progress_bar=False,
amine@400:                 command="some command",
amine@271:                 quiet=True,
amine@271:                 printf="abcd",
amine@271:                 time_format="%S",
amine@271:                 timestamp_format="%h:%M:%S",
amine@271:             )
amine@400:             reader.stop()
amine@400:             assert patched_player_for.called
amine@400:             assert isinstance(reader, StreamSaverWorker)
amine@400:             for obs, cls in zip(
amine@418:                 tokenizer_worker._observers,
amine@400:                 [RegionSaverWorker, PlayerWorker, CommandLineWorker],
amine@400:             ):
amine@400:                 assert isinstance(obs, cls)
amine@337: 
amine@337: 
amine@400: def test_initialize_workers_no_observers():
amine@400:     with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@418:         reader, tokenizer_worker = initialize_workers(
amine@400:             input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@400:             save_stream=None,
amine@400:             export_format="wave",
amine@400:             save_detections_as=None,
amine@400:             echo=True,
amine@400:             progress_bar=False,
amine@400:             command=None,
amine@400:             quiet=True,
amine@400:             printf="abcd",
amine@400:             time_format="%S",
amine@400:             timestamp_format="%h:%M:%S",
amine@400:         )
amine@400:         assert patched_player_for.called
amine@400:         assert not isinstance(reader, StreamSaverWorker)
amine@418:         assert len(tokenizer_worker._observers) == 1