Match samples to videos of the same name by their paths

mapillary_tools/geotag/geotag_images_from_exiftool.py

@@ -20,7 +20,7 @@
 from ..exiftool_runner import ExiftoolRunner
 from . import options
 from .base import GeotagImagesFromGeneric
-from .geotag_images_from_video import GeotagImagesFromVideo
+from .geotag_images_from_video import describe_ambiguous_sample, GeotagImagesFromVideo
 from .geotag_videos_from_exiftool import GeotagVideosFromExifToolXML
 from .image_extractors.exiftool import ImageExifToolExtractor
 
@@ -172,7 +172,9 @@ def _geotag_samples_of(
         source_path: options.SourcePathOption,
     ) -> list[types.ImageMetadataOrError]:
         # Find all video paths that have sample images
-        samples_by_video = utils.find_all_image_samples(image_paths, video_paths)
+        samples_by_video, ambiguous_samples = utils.find_image_samples(
+            image_paths, video_paths
+        )
 
         video_metadata_or_errors = GeotagVideosFromExifToolXML(
             source_path, num_processes=self.num_processes
@@ -182,9 +184,13 @@ def _geotag_samples_of(
             video_metadata_or_errors,
             offset_time=self.offset_time,
             num_processes=self.num_processes,
+            samples_by_video=samples_by_video,
         ).to_description(sample_paths)
 
-        return sample_metadata_or_errors
+        return sample_metadata_or_errors + [
+            describe_ambiguous_sample(image_path, candidates)
+            for image_path, candidates in ambiguous_samples.items()
+        ]
 
     @override
     def to_description(

mapillary_tools/geotag/geotag_images_from_video.py

@@ -32,10 +32,15 @@ def __init__(
         video_metadatas: T.Sequence[types.VideoMetadataOrError],
         offset_time: float = 0.0,
         num_processes: int | None = None,
+        # The samples of each video, as found among all the videos the samples
+        # could be of (see utils.find_image_samples), or None to find them
+        # among the videos of video_metadatas
+        samples_by_video: T.Mapping[Path, T.Sequence[Path]] | None = None,
     ):
         super().__init__(num_processes=num_processes)
         self.video_metadatas = video_metadatas
         self.offset_time = offset_time
+        self.samples_by_video = samples_by_video
 
     @override
     def to_description(
@@ -48,9 +53,25 @@ def to_description(
             self.video_metadatas
         )
 
+        # Found once among all the videos: a sample of one of several videos
+        # of the same name is a sample of that one only
+        if self.samples_by_video is None:
+            samples_by_video: T.Mapping[Path, T.Sequence[Path]] = (
+                utils.find_all_image_samples(
+                    image_paths,
+                    [metadata.filename for metadata in self.video_metadatas],
+                )
+            )
+        else:
+            given_paths = set(image_paths)
+            samples_by_video = {
+                video_path: [path for path in paths if path in given_paths]
+                for video_path, paths in self.samples_by_video.items()
+            }
+
         for video_error_metadata in video_error_metadatas:
             video_path = video_error_metadata.filename
-            sample_paths = list(utils.filter_video_samples(image_paths, video_path))
+            sample_paths = list(samples_by_video.get(video_path, []))
             LOG.debug(
                 "Found %d sample images from video %s with error: %s",
                 len(sample_paths),
@@ -68,7 +89,7 @@ def to_description(
         for video_metadata in video_metadatas:
             video_path = video_metadata.filename
 
-            sample_paths = list(utils.filter_video_samples(image_paths, video_path))
+            sample_paths = list(samples_by_video.get(video_path, []))
             LOG.debug(
                 "Found %d sample images from video %s",
                 len(sample_paths),
@@ -138,7 +159,7 @@ def to_description(
         self, image_paths: T.Sequence[Path]
     ) -> list[types.ImageMetadataOrError]:
         video_paths = utils.find_videos([self.source_path])
-        image_samples_by_video_path = utils.find_all_image_samples(
+        image_samples_by_video_path, ambiguous_samples = utils.find_image_samples(
             image_paths, video_paths
         )
         video_paths_with_image_samples = list(image_samples_by_video_path.keys())
@@ -150,6 +171,7 @@ def to_description(
             video_metadatas,
             offset_time=self.offset_time,
             num_processes=self.num_processes,
+            samples_by_video=image_samples_by_video_path,
         )
         sample_metadatas = geotag.to_description(image_paths)
 
@@ -158,7 +180,9 @@ def to_description(
         # that can geotag it (e.g. exif after this one) may still do
         sample_paths = {metadata.filename for metadata in sample_metadatas}
         return sample_metadatas + [
-            types.describe_error_metadata(
+            describe_ambiguous_sample(image_path, ambiguous_samples[image_path])
+            if image_path in ambiguous_samples
+            else types.describe_error_metadata(
                 exceptions.MapillaryGeoTaggingError(
                     f"Not a frame of any video in {self.source_path}"
                 ),
@@ -168,3 +192,26 @@ def to_description(
             for image_path in image_paths
             if image_path not in sample_paths
         ]
+
+
+def describe_ambiguous_sample(
+    image_path: Path, video_paths: T.Sequence[Path]
+) -> types.ErrorMetadata:
+    """
+    Describe a sample that could be of any of the videos of the same name
+    **video_paths**, as its folder does not tell which (see
+    utils.find_image_samples), rather than place it along one of them
+
+    >>> describe_ambiguous_sample(
+    ...     Path("clip.mp4/clip_0_000001.jpg"), [Path("x/clip.mp4"), Path("y/clip.mp4")]
+    ... ).error
+    MapillaryGeoTaggingError('Cannot tell which of the 2 videos named clip.mp4 this is a frame of')
+    """
+    return types.describe_error_metadata(
+        exceptions.MapillaryGeoTaggingError(
+            f"Cannot tell which of the {len(video_paths)} videos named "
+            f"{video_paths[0].name} this is a frame of"
+        ),
+        image_path,
+        filetype=types.FileType.IMAGE,
+    )

mapillary_tools/utils.py

@@ -111,23 +111,109 @@ def filter_video_samples(
 def find_all_image_samples(
     image_paths: T.Iterable[Path], video_paths: T.Iterable[Path]
 ) -> dict[Path, list[Path]]:
-    # TODO: not work with the same filenames, e.g. foo/hello.mp4 and bar/hello.mp4
-    video_basenames = {path.name: path for path in video_paths}
+    """
+    The samples of each video among the images, but those that could be of
+    several videos of the same name (see find_image_samples)
+    """
+    samples_by_video, _ = find_image_samples(image_paths, video_paths)
+    return samples_by_video
+
+
+def find_image_samples(
+    image_paths: T.Iterable[Path], video_paths: T.Iterable[Path]
+) -> tuple[dict[Path, list[Path]], dict[Path, list[Path]]]:
+    """
+    Find the samples of each video among the images. sample_video() names the
+    samples of a video after it, in a folder named after it, at the path of the
+    video relative to the folder of videos it samples (e.g. the samples of
+    VIDEOS/x/clip.mp4 are OUT/x/clip.mp4/clip_0_000001.jpg, ...).
+
+    Videos of the same name in different folders are told apart by their paths
+    relative to the deepest folder that holds them all: the folder of a sample
+    ends with the path of its video, the longest of them if several match. A
+    sample whose folder ends with none of them, which could be of any of them,
+    is a sample of none.
+
+    Returns the samples of each video, and the videos that each sample that
+    could be of several videos could be of.
+
+    >>> images = [
+    ...     Path("OUT/x/clip.mp4/clip_0_000001.jpg"),
+    ...     Path("OUT/y/clip.mp4/clip_0_000001.jpg"),
+    ...     Path("OUT/clip.mp4/clip_0_000001.jpg"),
+    ...     Path("OUT/x/clip.mp4/other.jpg"),
+    ... ]
+    >>> videos = [Path("VIDEOS/x/clip.mp4"), Path("VIDEOS/y/clip.mp4")]
+    >>> samples, ambiguous = find_image_samples(images, videos)
+    >>> {video.as_posix(): [p.as_posix() for p in paths] for video, paths in samples.items()}
+    {'VIDEOS/x/clip.mp4': ['OUT/x/clip.mp4/clip_0_000001.jpg'], 'VIDEOS/y/clip.mp4': ['OUT/y/clip.mp4/clip_0_000001.jpg']}
+    >>> {image.as_posix(): [p.as_posix() for p in paths] for image, paths in ambiguous.items()}
+    {'OUT/clip.mp4/clip_0_000001.jpg': ['VIDEOS/x/clip.mp4', 'VIDEOS/y/clip.mp4']}
+    """
+    videos = list(deduplicate_paths(video_paths))
+    resolved_videos = [_absolute(video) for video in videos]
+    try:
+        root: Path | None = (
+            Path(os.path.commonpath([video.parent for video in resolved_videos]))
+            if resolved_videos
+            else None
+        )
+    except ValueError:
+        # On different drives
+        root = None
+
+    # The videos of each name, with their paths relative to the root
+    videos_by_name: dict[str, list[tuple[Path, tuple[str, ...]]]] = {}
+    for video, resolved_video in zip(videos, resolved_videos):
+        if root is None:
+            relative_parts: tuple[str, ...] = (resolved_video.name,)
+        else:
+            relative_parts = resolved_video.relative_to(root).parts
+        videos_by_name.setdefault(resolved_video.name, []).append(
+            (video, relative_parts)
+        )
 
-    image_samples_by_video_path: dict[Path, list[Path]] = {}
+    samples_by_video: dict[Path, list[Path]] = {}
+    ambiguous: dict[Path, list[Path]] = {}
     for image_path in image_paths:
-        # If you want to walk an arbitrary filesystem path upwards,
-        # it is recommended to first call Path.resolve() so as to resolve symlinks and eliminate “..” components.
-        image_dirname = image_path.resolve().parent.name
-        if image_dirname in video_basenames:
-            root, _ = os.path.splitext(image_dirname)
-            if image_path.name.startswith(root + "_"):
-                video_path = video_basenames[image_dirname]
-                image_samples_by_video_path.setdefault(video_path, []).append(
-                    image_path
-                )
-
-    return image_samples_by_video_path
+        # Resolved first, to walk the path upwards: the folder of a sample is
+        # named after its video
+        sample_dir = _absolute(image_path).parent
+        candidates = videos_by_name.get(sample_dir.name, [])
+        stem, _ = os.path.splitext(sample_dir.name)
+        if not candidates or not image_path.name.startswith(stem + "_"):
+            continue
+
+        matching = [
+            (video, parts)
+            for video, parts in candidates
+            if sample_dir.parts[-len(parts) :] == parts
+        ]
+        if matching:
+            longest = max(len(parts) for _, parts in matching)
+            best = [video for video, parts in matching if len(parts) == longest]
+            if 1 < len(best):
+                # Told apart by no path, as videos on different drives
+                ambiguous[image_path] = best
+                continue
+            video = best[0]
+        elif len(candidates) == 1:
+            video, _ = candidates[0]
+        else:
+            ambiguous[image_path] = [video for video, _ in candidates]
+            continue
+        samples_by_video.setdefault(video, []).append(image_path)
+
+    return samples_by_video, ambiguous
+
+
+def _absolute(path: Path) -> Path:
+    """
+    The absolute path of **path**, with its symbolic links resolved where it
+    exists. Path.resolve() alone leaves a path that does not exist relative on
+    Windows with Python 3.9 and 3.10
+    """
+    return Path(os.path.abspath(path.resolve()))
 
 
 def deduplicate_paths(paths: T.Iterable[Path]) -> T.Generator[Path, None, None]: