diff --git a/README.md b/README.md
index a37bb6d9..8ac3b2f2 100644
--- a/README.md
+++ b/README.md
@@ -241,13 +241,14 @@ Unsupported entries include their filename, category, stable reason code, and
normalized extension. Hidden/system files and common GPS or metadata sidecars
are ignored during directory discovery, but an explicitly supplied unsupported
file is always reported. Such a file also fails processing, with a
-`MapillaryUnsupportedFileError` entry in the image description file, whether
-or not the report is requested, unless it is a known sidecar or companion file
-(`.thm`, `.lrv`, `.gpx`, `.nmea`, `.xml`, `.json`, `.srt`, `.txt`, `.csv`,
-`.log`, `.md5`, macOS `._*` files and `.DS_Store`): those are skipped with a
-warning, so that all the files of a camera folder can be named. The report is
-written only to the requested local path and contains absolute source
-filenames, so treat it as local capture metadata.
+`MapillaryUnsupportedFileError` entry in the image description file, whether or
+not the report is requested, unless it is a known sidecar or companion file
+(`.thm`, `.lrv`, `.lrf`, `.aae`, `.gpx`, `.nmea`, `.kml`, `.kmz`, `.fit`,
+`.xml`, `.json`, `.xmp`, `.srt`, `.txt`, `.csv`, `.log`, `.md5`, macOS `._*`
+files and `.DS_Store`, and Windows `Thumbs.db` and `desktop.ini`): those are
+skipped with a warning, so that all the files of a camera folder can be named.
+The report is written only to the requested local path and contains absolute
+source filenames, so treat it as local capture metadata.
## Upload
diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
index e6b210dc..f136f60d 100644
--- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py
+++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
@@ -40,18 +40,28 @@
# The byte order mark, which some editors write at the start of a UTF-8 file
_BOM = "\ufeff"
-# How much of a file tells whether it is text at all, before it is read whole:
-# a video given as an NMEA file is gigabytes
+# How much of a file tells whether it is an NMEA log at all, before it is read
+# whole: a video given as an NMEA file is gigabytes
_SNIFF_BYTES = 64 * 1024
+# An RMC or a GGA sentence, as in _SENTENCE_RE, in the bytes of a file
+_SENTENCE_BYTES_RE = re.compile(rb"\$[A-Z]{2}(RMC|GGA),")
+
+# A NUL byte, which a logger leaves in a log after a power cut, such as on an
+# SD card: it ends a line, as no sentence holds one
+_NUL = "\x00"
+
# The checksum that ends a sentence: *HH. It is optional
_CHECKSUM_RE = re.compile(r"\*[0-9A-Fa-f]{2}\s*$")
-# The fields of a complete sentence: an RMC sentence has 11 (NMEA 2.0 to
-# 2.2), 12 with the mode of NMEA 2.3, and 13 with the navigational status of
-# NMEA 4.1; a GGA sentence has 14. One without its checksum that has fewer was
-# cut short, such as the last one of a log when the recording stopped
-_MIN_FIELDS = {"RMC": 11, "GGA": 14}
+# The fields that a fix is read from: those of an RMC sentence through the
+# date, the 9th, and those of a GGA sentence through the altitude and its
+# unit, the 10th. One without its checksum that has fewer was cut short, such
+# as the last one of a log when the recording stopped. The fields after them
+# are optional, and receivers leave them off: the magnetic variation, the mode
+# and the navigational status of an RMC sentence, the height of the geoid and
+# the DGPS fields of a GGA sentence
+_MIN_FIELDS = {"RMC": 9, "GGA": 10}
# The time of a sentence (hhmmss, and a fraction of a second), and the date of
# an RMC sentence (ddmmyy). pynmea2 reads a field cut short as another time or
@@ -157,15 +167,16 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
own fields show it. A file of which no fix can be read raises ValueError,
which says why, if it has malformed sentences or bytes that are not text.
- A file that is not text at all, which its first bytes show, raises
- ValueError without being read further.
+ A file that is no NMEA log at all, such as a video, which its first bytes
+ show (see _looks_binary()), raises ValueError without being read further.
+ NUL bytes, which a logger leaves in a log after a power cut, end a line.
A file is read once in a process, however many times it is asked for.
"""
with nmea_file.open("rb") as f:
head = f.read(_SNIFF_BYTES)
- if not _looks_like_text(head):
- raise ValueError("it is not text")
+ if _looks_binary(head):
+ raise ValueError(_describe_binary(head))
data = head + f.read()
key = (str(nmea_file.resolve()), hashlib.sha256(data).hexdigest())
read = _READ.get(key)
@@ -182,25 +193,49 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
return [dataclasses.replace(p) for p in read]
-def _looks_like_text(head: bytes) -> bool:
+def _looks_binary(head: bytes) -> bool:
r"""
- Whether the first bytes **head** of a file can be those of a text file:
- without NUL bytes, and mostly bytes that decode as UTF-8, where a log may
- have a few bytes damaged
+ Whether the first bytes **head** of a file show it to be no NMEA log at
+ all, such as a video: they hold no RMC or GGA sentence, and NUL bytes or
+ mostly bytes that do not decode as UTF-8. A log padded with NUL bytes, as a
+ logger leaves it after a power cut, holds sentences, and so does one with
+ a few bytes damaged
- >>> _looks_like_text(b"$GPRMC,024636.00,A,3735.30670,N\r\n")
- True
- >>> _looks_like_text(b"$GPRMC,024636.00,\xe9,3735.30670,N\r\n")
- True
- >>> _looks_like_text(b"\x00\x00\x00\x18ftypmp42")
+ >>> _looks_binary(b"$GPRMC,024636.00,A,3735.30670,N\r\n" + b"\x00" * 100)
False
- >>> _looks_like_text(b"\xca\xfe\xba\xbe" * 100)
+ >>> _looks_binary(b"\x00" * 100 + b"$GPGGA,024636.00,3735.30670,N\r\n")
False
+ >>> _looks_binary(b"$GPRMC,024636.00,\xe9,3735.30670,N\r\n")
+ False
+ >>> _looks_binary(b"$GPGSV,3,1,11,03,03,111,00\r\n")
+ False
+ >>> _looks_binary(b"\x00\x00\x00\x18ftypmp42")
+ True
+ >>> _looks_binary(b"\xca\xfe\xba\xbe" * 100)
+ True
"""
- if b"\x00" in head:
+ if _SENTENCE_BYTES_RE.search(head):
return False
+ if b"\x00" in head:
+ return True
text = head.decode("utf-8", errors="replace")
- return text.count(_NOT_TEXT) * 2 <= len(text)
+ return len(text) < text.count(_NOT_TEXT) * 2
+
+
+def _describe_binary(head: bytes) -> str:
+ r"""
+ Why a file whose first bytes **head** show it to be no NMEA log is not
+ read: it is UTF-16 text, which some Windows tools write, or no text at all
+
+ >>> _describe_binary("$GPRMC,024636.00,A".encode("utf-16"))
+ 'it is UTF-16 text, which is not read: save it as UTF-8 or ASCII'
+ >>> _describe_binary(b"\x00\x00\x00\x18ftypmp42")
+ 'it is not text'
+ """
+ for encoding in ("utf-16", "utf-16-le", "utf-16-be"):
+ if _SENTENCE_RE.search(head.decode(encoding, errors="ignore")):
+ return "it is UTF-16 text, which is not read: save it as UTF-8 or ASCII"
+ return "it is not text"
def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
@@ -217,14 +252,15 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
io.BytesIO(data), encoding="utf-8-sig", errors="replace"
) as f:
lines = f.readlines()
- lines = [line.rstrip("\n\r").lstrip(_BOM) for line in lines]
not_text = any(_NOT_TEXT in line for line in lines)
- # (line number, line) of each RMC and GGA sentence
+ # (line number, line) of each RMC and GGA sentence, of the lines that NUL
+ # bytes end too (see _NUL)
sentences = [
- (number, line)
+ (number, piece.lstrip(_BOM))
for number, line in enumerate(lines, start=1)
- if _SENTENCE_RE.search(line)
+ for piece in line.rstrip("\n\r").split(_NUL)
+ if _SENTENCE_RE.search(piece)
]
messages = []
@@ -330,7 +366,7 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
"""
Parse an NMEA sentence, raising ValueError if it is malformed: if it does
not parse or its checksum does not match, if it has no checksum and fewer
- fields than a complete sentence (see _MIN_FIELDS), if its time or the date
+ fields than a fix is read from (see _MIN_FIELDS), if its time or the date
of an RMC sentence is not hhmmss[.ss] or ddmmyy, or if a field that a fix
is read from does not parse, which pynmea2 leaves a string, or if it has a
byte that is not text, decoded as U+FFFD
@@ -347,15 +383,17 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
Traceback (most recent call last):
...
ValueError: malformed datestamp: '2212'
- >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222")
+ >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0")
Traceback (most recent call last):
...
- ValueError: cut short: 9 of the 11 fields of an RMC sentence
+ ValueError: cut short: 8 of the 9 fields of an RMC sentence that a fix is read from
+ >>> type(_parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222")).__name__
+ 'RMC'
>>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12")
Traceback (most recent call last):
...
- ValueError: cut short: 9 of the 14 fields of a GGA sentence
- >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M,,")).__name__
+ ValueError: cut short: 9 of the 10 fields of a GGA sentence that a fix is read from
+ >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M")).__name__
'GGA'
"""
if _NOT_TEXT in line:
@@ -382,7 +420,7 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
if kind in _MIN_FIELDS and len(message.data) < _MIN_FIELDS[kind]:
article = "an" if kind == "RMC" else "a"
raise ValueError(
- f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence"
+ f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence that a fix is read from"
)
fields: dict[str, type] = {}
diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py
index fa0a0b4a..a1578934 100644
--- a/mapillary_tools/geotag/utils.py
+++ b/mapillary_tools/geotag/utils.py
@@ -32,6 +32,9 @@
] = collections.OrderedDict()
_MAX_PARSED_GPX = 4
+# GPX files are read in chunks of this size to tell whether they changed
+_CHUNK_SIZE = 1024 * 1024
+
# The warnings given in this process about the track files read, so that each
# is given once however many times a file is read
_WARNED: set[tuple[T.Hashable, ...]] = set()
@@ -58,18 +61,23 @@ def read_gpx_tracks(gpx_file: Path) -> list[Track]:
as a file that geotags several videos is read before the pool (see
GeotagVideosFromGPX), and would free them only when it exits.
"""
- data = gpx_file.read_bytes()
- key = (str(gpx_file.resolve()), hashlib.sha256(data).hexdigest())
+ path = str(gpx_file.resolve())
+ with gpx_file.open("rb") as f:
+ key = (path, _digest(f))
parsed = _PARSED_GPX.get(key)
if parsed is None:
- try:
- # Decoded as a file opened in text mode would be, and parsed
- # without the bytes, as a file is
- text = io.TextIOWrapper(io.BytesIO(data)).read()
- del data
- parsed = _parse_gpx_text(text)
- except (gpxpy.gpx.GPXException, UnicodeError) as ex:
- parsed = ex
+ with gpx_file.open("rb") as f:
+ reader = _DigestingReader(f)
+ 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)))
+ except (gpxpy.gpx.GPXException, UnicodeError) as ex:
+ parsed = ex
+ # By the content parsed, which a file changed since it was looked
+ # up by does not have
+ key = (path, reader.hexdigest())
if multiprocessing.parent_process() is None:
_PARSED_GPX[key] = parsed
while _MAX_PARSED_GPX < len(_PARSED_GPX):
@@ -110,8 +118,55 @@ def warn_once(
LOG.warning(message, *args)
-def _parse_gpx_text(text: str) -> list[Track]:
- gpx = gpxpy.parse(text)
+def _digest(f: T.BinaryIO) -> str:
+ """The SHA-256 digest of the rest of the binary file **f**, read in chunks"""
+ digest = hashlib.sha256()
+ for chunk in iter(lambda: f.read(_CHUNK_SIZE), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+class _DigestingReader(io.RawIOBase):
+ """
+ A binary file that computes the SHA-256 digest of the bytes read from it
+
+ >>> reader = _DigestingReader(io.BytesIO(b"abc"))
+ >>> io.BufferedReader(reader).read()
+ b'abc'
+ >>> reader.hexdigest() == hashlib.sha256(b"abc").hexdigest()
+ True
+ """
+
+ def __init__(self, raw: T.BinaryIO) -> None:
+ super().__init__()
+ self._raw = raw
+ self._digest = hashlib.sha256()
+
+ def readable(self) -> bool:
+ return True
+
+ def readinto(self, buffer: T.Any) -> int:
+ data = self._raw.read(len(buffer))
+ buffer[: len(data)] = data
+ self._digest.update(data)
+ return len(data)
+
+ def readall(self) -> bytes:
+ # At once, as a file opened in binary mode reads it, rather than in
+ # chunks joined into a second copy
+ data = self._raw.read()
+ self._digest.update(data)
+ return data
+
+ def hexdigest(self) -> str:
+ """The digest of the whole file: of the bytes read, and of the rest"""
+ for chunk in iter(lambda: self._raw.read(_CHUNK_SIZE), b""):
+ self._digest.update(chunk)
+ return self._digest.hexdigest()
+
+
+def _parse_gpx_file(f: T.TextIO) -> list[Track]:
+ gpx = gpxpy.parse(f)
tracks: list[Track] = []
diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py
index a0fa28be..8e6e177f 100644
--- a/mapillary_tools/process_geotag_properties.py
+++ b/mapillary_tools/process_geotag_properties.py
@@ -29,17 +29,25 @@
]
# Files that cameras and other tools write beside images and videos: GoPro
-# thumbnails (.THM) and low-resolution proxies (.LRV), GPS logs, metadata,
-# subtitles (.SRT), notes and checksums. Selecting all the files of a camera
-# folder names them along with its images and videos
+# thumbnails (.THM) and low-resolution proxies (.LRV), DJI low-resolution
+# proxies (.LRF), Apple photo edits (.AAE), GPS logs and tracks (.gpx, .nmea,
+# .kml, .kmz, .fit), metadata (.xml, .json, .xmp), subtitles (.SRT), notes and
+# checksums. Selecting all the files of a camera folder names them along with
+# its images and videos
_SIDECAR_EXTENSIONS = frozenset(
{
".thm",
".lrv",
+ ".lrf",
+ ".aae",
".gpx",
".nmea",
+ ".kml",
+ ".kmz",
+ ".fit",
".xml",
".json",
+ ".xmp",
".srt",
".txt",
".csv",
@@ -48,6 +56,11 @@
}
)
+# Files that operating systems write into folders of images and videos: the
+# Finder settings of macOS, and the thumbnail cache and folder settings of
+# Windows. Compared case-insensitively
+_SIDECAR_FILENAMES = frozenset({".ds_store", "thumbs.db", "desktop.ini"})
+
def _normalize_import_paths(import_path: Path | T.Sequence[Path]) -> T.Sequence[Path]:
import_paths: T.Sequence[Path]
@@ -155,21 +168,23 @@ def process_geotag_properties(
def _is_sidecar_file(path: Path) -> bool:
"""
Whether a file is one that accompanies images and videos (see
- _SIDECAR_EXTENSIONS), or one that macOS writes beside them: .DS_Store, and
- the AppleDouble file ._NAME of each file it copies to a drive that cannot
- hold its metadata, which has the extension of that file but is no image or
- video
+ _SIDECAR_EXTENSIONS), one that an operating system writes beside them (see
+ _SIDECAR_FILENAMES), or the AppleDouble file ._NAME that macOS writes for
+ each file it copies to a drive that cannot hold its metadata, which has the
+ extension of that file but is no image or video
>>> [_is_sidecar_file(Path(name)) for name in ["GH010001.THM", "GL010001.LRV", "track.GPX"]]
[True, True, True]
- >>> [_is_sidecar_file(Path(name)) for name in ["._GOPR0001.JPG", ".DS_Store"]]
- [True, True]
+ >>> [_is_sidecar_file(Path(name)) for name in ["DJI_0001.LRF", "IMG_0001.AAE", "photo.XMP"]]
+ [True, True, True]
+ >>> [_is_sidecar_file(Path(name)) for name in ["._GOPR0001.JPG", ".DS_Store", "Thumbs.db", "Desktop.ini"]]
+ [True, True, True, True]
>>> [_is_sidecar_file(Path(name)) for name in ["GOPR0001.JPG", "VID_00_001.insv", "photo.png"]]
[False, False, False]
"""
return (
path.name.startswith("._")
- or path.name.casefold() == ".ds_store"
+ or path.name.casefold() in _SIDECAR_FILENAMES
or path.suffix.lower() in _SIDECAR_EXTENSIONS
)
diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py
index 59b24b81..649382ed 100644
--- a/mapillary_tools/sample_video.py
+++ b/mapillary_tools/sample_video.py
@@ -203,10 +203,13 @@ def _parse_sample_sources(
options = factory.parse_source_options(
geotag_source=geotag_source,
video_geotag_source=[],
- geotag_source_path=(
- video_import_path if geotag_source_path is None else geotag_source_path
- ),
+ geotag_source_path=geotag_source_path,
)
+ # Without --geotag_source_path, the sources without a path of their own
+ # read the video import path, as process_geotag_properties() gives it, and
+ # those with one ignore it without a word
+ if geotag_source_path is None:
+ factory.give_video_import_path(options, video_import_path)
for option in options:
if option.interpolation is None:
option.interpolation = interpolation
diff --git a/tests/integration/test_process_report.py b/tests/integration/test_process_report.py
index c7bd5f44..86730bf9 100644
--- a/tests/integration/test_process_report.py
+++ b/tests/integration/test_process_report.py
@@ -215,3 +215,41 @@ def test_process_skips_named_sidecars_of_a_camera_folder(tmpdir: py.path.local):
(desc["filename"], desc["error"]["type"])
for desc in json.loads(desc_path.read_text(encoding="utf-8"))
] == [(str(original.resolve()), "MapillaryUnsupportedFileError")]
+
+
+def test_process_skips_named_companion_files_of_other_cameras_and_systems(
+ tmpdir: py.path.local,
+):
+ """
+ Named along with the images, DJI proxies, Apple photo edits, GPS tracks,
+ XMP metadata and the files that Windows writes into folders failed the run
+ """
+ root = Path(str(tmpdir))
+ folder = root / "DCIM"
+ folder.mkdir()
+ companions = [
+ folder / "DJI_0001.LRF",
+ folder / "IMG_0001.AAE",
+ folder / "IMG_0001.XMP",
+ folder / "track.kml",
+ folder / "track.KMZ",
+ folder / "ride.fit",
+ folder / "Thumbs.db",
+ folder / "desktop.ini",
+ ]
+ for companion in companions:
+ companion.write_bytes(b"not an image")
+
+ descs, report = _run_process_with_report(companions, root)
+ assert descs == []
+ assert report["processed_file_count"] == 0
+ assert sorted(skipped["filename"] for skipped in report["skipped_files"]) == sorted(
+ str(companion.resolve()) for companion in companions
+ )
+
+ desc_path = root / "description.json"
+ run_command(
+ ["--desc_path", str(desc_path), *(str(companion) for companion in companions)],
+ command="process",
+ )
+ assert json.loads(desc_path.read_text(encoding="utf-8")) == []
diff --git a/tests/unit/test_nmea_parser.py b/tests/unit/test_nmea_parser.py
index 1df9f548..4b2d216a 100644
--- a/tests/unit/test_nmea_parser.py
+++ b/tests/unit/test_nmea_parser.py
@@ -269,7 +269,7 @@ def test_sentences_cut_short_are_skipped_with_one_warning(tmp_path, caplog):
lines += [_rmc(f"02463{second}.00"), _gga(f"02463{second}.00")]
# Every other fix cut short, at another field each time
for idx in range(1, 20, 4):
- lines[idx] = lines[idx][: 10 + 3 * idx]
+ lines[idx] = lines[idx][: 6 + 3 * idx]
with caplog.at_level(logging.WARNING):
points = _read(tmp_path, *lines)
@@ -388,17 +388,43 @@ def test_complete_sentences_without_checksums_are_read(tmp_path, caplog, rmc_tai
assert not caplog.text
+@pytest.mark.parametrize(
+ "rmc, gga",
+ [
+ # Through the date, and without the height of the geoid and the DGPS
+ # fields
+ (
+ "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222",
+ "$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M",
+ ),
+ # Through the magnetic variation, and without the DGPS fields
+ (
+ "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,",
+ "$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M",
+ ),
+ ],
+ ids=["9 and 10 fields", "10 and 12 fields"],
+)
+def test_sentences_without_their_optional_fields_are_read(tmp_path, caplog, rmc, gga):
+ """A log without checksums whose sentences left them off was failed whole"""
+ with caplog.at_level(logging.WARNING):
+ fixes = _read(tmp_path, rmc, gga)
+
+ assert fixes == [_fix("02:46:36")]
+ assert not caplog.text
+
+
@pytest.mark.parametrize(
"cut, reason",
[
# Whole fields, the last of them maybe cut short too
(
- "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222",
- "cut short: 9 of the 11 fields of an RMC sentence",
+ "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0",
+ "cut short: 8 of the 9 fields of an RMC sentence that a fix is read from",
),
(
"$GPGGA,024637.00,3735.30670,N,12220.25810,W,1,08,1.0,12",
- "cut short: 9 of the 14 fields of a GGA sentence",
+ "cut short: 9 of the 10 fields of a GGA sentence that a fix is read from",
),
# A time that does not parse
("$GPGGA,02463", "malformed timestamp: '02463'"),
@@ -503,6 +529,54 @@ def test_a_file_that_is_not_text_fails_without_being_read_whole(tmp_path, head):
_read_bytes(tmp_path, head + padding + sentences)
+def test_a_log_padded_with_nul_bytes_is_read(tmp_path, caplog):
+ """
+ A logger leaves NUL bytes in a log after a power cut, such as on an SD card:
+ one NUL byte failed the whole log as not text
+ """
+ lines = [_rmc("024636.00"), _gga("024636.00"), _rmc("024637.00"), _gga("024637.00")]
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(tmp_path, _crlf(lines) + b"\x00" * 4096)
+
+ assert len(times) == 2
+ assert not caplog.text
+
+
+def test_nul_bytes_before_the_first_sentence_are_skipped(tmp_path, caplog):
+ lines = [_rmc("024636.00"), _gga("024636.00")]
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(tmp_path, b"\x00" * 512 + _crlf(lines))
+
+ assert len(times) == 1
+ assert not caplog.text
+
+
+def test_nul_bytes_end_a_sentence_as_a_line_end_does(tmp_path, caplog):
+ """A logger that resumed after a power cut wrote on after the NUL bytes"""
+ first = (_rmc("024636.00") + "\r\n" + _gga("024636.00")).encode("ascii")
+ second = _crlf([_rmc("024637.00"), _gga("024637.00")])
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(tmp_path, first + b"\x00" * 64 + second)
+
+ assert len(times) == 2
+ assert not caplog.text
+
+
+def test_a_sentence_with_nul_bytes_inside_is_skipped(tmp_path, caplog):
+ damaged = _gga("024637.00").replace(",N,", ",N\0\0", 1)
+ lines = [_rmc("024636.00"), _gga("024636.00"), _rmc("024637.00"), damaged]
+ content = "".join(f"{line}\r\n" for line in lines).encode("ascii")
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(tmp_path, content)
+
+ assert len(times) == 1
+ assert "Skipping 1 of the 4 RMC and GGA sentences" in caplog.text
+
+
def test_a_log_longer_than_what_tells_text_is_read_whole(tmp_path):
lines = []
for minute in range(60):
@@ -543,13 +617,14 @@ def test_logs_with_byte_order_marks_appended_are_read(tmp_path, caplog):
assert not caplog.text
-def test_a_utf16_log_is_not_read(tmp_path):
- """As one that is not text: NMEA is ASCII"""
+@pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-16-be"])
+def test_a_utf16_log_is_not_read(tmp_path, encoding):
+ """NMEA is ASCII: UTF-16 is not read, and the error says it is UTF-16"""
content = "".join(
f"{line}\r\n" for line in [_rmc("024636.00"), _gga("024636.00")]
- ).encode("utf-16")
+ ).encode(encoding)
- with pytest.raises(ValueError, match="^it is not text$"):
+ with pytest.raises(ValueError, match="^it is UTF-16 text, which is not read"):
_read_bytes(tmp_path, content)
diff --git a/tests/unit/test_process_geotag_properties.py b/tests/unit/test_process_geotag_properties.py
index 07bc3dd9..6d26ad8e 100644
--- a/tests/unit/test_process_geotag_properties.py
+++ b/tests/unit/test_process_geotag_properties.py
@@ -217,6 +217,15 @@ def process(paths, options, filetypes=None):
"logger.log",
"checksums.md5",
".DS_Store",
+ "photo.XMP",
+ "track.kml",
+ "track.KMZ",
+ "ride.fit",
+ "DJI_0001.LRF",
+ "IMG_0001.AAE",
+ "Thumbs.db",
+ "desktop.ini",
+ "Desktop.INI",
],
)
def test_sidecar_is_skipped_with_a_warning(
diff --git a/tests/unit/test_read_track_files_once.py b/tests/unit/test_read_track_files_once.py
index 960a74df..bc48bde3 100644
--- a/tests/unit/test_read_track_files_once.py
+++ b/tests/unit/test_read_track_files_once.py
@@ -94,6 +94,88 @@ def test_a_changed_gpx_file_is_parsed_again(tmp_path: Path, parses: list[int]):
assert len(parses) == 2
+def test_gpxpy_reads_the_file_itself(tmp_path: Path, parses: list[int], monkeypatch):
+ """
+ gpxpy reads the whole file into a string itself: a string of its content
+ passed to it stayed alive beside that while it parsed
+ """
+ arguments = []
+ parse = gpxpy.parse
+
+ def recording_parse(xml_or_file, *args, **kwargs):
+ arguments.append(xml_or_file)
+ return parse(xml_or_file, *args, **kwargs)
+
+ monkeypatch.setattr(gpxpy, "parse", recording_parse)
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+
+ assert len(utils.parse_gpx(gpx)) == 1
+ assert [hasattr(argument, "read") for argument in arguments] == [True]
+
+
+def test_a_gpx_file_larger_than_a_chunk_changed_is_parsed_again(
+ tmp_path: Path, parses: list[int], monkeypatch
+):
+ """The content is read in chunks to tell whether it changed"""
+ monkeypatch.setattr(utils, "_CHUNK_SIZE", 16)
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(TWO_TRACKS)
+ assert len(utils.parse_gpx(gpx)) == 2
+
+ # Of the same size, changed past the first chunk
+ changed = TWO_TRACKS.replace(b'lat="52.3"', b'lat="52.4"')
+ assert len(changed) == len(TWO_TRACKS)
+ gpx.write_bytes(changed)
+ assert utils.parse_gpx(gpx)[1][1].lat == 52.4
+ assert utils.parse_gpx(gpx)[1][1].lat == 52.4
+ assert len(parses) == 2
+
+
+def test_a_gpx_file_is_kept_by_the_content_parsed(
+ tmp_path: Path, parses: list[int], monkeypatch
+):
+ """A file changed between looking it up and parsing it is kept as parsed"""
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+ parse = gpxpy.parse
+
+ def changing_parse(xml_or_file, *args, **kwargs):
+ # Changed after it was looked up, before it is read
+ gpx.write_bytes(TWO_TRACKS)
+ return parse(xml_or_file, *args, **kwargs)
+
+ monkeypatch.setattr(gpxpy, "parse", changing_parse)
+ assert len(utils.read_gpx_tracks(gpx)) == 2
+
+ # Changed back: what was parsed is not served for it
+ monkeypatch.setattr(gpxpy, "parse", parse)
+ gpx.write_bytes(ONE_TRACK)
+ 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]
+):
+ """
+ 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
+ """
+ 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
+
+
def test_only_the_last_gpx_files_are_kept(tmp_path: Path, parses: list[int]):
paths = []
for idx in range(utils._MAX_PARSED_GPX + 1):
diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py
index 78485f36..1f073940 100644
--- a/tests/unit/test_sample_video.py
+++ b/tests/unit/test_sample_video.py
@@ -3198,3 +3198,48 @@ def test_it_fails_instead_of_writing_an_empty_sample_directory(
assert "none of its frames falls within its GPS track" in caplog.text
assert "before the video starts" in caplog.text
assert not Path(tmpdir).joinpath("hello.mp4").exists()
+
+
+class TestSampleSourcesPath:
+ """
+ The sources that the geotagging stage geotags the samples by read the
+ video import path when --geotag_source_path is not given. A source with a
+ path of its own was warned to ignore --geotag_source_path, although it was
+ not given
+ """
+
+ _WARNING = "The option --geotag_source_path is ignored"
+
+ @staticmethod
+ def _sources(geotag_source_path: Path | None) -> list[SourceOption]:
+ options = sample_video._parse_sample_sources(
+ [
+ '{"source": "gpx", "source_path": "/data/track.gpx"}',
+ "nmea",
+ ],
+ geotag_source_path,
+ Path("/data/videos"),
+ InterpolationOption(),
+ )
+ assert options is not None
+ return options
+
+ def test_not_warned_without_geotag_source_path(self, caplog):
+ with caplog.at_level(logging.WARNING):
+ gpx, nmea = self._sources(None)
+
+ assert self._WARNING not in caplog.text
+ assert gpx.source_path is not None
+ 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/videos")
+
+ def test_warned_when_given_and_ignored(self, caplog):
+ with caplog.at_level(logging.WARNING):
+ gpx, nmea = self._sources(Path("/data/other.gpx"))
+
+ assert caplog.text.count(self._WARNING) == 1
+ assert gpx.source_path is not None
+ 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")