diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py index a1578934..a85d5101 100644 --- a/mapillary_tools/geotag/utils.py +++ b/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 diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 649382ed..c3256b20 100644 --- a/mapillary_tools/sample_video.py +++ b/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): diff --git a/mapillary_tools/utils.py b/mapillary_tools/utils.py index ec22f9d7..00e64f86 100644 --- a/mapillary_tools/utils.py +++ b/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: diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index 4469e0ff..65c48adc 100644 --- a/tests/integration/test_video_process.py +++ b/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: diff --git a/tests/unit/test_read_track_files_once.py b/tests/unit/test_read_track_files_once.py index bc48bde3..ca38ae5a 100644 --- a/tests/unit/test_read_track_files_once.py +++ b/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 = "Stra\u00dfe \u00e9t\u00e9".encode("utf-8") - gpx.write_bytes(b"\xef\xbb\xbf" + ONE_TRACK.replace(b"", b"" + 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 = "\u00c1rbol \u5317\u4eac".encode("utf-8") + gpx.write_bytes(bom + ONE_TRACK.replace(b"", b"" + 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'', b'' + ).replace(b"", "Caf\u00e9".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]): diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 1f073940..3953fdeb 100644 --- a/tests/unit/test_sample_video.py +++ b/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"] diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 060fc484..e8b18a6f 100644 --- a/tests/unit/test_utils.py +++ b/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"""