amine@269: import os
amine@267: from unittest import TestCase
amine@271: from unittest.mock import patch
amine@269: from tempfile import TemporaryDirectory
amine@267: from collections import namedtuple
amine@267: from genty import genty, genty_dataset
amine@267: 
amine@267: from auditok.cmdline_util import (
amine@273:     _AUDITOK_LOGGER,
amine@267:     make_kwargs,
amine@291:     make_duration_formatter,
amine@267:     make_logger,
amine@267:     initialize_workers,
amine@267:     KeywordArguments,
amine@267: )
amine@271: from auditok.workers import (
amine@271:     StreamSaverWorker,
amine@271:     RegionSaverWorker,
amine@271:     PlayerWorker,
amine@271:     CommandLineWorker,
amine@271:     PrintWorker,
amine@271: )
amine@267: from auditok.exceptions import TimeFormatError
amine@267: 
amine@267: _ArgsNamespece = namedtuple(
amine@267:     "_ArgsNamespece",
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@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@267: @genty
amine@292: class TestCmdLineUtil(TestCase):
amine@267:     @genty_dataset(
amine@292:         no_record=("stream.ogg", None, False, None, "mix", "mix", False),
amine@292:         no_record_plot=("stream.ogg", None, True, None, None, None, False),
amine@267:         no_record_save_image=(
amine@267:             "stream.ogg",
amine@292:             None,
amine@267:             True,
amine@267:             "image.png",
amine@267:             None,
amine@267:             None,
amine@267:             False,
amine@267:         ),
amine@292:         record_plot=(None, None, True, None, None, None, True),
amine@292:         record_save_image=(None, None, False, "image.png", None, None, True),
amine@292:         int_use_channel=("stream.ogg", None, False, None, "1", 1, False),
amine@292:         save_detections_as=("stream.ogg", "{id}.wav", False, None, None, None, False)
amine@267:     )
amine@267:     def test_make_kwargs(
amine@267:         self,
amine@267:         save_stream,
amine@292:         save_detections_as,
amine@267:         plot,
amine@267:         save_image,
amine@267:         use_channel,
amine@267:         exp_use_channel,
amine@267:         exp_record,
amine@267:     ):
amine@267: 
amine@267:         args = (
amine@267:             "file",
amine@267:             30,
amine@267:             0.01,
amine@267:             16000,
amine@267:             2,
amine@267:             2,
amine@267:             use_channel,
amine@267:             "raw",
amine@267:             "ogg",
amine@267:             True,
amine@267:             None,
amine@267:             1,
amine@267:             save_stream,
amine@292:             save_detections_as,
amine@267:             plot,
amine@267:             save_image,
amine@267:             0.2,
amine@267:             10,
amine@267:             0.3,
amine@267:             False,
amine@267:             False,
amine@267:             55,
amine@267:         )
amine@292:         misc = (False, False, None, True, None, "TIME_FORMAT", "TIMESTAMP_FORMAT")
amine@267:         args_ns = _ArgsNamespece(*(args + misc))
amine@267: 
amine@267:         io_kwargs = {
amine@267:             "input": "file",
amine@267:             "max_read": 30,
amine@267:             "block_dur": 0.01,
amine@267:             "sampling_rate": 16000,
amine@267:             "sample_width": 2,
amine@267:             "channels": 2,
amine@267:             "use_channel": exp_use_channel,
amine@267:             "save_stream": save_stream,
amine@292:             "save_detections_as": save_detections_as,
amine@267:             "audio_format": "raw",
amine@267:             "export_format": "ogg",
amine@267:             "large_file": True,
amine@267:             "frames_per_buffer": None,
amine@267:             "input_device_index": 1,
amine@267:             "record": exp_record,
amine@267:         }
amine@267: 
amine@267:         split_kwargs = {
amine@267:             "min_dur": 0.2,
amine@267:             "max_dur": 10,
amine@267:             "max_silence": 0.3,
amine@267:             "drop_trailing_silence": False,
amine@267:             "strict_min_dur": False,
amine@267:             "energy_threshold": 55,
amine@267:         }
amine@267: 
amine@267:         miscellaneous = {
amine@267:             "echo": False,
amine@267:             "command": None,
amine@292:             "progress_bar": False,
amine@267:             "quiet": True,
amine@267:             "printf": None,
amine@267:             "time_format": "TIME_FORMAT",
amine@267:             "timestamp_format": "TIMESTAMP_FORMAT",
amine@267:         }
amine@267: 
amine@267:         expected = KeywordArguments(io_kwargs, split_kwargs, miscellaneous)
amine@267:         kwargs = make_kwargs(args_ns)
amine@267:         self.assertEqual(kwargs, expected)
amine@268: 
amine@268:     @genty_dataset(
amine@268:         only_seconds=("%S", 5400, "5400.000"),
amine@268:         only_millis=("%I", 5400, "5400000"),
amine@268:         full=("%h:%m:%s.%i", 3725.365, "01:02:05.365"),
amine@268:         full_zero_hours=("%h:%m:%s.%i", 1925.075, "00:32:05.075"),
amine@268:         full_zero_minutes=("%h:%m:%s.%i", 3659.075, "01:00:59.075"),
amine@268:         full_zero_seconds=("%h:%m:%s.%i", 3720.075, "01:02:00.075"),
amine@268:         full_zero_millis=("%h:%m:%s.%i", 3725, "01:02:05.000"),
amine@268:         duplicate_directive=(
amine@268:             "%h %h:%m:%s.%i %s",
amine@268:             3725.365,
amine@268:             "01 01:02:05.365 05",
amine@268:         ),
amine@268:         no_millis=("%h:%m:%s", 3725, "01:02:05"),
amine@268:         no_seconds=("%h:%m", 3725, "01:02"),
amine@268:         no_minutes=("%h", 3725, "01"),
amine@268:         no_hours=("%m:%s.%i", 3725, "02:05.000"),
amine@268:     )
amine@291:     def test_make_duration_formatter(self, fmt, duration, expected):
amine@291:         formatter = make_duration_formatter(fmt)
amine@268:         result = formatter(duration)
amine@268:         self.assertEqual(result, expected)
amine@268: 
amine@268:     @genty_dataset(
amine@268:         duplicate_only_seconds=("%S %S",),
amine@268:         duplicate_only_millis=("%I %I",),
amine@268:         unknown_directive=("%x",),
amine@268:     )
amine@291:     def test_make_duration_formatter_error(self, fmt):
amine@268:         with self.assertRaises(TimeFormatError):
amine@291:             make_duration_formatter(fmt)
amine@269: 
amine@273:     def test_make_logger_stderr_and_file(self):
amine@269:         with TemporaryDirectory() as tmpdir:
amine@269:             file = os.path.join(tmpdir, "file.log")
amine@273:             logger = make_logger(stderr=True, file=file)
amine@273:             self.assertEqual(logger.name, _AUDITOK_LOGGER)
amine@269:             self.assertEqual(len(logger.handlers), 2)
amine@273:             self.assertEqual(logger.handlers[0].stream.name, "<stderr>")
amine@269:             self.assertEqual(logger.handlers[1].stream.name, file)
amine@269: 
amine@269:     def test_make_logger_None(self):
amine@273:         logger = make_logger(stderr=False, file=None)
amine@269:         self.assertIsNone(logger)
amine@271: 
amine@271:     def test_initialize_workers_all(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285:             with TemporaryDirectory() as tmpdir:
amine@285:                 export_filename = os.path.join(tmpdir, "output.wav")
amine@285:                 reader, observers = initialize_workers(
amine@285:                     input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285:                     save_stream=export_filename,
amine@285:                     export_format="wave",
amine@285:                     save_detections_as="{id}.wav",
amine@285:                     echo=True,
amine@285:                     progress_bar=False,
amine@285:                     command="some command",
amine@285:                     quiet=False,
amine@285:                     printf="abcd",
amine@285:                     time_format="%S",
amine@285:                     timestamp_format="%h:%M:%S",
amine@285:                 )
amine@285:                 reader.stop()
amine@285:                 self.assertTrue(patched_player_for.called)
amine@285:                 self.assertIsInstance(reader, StreamSaverWorker)
amine@285:                 for obs, cls in zip(
amine@285:                     observers,
amine@285:                     [
amine@285:                         RegionSaverWorker,
amine@285:                         PlayerWorker,
amine@285:                         CommandLineWorker,
amine@285:                         PrintWorker,
amine@285:                     ],
amine@285:                 ):
amine@285:                     self.assertIsInstance(obs, cls)
amine@271: 
amine@271:     def test_initialize_workers_no_RegionSaverWorker(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285:             with TemporaryDirectory() as tmpdir:
amine@285:                 export_filename = os.path.join(tmpdir, "output.wav")
amine@285:                 reader, observers = initialize_workers(
amine@285:                     input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285:                     save_stream=export_filename,
amine@285:                     export_format="wave",
amine@285:                     save_detections_as=None,
amine@285:                     echo=True,
amine@285:                     progress_bar=False,
amine@285:                     command="some command",
amine@285:                     quiet=False,
amine@285:                     printf="abcd",
amine@285:                     time_format="%S",
amine@285:                     timestamp_format="%h:%M:%S",
amine@285:                 )
amine@285:                 reader.stop()
amine@285:                 self.assertTrue(patched_player_for.called)
amine@285:                 self.assertIsInstance(reader, StreamSaverWorker)
amine@285:                 for obs, cls in zip(
amine@285:                     observers, [PlayerWorker, CommandLineWorker, PrintWorker]
amine@285:                 ):
amine@285:                     self.assertIsInstance(obs, cls)
amine@271: 
amine@271:     def test_initialize_workers_no_PlayerWorker(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285:             with TemporaryDirectory() as tmpdir:
amine@285:                 export_filename = os.path.join(tmpdir, "output.wav")
amine@285:                 reader, observers = initialize_workers(
amine@285:                     input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285:                     save_stream=export_filename,
amine@285:                     export_format="wave",
amine@285:                     save_detections_as="{id}.wav",
amine@285:                     echo=False,
amine@285:                     progress_bar=False,
amine@285:                     command="some command",
amine@285:                     quiet=False,
amine@285:                     printf="abcd",
amine@285:                     time_format="%S",
amine@285:                     timestamp_format="%h:%M:%S",
amine@285:                 )
amine@285:                 reader.stop()
amine@285:                 self.assertFalse(patched_player_for.called)
amine@285:                 self.assertIsInstance(reader, StreamSaverWorker)
amine@285:                 for obs, cls in zip(
amine@285:                     observers,
amine@285:                     [RegionSaverWorker, CommandLineWorker, PrintWorker],
amine@285:                 ):
amine@285:                     self.assertIsInstance(obs, cls)
amine@271: 
amine@271:     def test_initialize_workers_no_CommandLineWorker(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285:             with TemporaryDirectory() as tmpdir:
amine@285:                 export_filename = os.path.join(tmpdir, "output.wav")
amine@285:                 reader, observers = initialize_workers(
amine@285:                     input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285:                     save_stream=export_filename,
amine@285:                     export_format="wave",
amine@285:                     save_detections_as="{id}.wav",
amine@285:                     echo=True,
amine@285:                     progress_bar=False,
amine@285:                     command=None,
amine@285:                     quiet=False,
amine@285:                     printf="abcd",
amine@285:                     time_format="%S",
amine@285:                     timestamp_format="%h:%M:%S",
amine@285:                 )
amine@285:                 reader.stop()
amine@285:                 self.assertTrue(patched_player_for.called)
amine@285:                 self.assertIsInstance(reader, StreamSaverWorker)
amine@285:                 for obs, cls in zip(
amine@285:                     observers, [RegionSaverWorker, PlayerWorker, PrintWorker]
amine@285:                 ):
amine@285:                     self.assertIsInstance(obs, cls)
amine@271: 
amine@271:     def test_initialize_workers_no_PrintWorker(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285:             with TemporaryDirectory() as tmpdir:
amine@285:                 export_filename = os.path.join(tmpdir, "output.wav")
amine@285:                 reader, observers = initialize_workers(
amine@285:                     input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285:                     save_stream=export_filename,
amine@285:                     export_format="wave",
amine@285:                     save_detections_as="{id}.wav",
amine@285:                     echo=True,
amine@285:                     progress_bar=False,
amine@285:                     command="some command",
amine@285:                     quiet=True,
amine@285:                     printf="abcd",
amine@285:                     time_format="%S",
amine@285:                     timestamp_format="%h:%M:%S",
amine@285:                 )
amine@285:                 reader.stop()
amine@285:                 self.assertTrue(patched_player_for.called)
amine@285:                 self.assertIsInstance(reader, StreamSaverWorker)
amine@285:                 for obs, cls in zip(
amine@285:                     observers,
amine@285:                     [RegionSaverWorker, PlayerWorker, CommandLineWorker],
amine@285:                 ):
amine@285:                     self.assertIsInstance(obs, cls)
amine@271: 
amine@271:     def test_initialize_workers_no_observers(self):
amine@271:         with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@271:             reader, observers = initialize_workers(
amine@271:                 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@271:                 save_stream=None,
amine@271:                 export_format="wave",
amine@271:                 save_detections_as=None,
amine@271:                 echo=True,
amine@271:                 progress_bar=False,
amine@271:                 command=None,
amine@271:                 quiet=True,
amine@271:                 printf="abcd",
amine@271:                 time_format="%S",
amine@271:                 timestamp_format="%h:%M:%S",
amine@271:             )
amine@271:             self.assertTrue(patched_player_for.called)
amine@271:             self.assertFalse(isinstance(reader, StreamSaverWorker))
amine@271:             self.assertTrue(len(observers), 0)