amine@269: import os amine@267: from unittest import TestCase 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@267: LOGGER_NAME, amine@267: make_kwargs, amine@267: make_duration_fromatter, amine@267: make_logger, amine@267: initialize_workers, amine@267: KeywordArguments, amine@267: ) amine@267: from auditok.workers import StreamSaverWorker amine@267: from auditok.exceptions import TimeFormatError amine@267: amine@267: _ArgsNamespece = namedtuple( amine@267: "_ArgsNamespece", amine@267: [ amine@267: "input", amine@267: "max_time", 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@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@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@268: class _TestCmdLineUtil(TestCase): amine@267: @genty_dataset( amine@267: no_record=("stream.ogg", False, None, "mix", "mix", False), amine@267: no_record_plot=("stream.ogg", True, None, None, None, False), amine@267: no_record_save_image=( amine@267: "stream.ogg", amine@267: True, amine@267: "image.png", amine@267: None, amine@267: None, amine@267: False, amine@267: ), amine@267: record_plot=(None, True, None, None, None, True), amine@267: record_save_image=(None, False, "image.png", None, None, True), amine@267: int_use_channel=("stream.ogg", False, None, "1", 1, False), amine@267: ) amine@267: def test_make_kwargs( amine@267: self, amine@267: save_stream, 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@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@267: misc = (False, None, True, None, "TIME_FORMAT", "TIMESTAMP_FORMAT") amine@267: 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@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@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@268: def test_make_duration_fromatter(self, fmt, duration, expected): amine@268: formatter = make_duration_fromatter(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@268: def test_make_duration_fromatter_error(self, fmt): amine@268: with self.assertRaises(TimeFormatError): amine@268: make_duration_fromatter(fmt) amine@269: amine@269: def test_make_logger_stdout_and_file(self): amine@269: with TemporaryDirectory() as tmpdir: amine@269: file = os.path.join(tmpdir, "file.log") amine@269: logger = make_logger(debug_stdout=True, debug_file=file) amine@269: self.assertEqual(logger.name, LOGGER_NAME) amine@269: self.assertEqual(len(logger.handlers), 2) amine@269: self.assertEqual(logger.handlers[0].stream.name, "") amine@269: self.assertEqual(logger.handlers[1].stream.name, file) amine@269: amine@269: def test_make_logger_None(self): amine@269: logger = make_logger(debug_stdout=False, debug_file=None) amine@269: self.assertIsNone(logger) amine@269: del logger