Add type annotations and mypy

Author: unmade

.gitignore

@@ -1,11 +1,14 @@
 *.egg-info
 pip-wheel-metadata
 dist
+build
 
 *coverage*
 !.coveragerc
 
 __pycache__
 .pytest_cache
 
+.mypy_cache
+
 .tox

.pre-commit-config.yaml

@@ -23,6 +23,11 @@ repos:
             '-e', 'git+https://github.com/pycqa/pyflakes@master#egg=pyflakes',
             '-e', 'git+https://github.com/pycqa/pycodestyle@master#egg=pycodestyle',
         ]
+-   repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.770
+    hooks:
+    -   id: mypy
+        exclude: 'tests/data/'
 -   repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v2.5.0
     hooks:

mypy.ini

@@ -0,0 +1,9 @@
+[mypy]
+mypy_path=src
+
+warn_return_any = True
+warn_redundant_casts = True
+warn_unused_configs = True
+
+[mypy-pytest.*]
+ignore_missing_imports = True

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "audiomatch"
-version = "0.2.0"
+version = "0.2.1"
 description = "A small command-line tool to find similar audio files"
 keywords = ["duplicate", "detection", "audio", "fingerprinting", "command-line"]
 readme = "README.rst"

src/audiomatch/files.py

@@ -41,10 +41,11 @@ def pair(
         )
 
     if len(files) == 1 and len(files[0]) > 1:
-        return itertools.combinations(*files, 2)
+        return itertools.combinations(*files, 2)  # type: ignore
     elif len(files) > 1:
         return itertools.chain.from_iterable(
-            itertools.product(*group) for group in itertools.combinations(files, 2)
+            itertools.product(*group)  # type: ignore
+            for group in itertools.combinations(files, 2)
         )
     else:
         raise NotEnoughFiles("Not enough input files.")

src/audiomatch/fingerprints.py

@@ -1,5 +1,9 @@
+from __future__ import annotations
+
 import statistics
 import subprocess
+from pathlib import Path
+from typing import Iterator, List, Tuple, Union
 
 from audiomatch import popcount
 
@@ -8,7 +12,7 @@
 SCORE_MEDIAN_DELTA = 0.04
 
 
-def calc(path, length=120):
+def calc(path: Path, length: int = 120) -> List[int]:
     # TODO: Probably it would be better to parse json output
     fp = subprocess.run(
         ["fpcalc", "-rate", "11025", "-raw", "-length", str(length), str(path)],
@@ -19,7 +23,7 @@ def calc(path, length=120):
     return []
 
 
-def compare(fp1, fp2):
+def compare(fp1: List[int], fp2: List[int]) -> float:
     # Take first 30 seconds of the the shortest fingerprint and try to find it in a
     # longer one
     if len(fp1) > len(fp2):
@@ -28,7 +32,7 @@ def compare(fp1, fp2):
         return find_best_score(fp2, fp1[: seconds(30)])
 
 
-def find_best_score(fp1, fp2):
+def find_best_score(fp1: List[int], fp2: List[int]) -> float:
     # Fingerprints lesser than 10 seconds don't have enough data for analysis
     if len(fp1) > seconds(10) and len(fp2) > seconds(10):
         results = [correlation(_fp1, _fp2) for _fp1, _fp2 in cross(fp1, fp2)]
@@ -58,12 +62,12 @@ def find_best_score(fp1, fp2):
     return 0.0
 
 
-def correlation(fp1, fp2):
+def correlation(fp1: List[int], fp2: List[int]) -> float:
     error = sum(popcount.popcount(x ^ y) for x, y in zip(fp1, fp2))
     return 1.0 - error / 32.0 / min(len(fp1), len(fp2))
 
 
-def cross(fp1, fp2):
+def cross(fp1: List[int], fp2: List[int]) -> Iterator[Tuple[List[int], List[int]]]:
     length = min(len(fp1), len(fp2))
     span = min(length // 4, seconds(5))
     limit = max(len(fp1), len(fp2)) - length - span
@@ -75,5 +79,5 @@ def cross(fp1, fp2):
         yield fp1[offset:], fp2
 
 
-def seconds(x) -> int:
+def seconds(x: Union[int, float]) -> int:
     return round(x * 7)

src/audiomatch/match.py

@@ -4,7 +4,7 @@
 import functools
 import itertools
 from pathlib import Path
-from typing import Dict, Iterable, Optional, Tuple
+from typing import Dict, Iterable, List, Optional, Tuple
 
 from audiomatch import files, fingerprints
 from audiomatch.constants import DEFAULT_LENGTH
@@ -36,12 +36,12 @@ def match(
     # Using multiprocessing.Pool.starmap method we can avoid writing wrapper to unpack
     # arguments. However, multiprocessing.Pool doesn't play nicely with coverage, and
     # require to explicitly call 'pool.join'
-    with concurrent.futures.ProcessPoolExecutor() as executor:
-        scores = executor.map(_compare, ((fps[a], fps[b]) for a, b in pairs))
+    with concurrent.futures.ProcessPoolExecutor() as pool:
+        scores = pool.map(_compare, ((fps[a], fps[b]) for a, b in pairs))
 
     return dict(zip(pairs, scores))
 
 
-def _compare(pair):
+def _compare(pair: Tuple[List[int], List[int]]) -> float:
     """Just a wrapper for fingerprints.compare, that unpack its first argument"""
     return fingerprints.compare(*pair)

src/audiomatch/popcount/__init__.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+from typing import List
+
 try:
     from audiomatch.popcount._popcount import popcount
 except ImportError:
@@ -6,13 +10,13 @@
     #
     # This popcount version works slightly faster than 'bin(x).count("1")'
 
-    def _popcount_table(size):
+    def _popcount_table(size: int) -> List[int]:
         table = [0] * 2 ** size
         for i in range(len(table)):
             table[i] = (i & 1) + table[i >> 1]
         return table
 
     _POPCOUNT_TABLE16 = _popcount_table(16)
 
-    def popcount(x):
+    def popcount(x: int) -> int:
         return _POPCOUNT_TABLE16[x & 0xFFFF] + _POPCOUNT_TABLE16[(x >> 16) & 0xFFFF]

src/audiomatch/popcount/_popcount.pyi

@@ -0,0 +1,2 @@
+def popcount(x: int) -> int:
+    ...

tests/data/edgecase-1/__init__.py src/audiomatch/py.typed

@@ -1,7 +1,11 @@
+from __future__ import annotations
+
 import sys
+from pathlib import Path
+from typing import Dict, Tuple
 
 
-def console(matches) -> None:
+def console(matches: Dict[Tuple[Path, Path], float]) -> None:
     """Print similar audio to standard output"""
     similars = _join(_adjancency_list(matches))
     lines = ["\n".join(str(node) for node in similar) for similar in sorted(similars)]
@@ -29,7 +33,7 @@ def _join(graph):
 
 
 def _adjancency_list(matches, score=0.61):
-    """Returns an adjacency list for matches"""
+    """Returns an adjacency list for matches with a given score or higher"""
     graph = {}
     for pair, score in matches.items():
         if score > 0.61:

src/audiomatch/reports.py

@@ -6,23 +6,23 @@
 from audiomatch.exceptions import NotEnoughFiles
 
 SAMPLES_DIR = Path(__file__).parent.joinpath("data")
-SAMPLES_1 = [SAMPLES_DIR.joinpath(f"sample-1/take-{i}.log") for i in range(1, 4)]
-SAMPLES_2 = [SAMPLES_DIR.joinpath(f"sample-2/take-{i}.log") for i in range(1, 5)]
+SAMPLES_1 = [SAMPLES_DIR.joinpath(f"sample_1/take-{i}.log") for i in range(1, 4)]
+SAMPLES_2 = [SAMPLES_DIR.joinpath(f"sample_2/take-{i}.log") for i in range(1, 5)]
 
 
 def sort_sublist(pairs):
     return sorted(pair.__class__(sorted(pair)) for pair in pairs)
 
 
 def test_pair_two_files():
-    file1 = SAMPLES_DIR.joinpath("sample-1/take-1.log")
-    file2 = SAMPLES_DIR.joinpath("sample-1/take-2.log")
+    file1 = SAMPLES_DIR.joinpath("sample_1/take-1.log")
+    file2 = SAMPLES_DIR.joinpath("sample_1/take-2.log")
     pairs = files.pair(file1, file2)
     assert sort_sublist(pairs) == [(SAMPLES_1[0], SAMPLES_1[1])]
 
 
 def test_pair_files_in_a_directory():
-    directory = SAMPLES_DIR.joinpath("sample-1")
+    directory = SAMPLES_DIR.joinpath("sample_1")
     pairs = files.pair(directory, extensions=[".log"])
     assert sort_sublist(pairs) == [
         (SAMPLES_1[0], SAMPLES_1[1]),
@@ -32,8 +32,8 @@ def test_pair_files_in_a_directory():
 
 
 def test_pair_a_file_and_all_files_in_a_directory():
-    file = SAMPLES_DIR.joinpath("sample-1/take-1.log")
-    directory = SAMPLES_DIR.joinpath("sample-1")
+    file = SAMPLES_DIR.joinpath("sample_1/take-1.log")
+    directory = SAMPLES_DIR.joinpath("sample_1")
     pairs = files.pair(file, directory, extensions=[".log"])
     assert sort_sublist(pairs) == [
         (SAMPLES_1[0], SAMPLES_1[0]),
@@ -43,16 +43,16 @@ def test_pair_a_file_and_all_files_in_a_directory():
 
 
 def test_pair_directories():
-    directory1 = SAMPLES_DIR.joinpath("sample-1")
-    directory2 = SAMPLES_DIR.joinpath("sample-2")
+    directory1 = SAMPLES_DIR.joinpath("sample_1")
+    directory2 = SAMPLES_DIR.joinpath("sample_2")
     pairs = files.pair(directory1, directory2, extensions=[".log"])
     assert sort_sublist(pairs) == [
         (sample_1, sample_2) for sample_1 in SAMPLES_1 for sample_2 in SAMPLES_2
     ]
 
 
 def test_pair_glob():
-    wildcard = SAMPLES_DIR.joinpath("sample-1/*.log")
+    wildcard = SAMPLES_DIR.joinpath("sample_1/*.log")
     pairs = files.pair(wildcard)
     assert sort_sublist(pairs) == [
         (SAMPLES_1[0], SAMPLES_1[1]),
@@ -62,7 +62,7 @@ def test_pair_glob():
 
 
 def test_pair_one_file():
-    file = SAMPLES_DIR.joinpath("sample-1/take-1.log")
+    file = SAMPLES_DIR.joinpath("sample_1/take-1.log")
     with pytest.raises(NotEnoughFiles) as excinfo:
         files.pair(file)
     assert str(excinfo.value) == "Not enough input files."

tests/data/edgecase-2/__init__.py tests/data/edgecase_1/__init__.py

@@ -15,11 +15,11 @@ def _load_fp(filepath: str) -> List[int]:
 
 
 def test_calc():
-    fp = resources.read_binary("tests.data.sample-1", "take-1.log")
+    fp = resources.read_binary("tests.data.sample_1", "take-1.log")
     fpcalc = mock.MagicMock(spec=CompletedProcess, stdout=fp)
     with mock.patch("subprocess.run", return_value=fpcalc) as fpcalc_mock:
         fp = fingerprints.calc("/path/to/audio", length=120)
-    assert fp == _load_fp("sample-1/take-1.log")
+    assert fp == _load_fp("sample_1/take-1.log")
     assert fpcalc_mock.called
 
 
@@ -35,12 +35,12 @@ def test_calc_empty_fingerprint():
 @pytest.mark.parametrize(
     ["a", "b", "score"],
     [
-        ("sample-1/take-1.log", "sample-1/take-1.log", 1.00),
-        ("sample-1/take-1.log", "sample-1/take-2.log", 0.77),
-        ("sample-1/take-1.log", "sample-2/take-2.log", 0.00),
-        ("sample-1/take-3.log", "sample-2/take-4.log", 0.00),
-        ("sample-2/take-1.log", "sample-2/take-2.log", 0.66),
-        ("sample-2/take-4.log", "sample-2/take-1.log", 0.63),
+        ("sample_1/take-1.log", "sample_1/take-1.log", 1.00),
+        ("sample_1/take-1.log", "sample_1/take-2.log", 0.77),
+        ("sample_1/take-1.log", "sample_2/take-2.log", 0.00),
+        ("sample_1/take-3.log", "sample_2/take-4.log", 0.00),
+        ("sample_2/take-1.log", "sample_2/take-2.log", 0.66),
+        ("sample_2/take-4.log", "sample_2/take-1.log", 0.63),
     ],
 )
 def test_compare(a, b, score):
@@ -50,26 +50,26 @@ def test_compare(a, b, score):
 
 
 def test_compare_add_correction_for_short_fingerprints():
-    fp1 = _load_fp("edgecase-1/sample-1.log")
-    fp2 = _load_fp("edgecase-1/sample-2.log")
+    fp1 = _load_fp("edgecase_1/sample-1.log")
+    fp2 = _load_fp("edgecase_1/sample-2.log")
     assert fingerprints.compare(fp1, fp2) == 0.0
 
 
 def test_compare_returns_immediately_for_score_greater_than_confidence_score():
-    fp1 = _load_fp("edgecase-2/sample-1.log")
-    fp2 = _load_fp("edgecase-2/sample-2.log")
+    fp1 = _load_fp("edgecase_2/sample-1.log")
+    fp2 = _load_fp("edgecase_2/sample-2.log")
     assert fingerprints.compare(fp1, fp2) > fingerprints.CONFIDENCE_SCORE
 
 
 def test_compare_requires_fingerprints_to_be_at_least_10_seconds_long():
-    fp1 = _load_fp("edgecase-3/sample-1.log")
-    fp2 = _load_fp("edgecase-3/sample-2.log")
+    fp1 = _load_fp("edgecase_3/sample-1.log")
+    fp2 = _load_fp("edgecase_3/sample-2.log")
     assert fingerprints.compare(fp1, fp2) == 0.0
 
 
 def test_compare_false_positive():
     # This two completely different fingerprints initially have a relatively good score.
     # 'compare' handles these cases by checking difference between median and max value.
-    fp1 = _load_fp("edgecase-4/sample-1.log")
-    fp2 = _load_fp("edgecase-4/sample-2.log")
+    fp1 = _load_fp("edgecase_4/sample-1.log")
+    fp2 = _load_fp("edgecase_4/sample-2.log")
     assert fingerprints.compare(fp1, fp2) == 0.0

tests/data/edgecase-1/sample-1.log tests/data/edgecase_1/sample-1.log

@@ -25,8 +25,8 @@ def fpcalc(filepath: Path, length):
 
 @pytest.mark.slow
 def test_match():
-    sample_1 = SAMPLES_DIR.joinpath("sample-1/take-1.log")
-    sample_2 = SAMPLES_DIR.joinpath("sample-2/take-1.log")
+    sample_1 = SAMPLES_DIR.joinpath("sample_1/take-1.log")
+    sample_2 = SAMPLES_DIR.joinpath("sample_2/take-1.log")
     with mock.patch("audiomatch.fingerprints.calc", side_effect=fpcalc) as fpcalc_mock:
         matches = match.match(sample_1, sample_2, extensions=[".log"])
     assert sort_keys(matches) == {(sample_1, sample_2): 0.0}
@@ -35,7 +35,7 @@ def test_match():
 
 @pytest.mark.slow
 def test_match_with_empty_fingerprint():
-    sample_1 = SAMPLES_DIR.joinpath("sample-1/take-1.log")
+    sample_1 = SAMPLES_DIR.joinpath("sample_1/take-1.log")
     empty = SAMPLES_DIR.joinpath("empty.log")
     with mock.patch("audiomatch.fingerprints.calc", side_effect=fpcalc) as fpcalc_mock:
         matches = match.match(sample_1, empty, extensions=[".log"])