Mercurial > hg > auditok
changeset 345:5732edbfae30
Use python 3.4 compatible syntax
Remove the need of circular imports between workers and command_line_util
| author | Amine Sehili <amine.sehili@gmail.com> |
|---|---|
| date | Mon, 11 Nov 2019 09:38:47 +0100 |
| parents | deb4bbed0ecd |
| children | 979343fe59e0 |
| files | auditok/cmdline_util.py auditok/io.py auditok/util.py auditok/workers.py tests/test_cmdline_util.py tests/test_util.py tests/test_workers.py |
| diffstat | 7 files changed, 85 insertions(+), 81 deletions(-) [+] |
line wrap: on
line diff
--- a/auditok/cmdline_util.py Sun Nov 10 20:20:51 2019 +0100 +++ b/auditok/cmdline_util.py Mon Nov 11 09:38:47 2019 +0100 @@ -4,7 +4,6 @@ from auditok import workers from .util import AudioDataSource from .io import player_for -from .exceptions import TimeFormatError _AUDITOK_LOGGER = "AUDITOK_LOGGER" KeywordArguments = namedtuple( @@ -61,43 +60,6 @@ return KeywordArguments(io_kwargs, split_kwargs, miscellaneous) -def make_duration_formatter(fmt): - """ - Accepted format directives: %i %s %m %h - """ - if fmt == "%S": - - def fromatter(seconds): - return "{:.3f}".format(seconds) - - elif fmt == "%I": - - def fromatter(seconds): - return "{0}".format(int(seconds * 1000)) - - else: - fmt = fmt.replace("%h", "{hrs:02d}") - fmt = fmt.replace("%m", "{mins:02d}") - fmt = fmt.replace("%s", "{secs:02d}") - fmt = fmt.replace("%i", "{millis:03d}") - try: - i = fmt.index("%") - raise TimeFormatError( - "Unknow time format directive '{0}'".format(fmt[i : i + 2]) - ) - except ValueError: - pass - - def fromatter(seconds): - millis = int(seconds * 1000) - hrs, millis = divmod(millis, 3600000) - mins, millis = divmod(millis, 60000) - secs, millis = divmod(millis, 1000) - return fmt.format(hrs=hrs, mins=mins, secs=secs, millis=millis) - - return fromatter - - def make_logger(stderr=False, file=None, name=_AUDITOK_LOGGER): if not stderr and file is None: return None @@ -118,7 +80,6 @@ def initialize_workers(logger=None, **kwargs): observers = [] - reader = AudioDataSource(source=kwargs["input"], **kwargs) if kwargs["save_stream"] is not None: reader = workers.StreamSaverWorker(
--- a/auditok/io.py Sun Nov 10 20:20:51 2019 +0100 +++ b/auditok/io.py Mon Nov 11 09:38:47 2019 +0100 @@ -586,7 +586,7 @@ chunk_gen, total=nb_chunks, duration=duration, - **progress_bar_kwargs, + **progress_bar_kwargs ) if self.stream.is_stopped(): self.stream.start_stream() @@ -667,7 +667,7 @@ return PyAudioSource( *_get_audio_parameters(kwargs), frames_per_buffer=frames_per_buffer, - input_device_index=input_device_index, + input_device_index=input_device_index )
--- a/auditok/util.py Sun Nov 10 20:20:51 2019 +0100 +++ b/auditok/util.py Mon Nov 11 09:38:47 2019 +0100 @@ -22,7 +22,11 @@ PyAudioSource, get_audio_source, ) -from .exceptions import DuplicateArgument, TooSamllBlockDuration +from .exceptions import ( + DuplicateArgument, + TooSamllBlockDuration, + TimeFormatError, +) try: from . import signal_numpy as signal @@ -31,6 +35,7 @@ __all__ = [ + "make_duration_formatter", "DataSource", "DataValidator", "StringDataSource", @@ -42,6 +47,43 @@ ] +def make_duration_formatter(fmt): + """ + Accepted format directives: %i %s %m %h + """ + if fmt == "%S": + + def fromatter(seconds): + return "{:.3f}".format(seconds) + + elif fmt == "%I": + + def fromatter(seconds): + return "{0}".format(int(seconds * 1000)) + + else: + fmt = fmt.replace("%h", "{hrs:02d}") + fmt = fmt.replace("%m", "{mins:02d}") + fmt = fmt.replace("%s", "{secs:02d}") + fmt = fmt.replace("%i", "{millis:03d}") + try: + i = fmt.index("%") + raise TimeFormatError( + "Unknow time format directive '{0}'".format(fmt[i : i + 2]) + ) + except ValueError: + pass + + def fromatter(seconds): + millis = int(seconds * 1000) + hrs, millis = divmod(millis, 3600000) + mins, millis = divmod(millis, 60000) + secs, millis = divmod(millis, 1000) + return fmt.format(hrs=hrs, mins=mins, secs=secs, millis=millis) + + return fromatter + + def make_channel_selector(sample_width, channels, selected=None): fmt = signal.FORMAT.get(sample_width) if fmt is None:
--- a/auditok/workers.py Sun Nov 10 20:20:51 2019 +0100 +++ b/auditok/workers.py Mon Nov 11 09:38:47 2019 +0100 @@ -9,9 +9,8 @@ import subprocess from queue import Queue, Empty from .io import _guess_audio_format -from .util import AudioDataSource +from .util import AudioDataSource, make_duration_formatter from .core import split -from . import cmdline_util from .exceptions import ( EndOfProcessing, AudioEncodingError, @@ -409,7 +408,7 @@ ): self._print_format = print_format - self._format_time = cmdline_util.make_duration_formatter(time_format) + self._format_time = make_duration_formatter(time_format) self._timestamp_format = timestamp_format self.detections = [] Worker.__init__(self, timeout=timeout)
--- a/tests/test_cmdline_util.py Sun Nov 10 20:20:51 2019 +0100 +++ b/tests/test_cmdline_util.py Mon Nov 11 09:38:47 2019 +0100 @@ -9,7 +9,6 @@ from auditok.cmdline_util import ( _AUDITOK_LOGGER, make_kwargs, - make_duration_formatter, make_logger, initialize_workers, KeywordArguments, @@ -21,7 +20,6 @@ CommandLineWorker, PrintWorker, ) -from auditok.exceptions import TimeFormatError _ArgsNamespece = namedtuple( "_ArgsNamespece", @@ -173,38 +171,6 @@ kwargs = make_kwargs(args_ns) self.assertEqual(kwargs, expected) - @genty_dataset( - only_seconds=("%S", 5400, "5400.000"), - only_millis=("%I", 5400, "5400000"), - full=("%h:%m:%s.%i", 3725.365, "01:02:05.365"), - full_zero_hours=("%h:%m:%s.%i", 1925.075, "00:32:05.075"), - full_zero_minutes=("%h:%m:%s.%i", 3659.075, "01:00:59.075"), - full_zero_seconds=("%h:%m:%s.%i", 3720.075, "01:02:00.075"), - full_zero_millis=("%h:%m:%s.%i", 3725, "01:02:05.000"), - duplicate_directive=( - "%h %h:%m:%s.%i %s", - 3725.365, - "01 01:02:05.365 05", - ), - no_millis=("%h:%m:%s", 3725, "01:02:05"), - no_seconds=("%h:%m", 3725, "01:02"), - no_minutes=("%h", 3725, "01"), - no_hours=("%m:%s.%i", 3725, "02:05.000"), - ) - def test_make_duration_formatter(self, fmt, duration, expected): - formatter = make_duration_formatter(fmt) - result = formatter(duration) - self.assertEqual(result, expected) - - @genty_dataset( - duplicate_only_seconds=("%S %S",), - duplicate_only_millis=("%I %I",), - unknown_directive=("%x",), - ) - def test_make_duration_formatter_error(self, fmt): - with self.assertRaises(TimeFormatError): - make_duration_formatter(fmt) - def test_make_logger_stderr_and_file(self): with TemporaryDirectory() as tmpdir: file = os.path.join(tmpdir, "file.log")
--- a/tests/test_util.py Sun Nov 10 20:20:51 2019 +0100 +++ b/tests/test_util.py Mon Nov 11 09:38:47 2019 +0100 @@ -3,8 +3,9 @@ import math from array import array from genty import genty, genty_dataset -from auditok.util import AudioEnergyValidator +from auditok.util import AudioEnergyValidator, make_duration_formatter from auditok.signal import FORMAT +from auditok.exceptions import TimeFormatError def _sample_generator(*data_buffers): @@ -56,6 +57,41 @@ @genty +class TestFunctions(TestCase): + @genty_dataset( + only_seconds=("%S", 5400, "5400.000"), + only_millis=("%I", 5400, "5400000"), + full=("%h:%m:%s.%i", 3725.365, "01:02:05.365"), + full_zero_hours=("%h:%m:%s.%i", 1925.075, "00:32:05.075"), + full_zero_minutes=("%h:%m:%s.%i", 3659.075, "01:00:59.075"), + full_zero_seconds=("%h:%m:%s.%i", 3720.075, "01:02:00.075"), + full_zero_millis=("%h:%m:%s.%i", 3725, "01:02:05.000"), + duplicate_directive=( + "%h %h:%m:%s.%i %s", + 3725.365, + "01 01:02:05.365 05", + ), + no_millis=("%h:%m:%s", 3725, "01:02:05"), + no_seconds=("%h:%m", 3725, "01:02"), + no_minutes=("%h", 3725, "01"), + no_hours=("%m:%s.%i", 3725, "02:05.000"), + ) + def test_make_duration_formatter(self, fmt, duration, expected): + formatter = make_duration_formatter(fmt) + result = formatter(duration) + self.assertEqual(result, expected) + + @genty_dataset( + duplicate_only_seconds=("%S %S",), + duplicate_only_millis=("%I %I",), + unknown_directive=("%x",), + ) + def test_make_duration_formatter_error(self, fmt): + with self.assertRaises(TimeFormatError): + make_duration_formatter(fmt) + + +@genty class TestAudioEnergyValidator(TestCase): @genty_dataset( mono_valid_uc_None=([350, 400], 1, None, True),
--- a/tests/test_workers.py Sun Nov 10 20:20:51 2019 +0100 +++ b/tests/test_workers.py Mon Nov 11 09:38:47 2019 +0100 @@ -243,7 +243,7 @@ expected_print_calls = [ call( "[{}] {:.3f} {:.3f}, dur: {:.3f}".format( - i, *exp, exp[1] - exp[0] + i, exp[0], exp[1], exp[1] - exp[0] ) ) for i, exp in enumerate(self.expected, 1)