Mercurial > hg > auditok

annotate tests/test_cmdline_util.py @ 386:059599823c34

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merge branch 'master' of https://github.com/amsehili/auditok
author www-data <www-data@c4dm-xenserv-virt2.eecs.qmul.ac.uk>
date Tue, 02 Mar 2021 19:17:59 +0000
parents 5732edbfae30
children 323d59b404a2
rev   line source
amine@269 1 import os
amine@337 2 import unittest
amine@267 3 from unittest import TestCase
amine@271 4 from unittest.mock import patch
amine@269 5 from tempfile import TemporaryDirectory
amine@267 6 from collections import namedtuple
amine@267 7 from genty import genty, genty_dataset
amine@267 8
amine@267 9 from auditok.cmdline_util import (
amine@273 10 _AUDITOK_LOGGER,
amine@267 11 make_kwargs,
amine@267 12 make_logger,
amine@267 13 initialize_workers,
amine@267 14 KeywordArguments,
amine@267 15 )
amine@271 16 from auditok.workers import (
amine@271 17 StreamSaverWorker,
amine@271 18 RegionSaverWorker,
amine@271 19 PlayerWorker,
amine@271 20 CommandLineWorker,
amine@271 21 PrintWorker,
amine@271 22 )
amine@267 23
amine@267 24 _ArgsNamespece = namedtuple(
amine@267 25 "_ArgsNamespece",
amine@267 26 [
amine@267 27 "input",
amine@292 28 "max_read",
amine@267 29 "analysis_window",
amine@267 30 "sampling_rate",
amine@267 31 "sample_width",
amine@267 32 "channels",
amine@267 33 "use_channel",
amine@267 34 "input_format",
amine@267 35 "output_format",
amine@267 36 "large_file",
amine@267 37 "frame_per_buffer",
amine@267 38 "input_device_index",
amine@267 39 "save_stream",
amine@292 40 "save_detections_as",
amine@267 41 "plot",
amine@267 42 "save_image",
amine@267 43 "min_duration",
amine@267 44 "max_duration",
amine@267 45 "max_silence",
amine@267 46 "drop_trailing_silence",
amine@267 47 "strict_min_duration",
amine@267 48 "energy_threshold",
amine@267 49 "echo",
amine@292 50 "progress_bar",
amine@267 51 "command",
amine@267 52 "quiet",
amine@267 53 "printf",
amine@267 54 "time_format",
amine@267 55 "timestamp_format",
amine@267 56 ],
amine@267 57 )
amine@267 58
amine@267 59
amine@267 60 @genty
amine@292 61 class TestCmdLineUtil(TestCase):
amine@267 62 @genty_dataset(
amine@292 63 no_record=("stream.ogg", None, False, None, "mix", "mix", False),
amine@292 64 no_record_plot=("stream.ogg", None, True, None, None, None, False),
amine@267 65 no_record_save_image=(
amine@267 66 "stream.ogg",
amine@292 67 None,
amine@267 68 True,
amine@267 69 "image.png",
amine@267 70 None,
amine@267 71 None,
amine@267 72 False,
amine@267 73 ),
amine@292 74 record_plot=(None, None, True, None, None, None, True),
amine@292 75 record_save_image=(None, None, False, "image.png", None, None, True),
amine@292 76 int_use_channel=("stream.ogg", None, False, None, "1", 1, False),
amine@331 77 save_detections_as=(
amine@331 78 "stream.ogg",
amine@331 79 "{id}.wav",
amine@331 80 False,
amine@331 81 None,
amine@331 82 None,
amine@331 83 None,
amine@331 84 False,
amine@331 85 ),
amine@267 86 )
amine@267 87 def test_make_kwargs(
amine@267 88 self,
amine@267 89 save_stream,
amine@292 90 save_detections_as,
amine@267 91 plot,
amine@267 92 save_image,
amine@267 93 use_channel,
amine@267 94 exp_use_channel,
amine@267 95 exp_record,
amine@267 96 ):
amine@267 97
amine@267 98 args = (
amine@267 99 "file",
amine@267 100 30,
amine@267 101 0.01,
amine@267 102 16000,
amine@267 103 2,
amine@267 104 2,
amine@267 105 use_channel,
amine@267 106 "raw",
amine@267 107 "ogg",
amine@267 108 True,
amine@267 109 None,
amine@267 110 1,
amine@267 111 save_stream,
amine@292 112 save_detections_as,
amine@267 113 plot,
amine@267 114 save_image,
amine@267 115 0.2,
amine@267 116 10,
amine@267 117 0.3,
amine@267 118 False,
amine@267 119 False,
amine@267 120 55,
amine@267 121 )
amine@331 122 misc = (
amine@331 123 False,
amine@331 124 False,
amine@331 125 None,
amine@331 126 True,
amine@331 127 None,
amine@331 128 "TIME_FORMAT",
amine@331 129 "TIMESTAMP_FORMAT",
amine@331 130 )
amine@267 131 args_ns = _ArgsNamespece(*(args + misc))
amine@267 132
amine@267 133 io_kwargs = {
amine@267 134 "input": "file",
amine@267 135 "max_read": 30,
amine@267 136 "block_dur": 0.01,
amine@267 137 "sampling_rate": 16000,
amine@267 138 "sample_width": 2,
amine@267 139 "channels": 2,
amine@267 140 "use_channel": exp_use_channel,
amine@267 141 "save_stream": save_stream,
amine@292 142 "save_detections_as": save_detections_as,
amine@267 143 "audio_format": "raw",
amine@267 144 "export_format": "ogg",
amine@267 145 "large_file": True,
amine@267 146 "frames_per_buffer": None,
amine@267 147 "input_device_index": 1,
amine@267 148 "record": exp_record,
amine@267 149 }
amine@267 150
amine@267 151 split_kwargs = {
amine@267 152 "min_dur": 0.2,
amine@267 153 "max_dur": 10,
amine@267 154 "max_silence": 0.3,
amine@267 155 "drop_trailing_silence": False,
amine@267 156 "strict_min_dur": False,
amine@267 157 "energy_threshold": 55,
amine@267 158 }
amine@267 159
amine@267 160 miscellaneous = {
amine@267 161 "echo": False,
amine@267 162 "command": None,
amine@292 163 "progress_bar": False,
amine@267 164 "quiet": True,
amine@267 165 "printf": None,
amine@267 166 "time_format": "TIME_FORMAT",
amine@267 167 "timestamp_format": "TIMESTAMP_FORMAT",
amine@267 168 }
amine@267 169
amine@267 170 expected = KeywordArguments(io_kwargs, split_kwargs, miscellaneous)
amine@267 171 kwargs = make_kwargs(args_ns)
amine@267 172 self.assertEqual(kwargs, expected)
amine@268 173
amine@273 174 def test_make_logger_stderr_and_file(self):
amine@269 175 with TemporaryDirectory() as tmpdir:
amine@269 176 file = os.path.join(tmpdir, "file.log")
amine@273 177 logger = make_logger(stderr=True, file=file)
amine@273 178 self.assertEqual(logger.name, _AUDITOK_LOGGER)
amine@269 179 self.assertEqual(len(logger.handlers), 2)
amine@273 180 self.assertEqual(logger.handlers[0].stream.name, "<stderr>")
amine@269 181 self.assertEqual(logger.handlers[1].stream.name, file)
amine@269 182
amine@269 183 def test_make_logger_None(self):
amine@273 184 logger = make_logger(stderr=False, file=None)
amine@269 185 self.assertIsNone(logger)
amine@271 186
amine@271 187 def test_initialize_workers_all(self):
amine@271 188 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285 189 with TemporaryDirectory() as tmpdir:
amine@285 190 export_filename = os.path.join(tmpdir, "output.wav")
amine@285 191 reader, observers = initialize_workers(
amine@285 192 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285 193 save_stream=export_filename,
amine@285 194 export_format="wave",
amine@285 195 save_detections_as="{id}.wav",
amine@285 196 echo=True,
amine@285 197 progress_bar=False,
amine@285 198 command="some command",
amine@285 199 quiet=False,
amine@285 200 printf="abcd",
amine@285 201 time_format="%S",
amine@285 202 timestamp_format="%h:%M:%S",
amine@285 203 )
amine@285 204 reader.stop()
amine@285 205 self.assertTrue(patched_player_for.called)
amine@285 206 self.assertIsInstance(reader, StreamSaverWorker)
amine@285 207 for obs, cls in zip(
amine@285 208 observers,
amine@285 209 [
amine@285 210 RegionSaverWorker,
amine@285 211 PlayerWorker,
amine@285 212 CommandLineWorker,
amine@285 213 PrintWorker,
amine@285 214 ],
amine@285 215 ):
amine@285 216 self.assertIsInstance(obs, cls)
amine@271 217
amine@271 218 def test_initialize_workers_no_RegionSaverWorker(self):
amine@271 219 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285 220 with TemporaryDirectory() as tmpdir:
amine@285 221 export_filename = os.path.join(tmpdir, "output.wav")
amine@285 222 reader, observers = initialize_workers(
amine@285 223 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285 224 save_stream=export_filename,
amine@285 225 export_format="wave",
amine@285 226 save_detections_as=None,
amine@285 227 echo=True,
amine@285 228 progress_bar=False,
amine@285 229 command="some command",
amine@285 230 quiet=False,
amine@285 231 printf="abcd",
amine@285 232 time_format="%S",
amine@285 233 timestamp_format="%h:%M:%S",
amine@285 234 )
amine@285 235 reader.stop()
amine@285 236 self.assertTrue(patched_player_for.called)
amine@285 237 self.assertIsInstance(reader, StreamSaverWorker)
amine@285 238 for obs, cls in zip(
amine@285 239 observers, [PlayerWorker, CommandLineWorker, PrintWorker]
amine@285 240 ):
amine@285 241 self.assertIsInstance(obs, cls)
amine@271 242
amine@271 243 def test_initialize_workers_no_PlayerWorker(self):
amine@271 244 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285 245 with TemporaryDirectory() as tmpdir:
amine@285 246 export_filename = os.path.join(tmpdir, "output.wav")
amine@285 247 reader, observers = initialize_workers(
amine@285 248 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285 249 save_stream=export_filename,
amine@285 250 export_format="wave",
amine@285 251 save_detections_as="{id}.wav",
amine@285 252 echo=False,
amine@285 253 progress_bar=False,
amine@285 254 command="some command",
amine@285 255 quiet=False,
amine@285 256 printf="abcd",
amine@285 257 time_format="%S",
amine@285 258 timestamp_format="%h:%M:%S",
amine@285 259 )
amine@285 260 reader.stop()
amine@285 261 self.assertFalse(patched_player_for.called)
amine@285 262 self.assertIsInstance(reader, StreamSaverWorker)
amine@285 263 for obs, cls in zip(
amine@285 264 observers,
amine@285 265 [RegionSaverWorker, CommandLineWorker, PrintWorker],
amine@285 266 ):
amine@285 267 self.assertIsInstance(obs, cls)
amine@271 268
amine@271 269 def test_initialize_workers_no_CommandLineWorker(self):
amine@271 270 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285 271 with TemporaryDirectory() as tmpdir:
amine@285 272 export_filename = os.path.join(tmpdir, "output.wav")
amine@285 273 reader, observers = initialize_workers(
amine@285 274 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285 275 save_stream=export_filename,
amine@285 276 export_format="wave",
amine@285 277 save_detections_as="{id}.wav",
amine@285 278 echo=True,
amine@285 279 progress_bar=False,
amine@285 280 command=None,
amine@285 281 quiet=False,
amine@285 282 printf="abcd",
amine@285 283 time_format="%S",
amine@285 284 timestamp_format="%h:%M:%S",
amine@285 285 )
amine@285 286 reader.stop()
amine@285 287 self.assertTrue(patched_player_for.called)
amine@285 288 self.assertIsInstance(reader, StreamSaverWorker)
amine@285 289 for obs, cls in zip(
amine@285 290 observers, [RegionSaverWorker, PlayerWorker, PrintWorker]
amine@285 291 ):
amine@285 292 self.assertIsInstance(obs, cls)
amine@271 293
amine@271 294 def test_initialize_workers_no_PrintWorker(self):
amine@271 295 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@285 296 with TemporaryDirectory() as tmpdir:
amine@285 297 export_filename = os.path.join(tmpdir, "output.wav")
amine@285 298 reader, observers = initialize_workers(
amine@285 299 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@285 300 save_stream=export_filename,
amine@285 301 export_format="wave",
amine@285 302 save_detections_as="{id}.wav",
amine@285 303 echo=True,
amine@285 304 progress_bar=False,
amine@285 305 command="some command",
amine@285 306 quiet=True,
amine@285 307 printf="abcd",
amine@285 308 time_format="%S",
amine@285 309 timestamp_format="%h:%M:%S",
amine@285 310 )
amine@285 311 reader.stop()
amine@285 312 self.assertTrue(patched_player_for.called)
amine@285 313 self.assertIsInstance(reader, StreamSaverWorker)
amine@285 314 for obs, cls in zip(
amine@285 315 observers,
amine@285 316 [RegionSaverWorker, PlayerWorker, CommandLineWorker],
amine@285 317 ):
amine@285 318 self.assertIsInstance(obs, cls)
amine@271 319
amine@271 320 def test_initialize_workers_no_observers(self):
amine@271 321 with patch("auditok.cmdline_util.player_for") as patched_player_for:
amine@271 322 reader, observers = initialize_workers(
amine@271 323 input="tests/data/test_16KHZ_mono_400Hz.wav",
amine@271 324 save_stream=None,
amine@271 325 export_format="wave",
amine@271 326 save_detections_as=None,
amine@271 327 echo=True,
amine@271 328 progress_bar=False,
amine@271 329 command=None,
amine@271 330 quiet=True,
amine@271 331 printf="abcd",
amine@271 332 time_format="%S",
amine@271 333 timestamp_format="%h:%M:%S",
amine@271 334 )
amine@271 335 self.assertTrue(patched_player_for.called)
amine@271 336 self.assertFalse(isinstance(reader, StreamSaverWorker))
amine@271 337 self.assertTrue(len(observers), 0)
amine@337 338
amine@337 339
amine@337 340 if __name__ == "__main__":
amine@337 341 unittest.main()