diff --git a/mapillary_tools/utils.py b/mapillary_tools/utils.py index b363e70b..de8996a4 100644 --- a/mapillary_tools/utils.py +++ b/mapillary_tools/utils.py @@ -6,12 +6,15 @@ from __future__ import annotations import concurrent.futures +import glob import hashlib import logging import os import typing as T from pathlib import Path +from . import exceptions + # Use "hashlib._Hash" instead of hashlib._Hash because: # AttributeError: module 'hashlib' has no attribute '_Hash' @@ -130,6 +133,36 @@ def find_all_image_samples( return image_samples_by_video_path +def expand_import_paths( + import_paths: T.Iterable[Path], + *, + predicate: T.Callable[[Path], bool] | None = None, + missing: str | None = None, +) -> list[Path]: + """Keep existing files/dirs; otherwise expand each path as a glob pattern.""" + out: list[Path] = [] + for path in import_paths: + if path.is_file() or path.is_dir(): + if predicate is None or predicate(path): + out.append(path) + continue + pattern = os.fspath(path) + matches = [ + Path(p) + for p in glob.glob(pattern, recursive="**" in pattern) + if predicate is None or predicate(Path(p)) + ] + if predicate is None: + matches = [p for p in matches if p.is_file() or p.is_dir()] + matches.sort(key=lambda p: p.name.lower()) + if not matches: + raise exceptions.MapillaryFileNotFoundError( + missing or f"Import file or directory not found: {path}" + ) + out.extend(matches) + return out + + def deduplicate_paths(paths: T.Iterable[Path]) -> T.Generator[Path, None, None]: resolved_paths: set[Path] = set() for p in paths: diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 15b54e81..c2a2547b 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -126,6 +126,17 @@ def test_filter_all(tmpdir: py.path.local): ) +def test_expand_import_paths_glob(tmp_path: Path): + (tmp_path / "GS100130.360").mkdir() + (tmp_path / "GS110130.360").mkdir() + (tmp_path / "GS090130.360").mkdir() + (tmp_path / "skip.txt").write_text("x") + matches = utils.expand_import_paths([tmp_path / "GS1?0130.360"]) + assert [p.name for p in matches] == ["GS100130.360", "GS110130.360"] + one = utils.expand_import_paths([tmp_path / "GS090130.360"]) + assert [p.name for p in one] == ["GS090130.360"] + + class TestSanitizeSerial: """Tests for sanitize_serial function"""