|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from collections.abc import Callable |
| 4 | +from collections.abc import Sequence |
1 | 5 | import errno |
2 | 6 | import inspect |
3 | 7 | import itertools |
4 | 8 | import os |
5 | | -import typing |
6 | 9 |
|
7 | 10 | import numpy as np |
8 | 11 | import pandas as pd |
@@ -155,20 +158,20 @@ class Process: |
155 | 158 | def __init__( |
156 | 159 | self, |
157 | 160 | *, |
158 | | - process_func: typing.Callable[..., typing.Any] = None, |
159 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 161 | + process_func: Callable[..., object] | None = None, |
| 162 | + process_func_args: dict[str, object] | None = None, |
160 | 163 | process_func_is_mono: bool = False, |
161 | | - sampling_rate: int = None, |
| 164 | + sampling_rate: int | None = None, |
162 | 165 | resample: bool = False, |
163 | | - channels: typing.Union[int, typing.Sequence[int]] = None, |
| 166 | + channels: int | Sequence[int] | None = None, |
164 | 167 | mixdown: bool = False, |
165 | | - win_dur: Timestamp = None, |
166 | | - hop_dur: Timestamp = None, |
167 | | - min_signal_dur: Timestamp = None, |
168 | | - max_signal_dur: Timestamp = None, |
169 | | - segment: Segment = None, |
| 168 | + win_dur: Timestamp | None = None, |
| 169 | + hop_dur: Timestamp | None = None, |
| 170 | + min_signal_dur: Timestamp | None = None, |
| 171 | + max_signal_dur: Timestamp | None = None, |
| 172 | + segment: Segment | None = None, |
170 | 173 | keep_nat: bool = False, |
171 | | - num_workers: typing.Optional[int] = 1, |
| 174 | + num_workers: int | None = 1, |
172 | 175 | multiprocessing: bool = False, |
173 | 176 | verbose: bool = False, |
174 | 177 | ): |
@@ -241,15 +244,15 @@ def _process_file( |
241 | 244 | file: str, |
242 | 245 | *, |
243 | 246 | idx: int = 0, |
244 | | - root: str = None, |
245 | | - start: pd.Timedelta = None, |
246 | | - end: pd.Timedelta = None, |
247 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
248 | | - ) -> typing.Tuple[ |
249 | | - typing.List[typing.Any], |
250 | | - typing.List[str], |
251 | | - typing.List[pd.Timedelta], |
252 | | - typing.List[pd.Timedelta], |
| 247 | + root: str | None = None, |
| 248 | + start: pd.Timedelta | None = None, |
| 249 | + end: pd.Timedelta | None = None, |
| 250 | + process_func_args: dict[str, object] | None = None, |
| 251 | + ) -> tuple[ |
| 252 | + list[object], |
| 253 | + list[str], |
| 254 | + list[pd.Timedelta], |
| 255 | + list[pd.Timedelta], |
253 | 256 | ]: |
254 | 257 | if start is not None: |
255 | 258 | start = utils.to_timedelta(start, self.sampling_rate) |
@@ -300,10 +303,10 @@ def process_file( |
300 | 303 | self, |
301 | 304 | file: str, |
302 | 305 | *, |
303 | | - start: Timestamp = None, |
304 | | - end: Timestamp = None, |
305 | | - root: str = None, |
306 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 306 | + start: Timestamp | None = None, |
| 307 | + end: Timestamp | None = None, |
| 308 | + root: str | None = None, |
| 309 | + process_func_args: dict[str, object] | None = None, |
307 | 310 | ) -> pd.Series: |
308 | 311 | r"""Process the content of an audio file. |
309 | 312 |
|
@@ -358,12 +361,12 @@ def process_file( |
358 | 361 |
|
359 | 362 | def process_files( |
360 | 363 | self, |
361 | | - files: typing.Sequence[str], |
| 364 | + files: Sequence[str], |
362 | 365 | *, |
363 | | - starts: Timestamps = None, |
364 | | - ends: Timestamps = None, |
365 | | - root: str = None, |
366 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 366 | + starts: Timestamps | None = None, |
| 367 | + ends: Timestamps | None = None, |
| 368 | + root: str | None = None, |
| 369 | + process_func_args: dict[str, object] | None = None, |
367 | 370 | ) -> pd.Series: |
368 | 371 | r"""Process a list of files. |
369 | 372 |
|
@@ -460,7 +463,7 @@ def process_folder( |
460 | 463 | *, |
461 | 464 | filetype: str = "wav", |
462 | 465 | include_root: bool = True, |
463 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 466 | + process_func_args: dict[str, object] | None = None, |
464 | 467 | ) -> pd.Series: |
465 | 468 | r"""Process files in a folder. |
466 | 469 |
|
@@ -512,8 +515,8 @@ def process_folder( |
512 | 515 | def _process_index_wo_segment( |
513 | 516 | self, |
514 | 517 | index: pd.Index, |
515 | | - root: typing.Optional[str], |
516 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 518 | + root: str | None, |
| 519 | + process_func_args: dict[str, object] | None = None, |
517 | 520 | ) -> pd.Series: |
518 | 521 | r"""Like process_index, but does not apply segmentation.""" |
519 | 522 | if index.empty: |
@@ -558,9 +561,9 @@ def process_index( |
558 | 561 | index: pd.Index, |
559 | 562 | *, |
560 | 563 | preserve_index: bool = False, |
561 | | - root: str = None, |
562 | | - cache_root: str = None, |
563 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 564 | + root: str | None = None, |
| 565 | + cache_root: str | None = None, |
| 566 | + process_func_args: dict[str, object] | None = None, |
564 | 567 | ) -> pd.Series: |
565 | 568 | r"""Process from an index conform to audformat_. |
566 | 569 |
|
@@ -638,16 +641,16 @@ def _process_signal( |
638 | 641 | sampling_rate: int, |
639 | 642 | *, |
640 | 643 | idx: int = 0, |
641 | | - root: str = None, |
642 | | - file: str = None, |
643 | | - start: pd.Timedelta = None, |
644 | | - end: pd.Timedelta = None, |
645 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
646 | | - ) -> typing.Tuple[ |
647 | | - typing.List[typing.Any], |
648 | | - typing.List[str], |
649 | | - typing.List[pd.Timedelta], |
650 | | - typing.List[pd.Timedelta], |
| 644 | + root: str | None = None, |
| 645 | + file: str | None = None, |
| 646 | + start: pd.Timedelta | None = None, |
| 647 | + end: pd.Timedelta | None = None, |
| 648 | + process_func_args: dict[str, object] | None = None, |
| 649 | + ) -> tuple[ |
| 650 | + list[object], |
| 651 | + list[str], |
| 652 | + list[pd.Timedelta], |
| 653 | + list[pd.Timedelta], |
651 | 654 | ]: |
652 | 655 | signal = np.atleast_2d(signal) |
653 | 656 |
|
@@ -718,10 +721,10 @@ def process_signal( |
718 | 721 | signal: np.ndarray, |
719 | 722 | sampling_rate: int, |
720 | 723 | *, |
721 | | - file: str = None, |
722 | | - start: Timestamp = None, |
723 | | - end: Timestamp = None, |
724 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 724 | + file: str | None = None, |
| 725 | + start: Timestamp | None = None, |
| 726 | + end: Timestamp | None = None, |
| 727 | + process_func_args: dict[str, object] | None = None, |
725 | 728 | ) -> pd.Series: |
726 | 729 | r"""Process audio signal and return result. |
727 | 730 |
|
@@ -799,7 +802,7 @@ def _process_signal_from_index_wo_segment( |
799 | 802 | signal: np.ndarray, |
800 | 803 | sampling_rate: int, |
801 | 804 | index: pd.Index, |
802 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 805 | + process_func_args: dict[str, object] | None = None, |
803 | 806 | ) -> pd.Series: |
804 | 807 | r"""Like process_signal_from_index, but does not apply segmentation.""" |
805 | 808 | if index.empty: |
@@ -865,7 +868,7 @@ def process_signal_from_index( |
865 | 868 | signal: np.ndarray, |
866 | 869 | sampling_rate: int, |
867 | 870 | index: pd.Index, |
868 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
| 871 | + process_func_args: dict[str, object] | None = None, |
869 | 872 | ) -> pd.Series: |
870 | 873 | r"""Split a signal into segments and process each segment. |
871 | 874 |
|
@@ -919,10 +922,10 @@ def _call( |
919 | 922 | sampling_rate: int, |
920 | 923 | *, |
921 | 924 | idx: int = 0, |
922 | | - root: str = None, |
923 | | - file: str = None, |
924 | | - process_func_args: typing.Dict[str, typing.Any] = None, |
925 | | - ) -> typing.Any: |
| 925 | + root: str | None = None, |
| 926 | + file: str | None = None, |
| 927 | + process_func_args: dict[str, object] | None = None, |
| 928 | + ) -> object: |
926 | 929 | r"""Call processing function, possibly pass special args.""" |
927 | 930 | signal, sampling_rate = utils.preprocess_signal( |
928 | 931 | signal, |
@@ -980,7 +983,7 @@ def __call__( |
980 | 983 | self, |
981 | 984 | signal: np.ndarray, |
982 | 985 | sampling_rate: int, |
983 | | - ) -> typing.Any: |
| 986 | + ) -> object: |
984 | 987 | r"""Apply processing to signal. |
985 | 988 |
|
986 | 989 | This function processes the signal **without** transforming the output |
|
0 commit comments