Sample a symlinked video into a folder named after the link

mapillary_tools/geotag/utils.py

@@ -71,8 +71,15 @@ def read_gpx_tracks(gpx_file: Path) -> list[Track]:
             try:
                 # Parsed from the file opened in text mode, which gpxpy reads
                 # whole itself, so that no other copy of its content is kept
-                # while it parses
-                parsed = _parse_gpx_file(io.TextIOWrapper(io.BufferedReader(reader)))
+                # while it parses. As UTF-8, which GPX files almost always
+                # are, on every platform rather than in the encoding of the
+                # locale: a byte that is not UTF-8 only turns a letter of a
+                # name into U+FFFD, as for NMEA
+                parsed = _parse_gpx_file(
+                    io.TextIOWrapper(
+                        io.BufferedReader(reader), encoding="utf-8", errors="replace"
+                    )
+                )
             except (gpxpy.gpx.GPXException, UnicodeError) as ex:
                 parsed = ex
             # By the content parsed, which a file changed since it was looked

mapillary_tools/sample_video.py

@@ -61,11 +61,13 @@ def _normalize_path(
         video_list = sorted(
             utils.find_videos([video_import_path], skip_subfolders=skip_subfolders)
         )
-        video_dir = video_import_path.resolve()
+        # Absolute with its symbolic links kept, as the paths of the videos
+        # in it are made (see _sample_dir)
+        video_dir = Path(os.path.abspath(video_import_path))
         LOG.debug("Found %d videos in %s", len(video_list), video_dir)
     elif video_import_path.is_file():
         video_list = [video_import_path]
-        video_dir = video_import_path.resolve().parent
+        video_dir = Path(os.path.abspath(video_import_path)).parent
     else:
         raise exceptions.MapillaryFileNotFoundError(
             f"Video file or directory not found: {video_import_path}"
@@ -89,8 +91,12 @@ def find_sample_dirs(
 
 
 def _sample_dir(import_path: Path, video_dir: Path, video_path: Path) -> Path:
-    # need to resolve video_path because video_dir might be absolute
-    return Path(import_path).joinpath(video_path.resolve().relative_to(video_dir))
+    # Absolute, as video_dir is, with its symbolic links kept: a link to a
+    # video outside video_dir is sampled into a folder named after the link,
+    # as its samples are
+    return Path(import_path).joinpath(
+        Path(os.path.abspath(video_path)).relative_to(video_dir)
+    )
 
 
 def xor(a: bool, b: bool):

mapillary_tools/utils.py

@@ -151,11 +151,13 @@ def find_image_samples(
     {'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]
+    # Absolute with their symbolic links kept, as sample_video() names the
+    # folders of the videos after the paths it was given
+    absolute_videos = [Path(os.path.abspath(video)) for video in videos]
     try:
         root: Path | None = (
-            Path(os.path.commonpath([video.parent for video in resolved_videos]))
-            if resolved_videos
+            Path(os.path.commonpath([video.parent for video in absolute_videos]))
+            if absolute_videos
             else None
         )
     except ValueError:
@@ -164,21 +166,21 @@ def find_image_samples(
 
     # 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):
+    for video, absolute_video in zip(videos, absolute_videos):
         if root is None:
-            relative_parts: tuple[str, ...] = (resolved_video.name,)
+            relative_parts: tuple[str, ...] = (absolute_video.name,)
         else:
-            relative_parts = resolved_video.relative_to(root).parts
-        videos_by_name.setdefault(resolved_video.name, []).append(
+            relative_parts = absolute_video.relative_to(root).parts
+        videos_by_name.setdefault(absolute_video.name, []).append(
             (video, relative_parts)
         )
 
     samples_by_video: dict[Path, list[Path]] = {}
     ambiguous: dict[Path, list[Path]] = {}
     for image_path in image_paths:
-        # Resolved first, to walk the path upwards: the folder of a sample is
-        # named after its video
-        sample_dir = _absolute(image_path).parent
+        # Made absolute first, to walk the path upwards: the folder of a
+        # sample is named after its video
+        sample_dir = Path(os.path.abspath(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 + "_"):
@@ -207,15 +209,6 @@ def find_image_samples(
     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]:
     resolved_paths: set[Path] = set()
     for p in paths:

tests/integration/test_video_process.py

@@ -868,6 +868,37 @@ def test_video_process_processes_the_videos_that_sampled(setup_data: py.path.loc
         assert Path(desc["filename"]).parent.name == "max-360mode.mp4"
 
 
+def test_video_process_samples_a_symlinked_video(setup_data: py.path.local):
+    """
+    A link in the folder to a video outside it crashed with "is not in the
+    subpath of". Its samples are in a folder named after the link, as they are
+    named, and are geotagged
+    """
+    pytest_skip_if_not_ffmpeg_installed()
+
+    video_dir = Path(setup_data, "linked_videos")
+    video_dir.mkdir()
+    try:
+        os.symlink(
+            Path(setup_data, "gopro_data", "max-360mode.mp4"), video_dir / "link.mp4"
+        )
+    except (OSError, NotImplementedError) as ex:
+        pytest.skip(f"Cannot create symbolic links here: {ex}")
+
+    descs = run_video_process_for_descs(
+        [
+            "--video_sample_distance=6",
+            str(video_dir),
+            str(Path(setup_data, "my_samples")),
+        ]
+    )
+    assert descs
+    for desc in descs:
+        assert "error" not in desc, desc
+        assert Path(desc["filename"]).parent.name == "link.mp4"
+        assert Path(desc["filename"]).name.startswith("link_")
+
+
 def test_video_process_skips_the_only_video(setup_data: py.path.local):
     """
     Skipping every video leaves nothing to process, which is not an error:

tests/unit/test_read_track_files_once.py

@@ -10,14 +10,15 @@
 
 from __future__ import annotations
 
+import io
 import logging
 import multiprocessing
+import types
 from pathlib import Path
 
 import gpxpy
 import pytest
-
-from mapillary_tools import sample_video
+from mapillary_tools import exceptions, sample_video
 from mapillary_tools.geotag import (
     geotag_images_from_gpx_file,
     geotag_videos_from_gpx,
@@ -154,26 +155,67 @@ def changing_parse(xml_or_file, *args, **kwargs):
     assert len(utils.read_gpx_tracks(gpx)) == 1
 
 
-def test_a_gpx_file_is_decoded_as_a_file_opened_in_text_mode(
-    tmp_path: Path, parses: list[int]
+def _points(tracks: list[utils.Track]) -> list[list[tuple[float, float]]]:
+    return [[(p.time, p.lat) for p in track] for track in tracks]
+
+
+@pytest.fixture
+def cp1252_locale(monkeypatch):
+    """
+    Open files in text mode with cp1252 when no encoding is given, as on
+    Western Windows
+    """
+    text_io_wrapper = io.TextIOWrapper
+
+    def wrapper(buffer, encoding=None, *args, **kwargs):
+        return text_io_wrapper(buffer, encoding or "cp1252", *args, **kwargs)
+
+    monkeypatch.setattr(
+        utils,
+        "io",
+        types.SimpleNamespace(TextIOWrapper=wrapper, BufferedReader=io.BufferedReader),
+    )
+
+
+@pytest.mark.parametrize("bom", [b"", b"\xef\xbb\xbf"])
+def test_a_gpx_file_is_decoded_as_utf8_whatever_the_locale(
+    tmp_path: Path, parses: list[int], cp1252_locale, bom: bytes
 ):
     """
-    As gpxpy parsed a file opened in text mode before, with the encoding of
-    the platform: a byte order mark and names that are not ASCII
+    Names that cp1252 cannot decode failed the whole file on Western
+    Windows, with or without a byte order mark
     """
+    plain = tmp_path / "plain.gpx"
+    plain.write_bytes(ONE_TRACK)
     gpx = tmp_path / "track.gpx"
-    name = "<name>Stra\u00dfe \u00e9t\u00e9</name>".encode("utf-8")
-    gpx.write_bytes(b"\xef\xbb\xbf" + ONE_TRACK.replace(b"<trk>", b"<trk>" + name))
-
-    def outcome(read):
-        try:
-            return [[(p.time, p.lat) for p in track] for track in read()]
-        except Exception as ex:
-            return type(ex)
-
-    with gpx.open("r") as f:
-        expected = outcome(lambda: utils._parse_gpx_file(f))
-    assert outcome(lambda: utils.parse_gpx(gpx)) == expected
+    name = "<name>\u00c1rbol \u5317\u4eac</name>".encode("utf-8")
+    gpx.write_bytes(bom + ONE_TRACK.replace(b"<trk>", b"<trk>" + name))
+
+    assert _points(utils.parse_gpx(gpx)) == _points(utils.parse_gpx(plain))
+
+
+def test_a_byte_that_is_not_utf8_only_spoils_a_name(tmp_path: Path, parses: list[int]):
+    """An ISO-8859-1 GPX reads with the points of its UTF-8 copy"""
+    plain = tmp_path / "plain.gpx"
+    plain.write_bytes(ONE_TRACK)
+    gpx = tmp_path / "track.gpx"
+    gpx.write_bytes(
+        ONE_TRACK.replace(
+            b'<?xml version="1.0"?>', b'<?xml version="1.0" encoding="ISO-8859-1"?>'
+        ).replace(b"<trk>", "<trk><name>Caf\u00e9</name>".encode("latin-1"))
+    )
+
+    assert _points(utils.parse_gpx(gpx)) == _points(utils.parse_gpx(plain))
+
+
+def test_a_utf16_gpx_file_is_still_invalid(tmp_path: Path, parses: list[int]):
+    gpx = tmp_path / "track.gpx"
+    gpx.write_bytes(ONE_TRACK.decode("utf-8").encode("utf-16"))
+
+    with pytest.raises(exceptions.MapillaryInvalidGPXError):
+        geotag_images_from_gpx_file._read_gpx_points(gpx)
+    with pytest.raises(exceptions.MapillaryInvalidGPXError):
+        GPXVideoExtractor(tmp_path / "video.mp4", gpx)._read_tracks()
 
 
 def test_only_the_last_gpx_files_are_kept(tmp_path: Path, parses: list[int]):

tests/unit/test_sample_video.py

@@ -3243,3 +3243,35 @@ def test_warned_when_given_and_ignored(self, caplog):
         assert gpx.source_path.source_path == Path("/data/track.gpx")
         assert nmea.source_path is not None
         assert nmea.source_path.source_path == Path("/data/other.gpx")
+
+
+def _symlink(link: Path, target: Path) -> None:
+    try:
+        os.symlink(target, link)
+    except (OSError, NotImplementedError) as ex:
+        pytest.skip(f"Cannot create symbolic links here: {ex}")
+
+
+def test_a_symlinked_video_is_sampled_into_a_folder_named_after_the_link(
+    tmp_path: Path,
+):
+    """
+    A link in the folder to a video outside it crashed with "is not in the
+    subpath of", and a link given itself was sampled into a folder named after
+    the video it links to, while its samples were named after the link
+    """
+    target = tmp_path / "elsewhere" / "video.mp4"
+    target.parent.mkdir()
+    target.write_bytes(b"")
+    videos = tmp_path / "videos"
+    videos.mkdir()
+    _symlink(videos / "link.mp4", target)
+    out = tmp_path / "out"
+
+    assert sample_video.find_sample_dirs(videos, out) == [out / "link.mp4"]
+    assert sample_video.find_sample_dirs(videos / "link.mp4", out) == [out / "link.mp4"]
+
+    # Two links to one video are still sampled once, by either link
+    _symlink(videos / "link2.mp4", target)
+    [sample_dir] = sample_video.find_sample_dirs(videos, out)
+    assert sample_dir in [out / "link.mp4", out / "link2.mp4"]

tests/unit/test_utils.py

@@ -86,6 +86,27 @@ def test_find_image_samples_where_resolve_leaves_paths_relative(monkeypatch):
     assert {} == ambiguous
 
 
+def test_find_image_samples_of_a_symlinked_video(tmp_path: Path):
+    """
+    sample_video() names the folder of a symlinked video after the link, which
+    a video path resolved to the file it links to did not match
+    """
+    target = tmp_path / "elsewhere" / "video.mp4"
+    target.parent.mkdir()
+    target.write_bytes(b"")
+    link = tmp_path / "videos" / "link.mp4"
+    link.parent.mkdir()
+    try:
+        os.symlink(target, link)
+    except (OSError, NotImplementedError) as ex:
+        pytest.skip(f"Cannot create symbolic links here: {ex}")
+    sample = tmp_path / "out" / "link.mp4" / "link_0_000001.jpg"
+
+    samples, ambiguous = utils.find_image_samples([sample], [link])
+    assert {link: [sample]} == samples
+    assert {} == ambiguous
+
+
 def test_find_image_samples_of_videos_on_different_drives(monkeypatch):
     """Videos of the same name that no path tells apart could be any of them"""