From 55953ab4edd86ffed7fe4479a6df9da6eecab362 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 28 Jan 2026 18:54:23 +0100 Subject: [PATCH] Extract and use camera serial number for sequence building Summary: This change adds camera serial number extraction from EXIF/XMP metadata and uses it for sequence building. Previously, images were grouped into sequences only by directory, device make/model, and dimensions. This could incorrectly group images from different physical cameras of the same model into the same sequence. With this change, images from different camera bodies (identified by their unique serial numbers) are now correctly split into separate sequences, even if they have the same make/model. **Changes:** - Added `extract_camera_uuid()` method to `ExifReadFromXMP`, `ExifReadFromEXIF`, and `ExifToolReadVideo` classes - Extracts body serial and lens serial from various EXIF/XMP tags, creating a composite ID when both are present - Added `MAPCameraUUID` field to `VideoMetadata` type and propagated through the pipeline - Added `MAPCameraUUID` to sequence grouping criteria in `process_sequence_properties.py` - Updated JSON serialization to include the new field Test Plan: Added comprehensive unit tests covering: - EXIF body/lens serial extraction with priority ordering - XMP serial extraction from various namespace tags (exif, exifEX, aux) - Video metadata serial extraction (GoPro, Insta360, ExifIFD) - Sequence grouping by camera UUID (verifying images with different camera UUIDs are split into separate sequences) All existing tests pass. --- mapillary_tools/exif_read.py | 86 +++++ mapillary_tools/exiftool_read.py | 85 +++++ mapillary_tools/exiftool_read_video.py | 52 +++ .../geotag/geotag_images_from_video.py | 1 + .../geotag/image_extractors/exif.py | 1 + .../geotag/video_extractors/exiftool.py | 2 + .../process_sequence_properties.py | 1 + mapillary_tools/serializer/description.py | 8 + mapillary_tools/types.py | 1 + schema/image_description_schema.json | 4 + tests/unit/test_exifread.py | 353 ++++++++++++++++++ tests/unit/test_sequence_processing.py | 94 +++++ tests/unit/test_types.py | 9 + 13 files changed, 697 insertions(+) diff --git a/mapillary_tools/exif_read.py b/mapillary_tools/exif_read.py index 970021b8a..d7b8b1422 100644 --- a/mapillary_tools/exif_read.py +++ b/mapillary_tools/exif_read.py @@ -29,6 +29,7 @@ "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#", "x": "adobe:ns:meta/", "GPano": "http://ns.google.com/photos/1.0/panorama/", + "aux": "http://ns.adobe.com/exif/1.0/aux/", } # https://github.com/ianare/exif-py/issues/167 EXIFREAD_LOG = logging.getLogger("exifread") @@ -334,6 +335,10 @@ def extract_height(self) -> int | None: def extract_orientation(self) -> int: raise NotImplementedError + @abc.abstractmethod + def extract_camera_uuid(self) -> str | None: + raise NotImplementedError + class ExifReadFromXMP(ExifReadABC): def __init__(self, etree: et.ElementTree): @@ -482,6 +487,41 @@ def extract_orientation(self) -> int: return 1 return orientation + def extract_camera_uuid(self) -> str | None: + """ + Extract camera unique identifier from serial number tags in XMP. + Builds a composite ID from body and lens serial numbers. + """ + body_serial = self._extract_alternative_fields( + [ + "exif:SerialNumber", + "exif:BodySerialNumber", + "exif:CameraSerialNumber", + "exifEX:SerialNumber", + "exifEX:BodySerialNumber", + "aux:SerialNumber", + ], + str, + ) + lens_serial = self._extract_alternative_fields( + [ + "exif:LensSerialNumber", + "exifEX:LensSerialNumber", + "aux:LensSerialNumber", + ], + str, + ) + + parts = [] + if body_serial: + parts.append(body_serial.strip()) + if lens_serial: + parts.append(lens_serial.strip()) + + if parts: + return "_".join(parts) + return None + def _extract_alternative_fields( self, fields: T.Iterable[str], @@ -816,6 +856,40 @@ def extract_orientation(self) -> int: return 1 return orientation + def extract_camera_uuid(self) -> str | None: + """ + Extract camera unique identifier from serial number EXIF tags. + Builds a composite ID from body and lens serial numbers. + """ + body_serial = self._extract_alternative_fields( + [ + "EXIF BodySerialNumber", + "EXIF SerialNumber", + "EXIF CameraSerialNumber", + "Image BodySerialNumber", + "MakerNote SerialNumber", + "MakerNote InternalSerialNumber", + ], + str, + ) + lens_serial = self._extract_alternative_fields( + [ + "EXIF LensSerialNumber", + "Image LensSerialNumber", + ], + str, + ) + + parts = [] + if body_serial: + parts.append(body_serial.strip()) + if lens_serial: + parts.append(lens_serial.strip()) + + if parts: + return "_".join(parts) + return None + def _extract_alternative_fields( self, fields: T.Iterable[str], @@ -987,3 +1061,15 @@ def extract_height(self) -> int | None: if val is not None: return val return None + + def extract_camera_uuid(self) -> str | None: + val = super().extract_camera_uuid() + if val is not None: + return val + xmp = self._xmp_with_reason("camera_uuid") + if xmp is None: + return None + val = xmp.extract_camera_uuid() + if val is not None: + return val + return None diff --git a/mapillary_tools/exiftool_read.py b/mapillary_tools/exiftool_read.py index 969895720..bbf7c6cc8 100644 --- a/mapillary_tools/exiftool_read.py +++ b/mapillary_tools/exiftool_read.py @@ -17,10 +17,13 @@ EXIFTOOL_NAMESPACES: dict[str, str] = { "Adobe": "http://ns.exiftool.org/APP14/Adobe/1.0/", "Apple": "http://ns.exiftool.org/MakerNotes/Apple/1.0/", + "Canon": "http://ns.exiftool.org/MakerNotes/Canon/1.0/", "Composite": "http://ns.exiftool.org/Composite/1.0/", "ExifIFD": "http://ns.exiftool.org/EXIF/ExifIFD/1.0/", "ExifTool": "http://ns.exiftool.org/ExifTool/1.0/", "File": "http://ns.exiftool.org/File/1.0/", + "FLIR": "http://ns.exiftool.org/APP1/FLIR/1.0/", + "FujiFilm": "http://ns.exiftool.org/MakerNotes/FujiFilm/1.0/", "GPS": "http://ns.exiftool.org/EXIF/GPS/1.0/", "GoPro": "http://ns.exiftool.org/APP6/GoPro/1.0/", "ICC-chrm": "http://ns.exiftool.org/ICC_Profile/ICC-chrm/1.0/", @@ -33,11 +36,20 @@ "IPTC": "http://ns.exiftool.org/IPTC/IPTC/1.0/", "InteropIFD": "http://ns.exiftool.org/EXIF/InteropIFD/1.0/", "JFIF": "http://ns.exiftool.org/JFIF/JFIF/1.0/", + "Kodak": "http://ns.exiftool.org/MakerNotes/Kodak/1.0/", + "Leica": "http://ns.exiftool.org/MakerNotes/Leica/1.0/", "MPF0": "http://ns.exiftool.org/MPF/MPF0/1.0/", "MPImage1": "http://ns.exiftool.org/MPF/MPImage1/1.0/", "MPImage2": "http://ns.exiftool.org/MPF/MPImage2/1.0/", + "Nikon": "http://ns.exiftool.org/MakerNotes/Nikon/1.0/", + "Olympus": "http://ns.exiftool.org/MakerNotes/Olympus/1.0/", + "Panasonic": "http://ns.exiftool.org/MakerNotes/Panasonic/1.0/", + "Pentax": "http://ns.exiftool.org/MakerNotes/Pentax/1.0/", "Photoshop": "http://ns.exiftool.org/Photoshop/Photoshop/1.0/", + "Ricoh": "http://ns.exiftool.org/MakerNotes/Ricoh/1.0/", "Samsung": "http://ns.exiftool.org/MakerNotes/Samsung/1.0/", + "Sigma": "http://ns.exiftool.org/MakerNotes/Sigma/1.0/", + "Sony": "http://ns.exiftool.org/MakerNotes/Sony/1.0/", "System": "http://ns.exiftool.org/File/System/1.0/", "XMP-GAudio": "http://ns.exiftool.org/XMP/XMP-GAudio/1.0/", "XMP-GImage": "http://ns.exiftool.org/XMP/XMP-GImage/1.0/", @@ -53,6 +65,8 @@ "XMP-xmp": "http://ns.exiftool.org/XMP/XMP-xmp/1.0/", "XMP-xmpMM": "http://ns.exiftool.org/XMP/XMP-xmpMM/1.0/", "XMP-xmpNote": "http://ns.exiftool.org/XMP/XMP-xmpNote/1.0/", + "XMP-drone-dji": "http://ns.exiftool.org/XMP/XMP-drone-dji/1.0/", + "DJI": "http://ns.exiftool.org/MakerNotes/DJI/1.0/", "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#", } @@ -426,6 +440,77 @@ def extract_orientation(self) -> int: return 1 return orientation + def extract_camera_uuid(self) -> str | None: + """ + Extract camera UUID from serial numbers. + Returns a composite ID from body serial and lens serial if available. + """ + # Try body serial number from various sources + body_serial = self._extract_alternative_fields( + [ + # Standard EXIF tags (BodySerialNumber has priority over generic SerialNumber) + "ExifIFD:BodySerialNumber", + "ExifIFD:SerialNumber", + "IFD0:CameraSerialNumber", + "IFD0:SerialNumber", + # MakerNotes - camera specific + "Canon:SerialNumber", + "Canon:InternalSerialNumber", + "DJI:SerialNumber", + "XMP-drone-dji:CameraSerialNumber", + "XMP-drone-dji:DroneSerialNumber", + "FLIR:CameraSerialNumber", + "FujiFilm:InternalSerialNumber", + "GoPro:CameraSerialNumber", + "Kodak:SerialNumber", + "Leica:SerialNumber", + "Leica:InternalSerialNumber", + "Nikon:SerialNumber", + "Olympus:SerialNumber", + "Olympus:InternalSerialNumber", + "Panasonic:InternalSerialNumber", + "Pentax:SerialNumber", + "Pentax:InternalSerialNumber", + "Ricoh:SerialNumber", + "Ricoh:InternalSerialNumber", + "Ricoh:BodySerialNumber", + "Sigma:SerialNumber", + "Sony:InternalSerialNumber", + # XMP equivalents + "XMP-exif:SerialNumber", + "XMP-exif:BodySerialNumber", + "XMP-exifEX:SerialNumber", + "XMP-exif:CameraSerialNumber", + "XMP-exifEX:BodySerialNumber", + "XMP-aux:SerialNumber", + ], + str, + ) + + # Try lens serial number + lens_serial = self._extract_alternative_fields( + [ + "ExifIFD:LensSerialNumber", + "FLIR:LensSerialNumber", + "Olympus:LensSerialNumber", + "Panasonic:LensSerialNumber", + "Ricoh:LensSerialNumber", + "XMP-exifEX:LensSerialNumber", + "XMP-aux:LensSerialNumber", + ], + str, + ) + + parts = [] + if body_serial: + parts.append(body_serial.strip()) + if lens_serial: + parts.append(lens_serial.strip()) + + if parts: + return "_".join(parts) + return None + def _extract_alternative_fields( self, fields: T.Sequence[str], diff --git a/mapillary_tools/exiftool_read_video.py b/mapillary_tools/exiftool_read_video.py index a4d6d3e56..b1bff9d57 100644 --- a/mapillary_tools/exiftool_read_video.py +++ b/mapillary_tools/exiftool_read_video.py @@ -19,10 +19,16 @@ EXIFTOOL_NAMESPACES: dict[str, str] = { "Keys": "http://ns.exiftool.org/QuickTime/Keys/1.0/", "IFD0": "http://ns.exiftool.org/EXIF/IFD0/1.0/", + "ExifIFD": "http://ns.exiftool.org/EXIF/ExifIFD/1.0/", "QuickTime": "http://ns.exiftool.org/QuickTime/QuickTime/1.0/", "UserData": "http://ns.exiftool.org/QuickTime/UserData/1.0/", "Insta360": "http://ns.exiftool.org/Trailer/Insta360/1.0/", "GoPro": "http://ns.exiftool.org/QuickTime/GoPro/1.0/", + "Ricoh": "http://ns.exiftool.org/MakerNotes/Ricoh/1.0/", + "XMP-GSpherical": "http://ns.exiftool.org/XMP/XMP-GSpherical/1.0/", + "XMP-aux": "http://ns.exiftool.org/XMP/XMP-aux/1.0/", + "DJI": "http://ns.exiftool.org/MakerNotes/DJI/1.0/", + "XMP-drone-dji": "http://ns.exiftool.org/XMP/XMP-drone-dji/1.0/", **{ f"Track{track_id}": f"http://ns.exiftool.org/QuickTime/Track{track_id}/1.0/" for track_id in range(1, MAX_TRACK_ID + 1) @@ -408,6 +414,52 @@ def extract_model(self) -> str | None: _, model = self._extract_make_and_model() return model + def extract_camera_uuid(self) -> str | None: + """ + Extract camera unique identifier from serial number tags in video metadata. + Builds a composite ID from body and lens serial numbers. + """ + # Try camera-specific serial numbers first + body_serial = self._extract_alternative_fields( + [ + # Camera-specific tags + "GoPro:SerialNumber", + "GoPro:CameraSerialNumber", + "Ricoh:SerialNumber", + "XMP-GSpherical:PiDeviceSN", # Labpano cameras + "Insta360:SerialNumber", + "DJI:SerialNumber", + "XMP-drone-dji:CameraSerialNumber", + "XMP-drone-dji:DroneSerialNumber", + # Generic tags + "ExifIFD:BodySerialNumber", + "ExifIFD:SerialNumber", + "IFD0:SerialNumber", + "UserData:SerialNumber", + "XMP-aux:SerialNumber", + "UserData:SerialNumberHash", + ], + str, + ) + lens_serial = self._extract_alternative_fields( + [ + "UserData:LensSerialNumber", + "ExifIFD:LensSerialNumber", + "XMP-aux:LensSerialNumber", + ], + str, + ) + + parts = [] + if body_serial: + parts.append(body_serial.strip()) + if lens_serial: + parts.append(lens_serial.strip()) + + if parts: + return "_".join(parts) + return None + def _extract_gps_track_from_track(self) -> list[GPSPoint]: root = self.etree.getroot() if root is None: diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index 4e20dcbef..c20d9f1a3 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -87,6 +87,7 @@ def to_description( if isinstance(metadata, types.ImageMetadata): metadata.MAPDeviceMake = video_metadata.make metadata.MAPDeviceModel = video_metadata.model + metadata.MAPCameraUUID = video_metadata.camera_uuid final_image_metadatas.extend(image_metadatas) diff --git a/mapillary_tools/geotag/image_extractors/exif.py b/mapillary_tools/geotag/image_extractors/exif.py index 252af6668..01470964f 100644 --- a/mapillary_tools/geotag/image_extractors/exif.py +++ b/mapillary_tools/geotag/image_extractors/exif.py @@ -60,6 +60,7 @@ def extract(self) -> types.ImageMetadata: MAPOrientation=exif.extract_orientation(), MAPDeviceMake=exif.extract_make(), MAPDeviceModel=exif.extract_model(), + MAPCameraUUID=exif.extract_camera_uuid(), ) return image_metadata diff --git a/mapillary_tools/geotag/video_extractors/exiftool.py b/mapillary_tools/geotag/video_extractors/exiftool.py index a3b8202b3..09971d7f6 100644 --- a/mapillary_tools/geotag/video_extractors/exiftool.py +++ b/mapillary_tools/geotag/video_extractors/exiftool.py @@ -31,6 +31,7 @@ def extract(self) -> types.VideoMetadata: make = exif.extract_make() model = exif.extract_model() + camera_uuid = exif.extract_camera_uuid() is_gopro = make is not None and make.upper() in ["GOPRO"] @@ -70,6 +71,7 @@ def extract(self) -> types.VideoMetadata: points=points, make=make, model=model, + camera_uuid=camera_uuid, ) return video_metadata diff --git a/mapillary_tools/process_sequence_properties.py b/mapillary_tools/process_sequence_properties.py index 7e53c5f25..f910d4a48 100644 --- a/mapillary_tools/process_sequence_properties.py +++ b/mapillary_tools/process_sequence_properties.py @@ -355,6 +355,7 @@ def _group_by_folder_and_camera( image_metadatas, lambda metadata: ( str(metadata.filename.parent), + metadata.MAPCameraUUID, metadata.MAPDeviceMake, metadata.MAPDeviceModel, metadata.width, diff --git a/mapillary_tools/serializer/description.py b/mapillary_tools/serializer/description.py index 1f86a3671..9f50611b2 100644 --- a/mapillary_tools/serializer/description.py +++ b/mapillary_tools/serializer/description.py @@ -89,6 +89,7 @@ class VideoDescription(_SharedDescription, total=False): MAPGPSTrack: Required[list[T.Sequence[float | int | None]]] MAPDeviceMake: str MAPDeviceModel: str + MAPCameraUUID: str class _ErrorObject(TypedDict, total=False): @@ -206,6 +207,10 @@ class ErrorDescription(TypedDict, total=False): "type": "string", "description": "Device model, e.g. HERO10 Black, DR900S-1CH, Insta360 Titan", }, + "MAPCameraUUID": { + "type": "string", + "description": "Camera unique identifier, typically derived from camera serial number", + }, }, "required": [ "MAPGPSTrack", @@ -402,6 +407,8 @@ def _as_video_desc(cls, metadata: VideoMetadata) -> VideoDescription: desc["MAPDeviceMake"] = metadata.make if metadata.model: desc["MAPDeviceModel"] = metadata.model + if metadata.camera_uuid: + desc["MAPCameraUUID"] = metadata.camera_uuid return desc @classmethod @@ -495,6 +502,7 @@ def _from_video_desc(cls, desc: VideoDescription) -> VideoMetadata: points=[PointEncoder.decode(entry) for entry in desc["MAPGPSTrack"]], make=desc.get("MAPDeviceMake"), model=desc.get("MAPDeviceModel"), + camera_uuid=desc.get("MAPCameraUUID"), ) diff --git a/mapillary_tools/types.py b/mapillary_tools/types.py index eafec81c4..a6f4bb9a6 100644 --- a/mapillary_tools/types.py +++ b/mapillary_tools/types.py @@ -77,6 +77,7 @@ class VideoMetadata: make: str | None = None model: str | None = None filesize: int | None = None + camera_uuid: str | None = None def update_md5sum(self) -> None: if self.md5sum is None: diff --git a/schema/image_description_schema.json b/schema/image_description_schema.json index 2415e3ffa..2172036fa 100644 --- a/schema/image_description_schema.json +++ b/schema/image_description_schema.json @@ -46,6 +46,10 @@ "type": "string", "description": "Device model, e.g. HERO10 Black, DR900S-1CH, Insta360 Titan" }, + "MAPCameraUUID": { + "type": "string", + "description": "Camera unique identifier, typically derived from camera serial number" + }, "filename": { "type": "string", "description": "Absolute path of the video" diff --git a/tests/unit/test_exifread.py b/tests/unit/test_exifread.py index 875c18276..4276ab958 100644 --- a/tests/unit/test_exifread.py +++ b/tests/unit/test_exifread.py @@ -263,3 +263,356 @@ def test_read_and_write(setup_data: py.path.local): actual = read.extract_gps_datetime() assert actual assert geo.as_unix_time(dt) == geo.as_unix_time(actual) + + +# Tests for extract_camera_uuid + + +class MockExifTag: + """Mock class for exifread tag values""" + + def __init__(self, values): + self.values = values + + +class TestExtractCameraUuidFromEXIF: + """Test extract_camera_uuid from EXIF tags""" + + def test_body_serial_only(self): + """Test with only body serial number present""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF BodySerialNumber": MockExifTag("ABC123"), + } + assert reader.extract_camera_uuid() == "ABC123" + + def test_lens_serial_only(self): + """Test with only lens serial number present""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF LensSerialNumber": MockExifTag("LNS456"), + } + assert reader.extract_camera_uuid() == "LNS456" + + def test_both_body_and_lens_serial(self): + """Test with both body and lens serial numbers present""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF BodySerialNumber": MockExifTag("BODY123"), + "EXIF LensSerialNumber": MockExifTag("LENS456"), + } + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + def test_no_serial_numbers(self): + """Test with no serial numbers present""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = {} + assert reader.extract_camera_uuid() is None + + def test_generic_serial_fallback(self): + """Test fallback to generic EXIF SerialNumber""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF SerialNumber": MockExifTag("GENERIC789"), + } + assert reader.extract_camera_uuid() == "GENERIC789" + + def test_makernote_serial_fallback(self): + """Test fallback to MakerNote SerialNumber""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "MakerNote SerialNumber": MockExifTag("MAKER123"), + } + assert reader.extract_camera_uuid() == "MAKER123" + + def test_body_serial_priority_over_generic(self): + """Test that BodySerialNumber takes priority over generic SerialNumber""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF BodySerialNumber": MockExifTag("BODY123"), + "EXIF SerialNumber": MockExifTag("GENERIC789"), + } + assert reader.extract_camera_uuid() == "BODY123" + + def test_whitespace_stripped(self): + """Test that whitespace is stripped from serial numbers""" + from mapillary_tools.exif_read import ExifReadFromEXIF + + reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) + reader.tags = { + "EXIF BodySerialNumber": MockExifTag(" BODY123 "), + "EXIF LensSerialNumber": MockExifTag(" LENS456 "), + } + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + +class TestExtractCameraUuidFromXMP: + """Test extract_camera_uuid from XMP tags""" + + def _create_xmp_reader(self, tags_dict: dict): + """Helper to create an ExifReadFromXMP with mocked tags""" + from mapillary_tools.exif_read import ExifReadFromXMP, XMP_NAMESPACES + import xml.etree.ElementTree as ET + + # Build a minimal XMP document + rdf_ns = XMP_NAMESPACES["rdf"] + xmp_xml = f''' + + + + + + """ + + etree = ET.ElementTree(ET.fromstring(xmp_xml)) + return ExifReadFromXMP(etree) + + def test_xmp_body_serial_only(self): + """Test XMP with only body serial number""" + reader = self._create_xmp_reader({"exifEX:BodySerialNumber": "XMP_BODY123"}) + assert reader.extract_camera_uuid() == "XMP_BODY123" + + def test_xmp_lens_serial_only(self): + """Test XMP with only lens serial number""" + reader = self._create_xmp_reader({"exifEX:LensSerialNumber": "XMP_LENS456"}) + assert reader.extract_camera_uuid() == "XMP_LENS456" + + def test_xmp_both_serials(self): + """Test XMP with both body and lens serial numbers""" + reader = self._create_xmp_reader( + { + "exifEX:BodySerialNumber": "XMP_BODY", + "exifEX:LensSerialNumber": "XMP_LENS", + } + ) + assert reader.extract_camera_uuid() == "XMP_BODY_XMP_LENS" + + def test_xmp_no_serials(self): + """Test XMP with no serial numbers""" + reader = self._create_xmp_reader({}) + assert reader.extract_camera_uuid() is None + + def test_xmp_aux_serial_number(self): + """Test XMP with aux:SerialNumber (Adobe auxiliary namespace)""" + reader = self._create_xmp_reader({"aux:SerialNumber": "AUX_SERIAL123"}) + assert reader.extract_camera_uuid() == "AUX_SERIAL123" + + def test_xmp_aux_lens_serial_number(self): + """Test XMP with aux:LensSerialNumber""" + reader = self._create_xmp_reader({"aux:LensSerialNumber": "AUX_LENS456"}) + assert reader.extract_camera_uuid() == "AUX_LENS456" + + +class TestExtractCameraUuidIntegration: + """Integration tests using real image file""" + + def test_real_image_camera_uuid(self): + """Test extract_camera_uuid on test image (likely returns None as test image may not have serial)""" + exif_data = ExifRead(TEST_EXIF_FILE) + # The test image likely doesn't have serial numbers, so we just verify it doesn't crash + result = exif_data.extract_camera_uuid() + assert result is None or isinstance(result, str) + + +class TestVideoExtractCameraUuid: + """Test extract_camera_uuid for video EXIF reader""" + + def _create_video_exif_reader(self, tags_dict: dict): + """Helper to create an ExifToolReadVideo with mocked tags""" + from mapillary_tools.exiftool_read_video import ( + ExifToolReadVideo, + EXIFTOOL_NAMESPACES, + ) + import xml.etree.ElementTree as ET + + # Build XML with child elements (not attributes) - this is how ExifTool XML works + root = ET.Element( + "rdf:RDF", {"xmlns:rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#"} + ) + + # Add child elements for each tag + for key, value in tags_dict.items(): + prefix, tag_name = key.split(":") + if prefix in EXIFTOOL_NAMESPACES: + full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name + child = ET.SubElement(root, full_tag) + child.text = value + + etree = ET.ElementTree(root) + return ExifToolReadVideo(etree) + + def test_gopro_serial(self): + """Test extraction of GoPro serial number""" + reader = self._create_video_exif_reader( + {"GoPro:SerialNumber": "C3456789012345"} + ) + assert reader.extract_camera_uuid() == "C3456789012345" + + def test_insta360_serial(self): + """Test extraction of Insta360 serial number""" + reader = self._create_video_exif_reader( + {"Insta360:SerialNumber": "INST360SERIAL"} + ) + assert reader.extract_camera_uuid() == "INST360SERIAL" + + def test_exif_body_serial(self): + """Test extraction of standard EXIF body serial number""" + reader = self._create_video_exif_reader({"ExifIFD:BodySerialNumber": "BODY123"}) + assert reader.extract_camera_uuid() == "BODY123" + + def test_exif_body_and_lens_serial(self): + """Test extraction of both body and lens serial numbers""" + reader = self._create_video_exif_reader( + { + "ExifIFD:BodySerialNumber": "BODY123", + "ExifIFD:LensSerialNumber": "LENS456", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + def test_no_serial(self): + """Test with no serial numbers present""" + reader = self._create_video_exif_reader({}) + assert reader.extract_camera_uuid() is None + + def test_gopro_priority(self): + """Test that GoPro serial takes priority over generic serial""" + reader = self._create_video_exif_reader( + { + "GoPro:SerialNumber": "GOPRO123", + "IFD0:SerialNumber": "GENERIC789", + } + ) + assert reader.extract_camera_uuid() == "GOPRO123" + + +class TestExifToolReadExtractCameraUuid: + """Test extract_camera_uuid for ExifToolRead (image EXIF via ExifTool XML)""" + + def _create_exiftool_reader(self, tags_dict: dict): + """Helper to create an ExifToolRead with mocked tags""" + from mapillary_tools.exiftool_read import ExifToolRead, EXIFTOOL_NAMESPACES + import xml.etree.ElementTree as ET + + # Build XML structure that ExifToolRead expects + root = ET.Element("rdf:Description") + + for tag, value in tags_dict.items(): + prefix, tag_name = tag.split(":", 1) + if prefix in EXIFTOOL_NAMESPACES: + full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name + child = ET.SubElement(root, full_tag) + child.text = value + + etree = ET.ElementTree(root) + return ExifToolRead(etree) + + def test_body_serial_only(self): + """Test extraction with only body serial number""" + reader = self._create_exiftool_reader({"ExifIFD:BodySerialNumber": "BODY12345"}) + assert reader.extract_camera_uuid() == "BODY12345" + + def test_lens_serial_only(self): + """Test extraction with only lens serial number""" + reader = self._create_exiftool_reader({"ExifIFD:LensSerialNumber": "LENS67890"}) + assert reader.extract_camera_uuid() == "LENS67890" + + def test_both_body_and_lens_serial(self): + """Test extraction with both body and lens serial numbers""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": "BODY123", + "ExifIFD:LensSerialNumber": "LENS456", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + def test_no_serial_numbers(self): + """Test with no serial numbers present""" + reader = self._create_exiftool_reader({}) + assert reader.extract_camera_uuid() is None + + def test_generic_serial_fallback(self): + """Test that ExifIFD:SerialNumber is used as fallback for body serial""" + reader = self._create_exiftool_reader({"ExifIFD:SerialNumber": "GENERIC123"}) + assert reader.extract_camera_uuid() == "GENERIC123" + + def test_ifd0_serial_fallback(self): + """Test that IFD0:SerialNumber is used as fallback""" + reader = self._create_exiftool_reader({"IFD0:SerialNumber": "IFD0_SN_123"}) + assert reader.extract_camera_uuid() == "IFD0_SN_123" + + def test_body_serial_priority_over_generic(self): + """Test that BodySerialNumber takes priority over generic SerialNumber""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": "BODY999", + "ExifIFD:SerialNumber": "GENERIC888", + } + ) + assert reader.extract_camera_uuid() == "BODY999" + + def test_xmp_exifex_body_serial(self): + """Test XMP-exifEX:BodySerialNumber extraction""" + reader = self._create_exiftool_reader( + {"XMP-exifEX:BodySerialNumber": "XMPBODY123"} + ) + assert reader.extract_camera_uuid() == "XMPBODY123" + + def test_xmp_aux_serial(self): + """Test XMP-aux:SerialNumber extraction""" + reader = self._create_exiftool_reader({"XMP-aux:SerialNumber": "AUX_SN_456"}) + assert reader.extract_camera_uuid() == "AUX_SN_456" + + def test_xmp_aux_lens_serial(self): + """Test XMP-aux:LensSerialNumber extraction""" + reader = self._create_exiftool_reader( + {"XMP-aux:LensSerialNumber": "AUX_LENS_789"} + ) + assert reader.extract_camera_uuid() == "AUX_LENS_789" + + def test_xmp_combined(self): + """Test XMP body and lens serial combined""" + reader = self._create_exiftool_reader( + { + "XMP-exifEX:BodySerialNumber": "XMP_BODY", + "XMP-exifEX:LensSerialNumber": "XMP_LENS", + } + ) + assert reader.extract_camera_uuid() == "XMP_BODY_XMP_LENS" + + def test_whitespace_stripped(self): + """Test that whitespace is stripped from serial numbers""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": " BODY123 ", + "ExifIFD:LensSerialNumber": " LENS456 ", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" diff --git a/tests/unit/test_sequence_processing.py b/tests/unit/test_sequence_processing.py index 4b64eae72..ebcfa6e70 100644 --- a/tests/unit/test_sequence_processing.py +++ b/tests/unit/test_sequence_processing.py @@ -176,6 +176,100 @@ def test_find_sequences_by_camera(tmpdir: py.path.local): assert len(uuids) == 3 +def test_find_sequences_by_camera_uuid(tmpdir: py.path.local): + """Test that images are grouped by MAPCameraUUID when available.""" + curdir = tmpdir.mkdir("camera_uuid_test") + sequence: T.List[types.MetadataOrError] = [ + # s1 - camera with UUID "CAMERA_A" + _make_image_metadata( + Path(curdir) / Path("img1.jpg"), + 1.00001, + 1.00001, + 1, + 11, + MAPDeviceMake="Canon", + MAPDeviceModel="EOS R5", + MAPCameraUUID="CAMERA_A_SERIAL", + width=1920, + height=1080, + ), + _make_image_metadata( + Path(curdir) / Path("img2.jpg"), + 1.00002, + 1.00002, + 2, + 22, + MAPDeviceMake="Canon", + MAPDeviceModel="EOS R5", + MAPCameraUUID="CAMERA_A_SERIAL", + width=1920, + height=1080, + ), + # s2 - different camera with UUID "CAMERA_B" but same make/model + _make_image_metadata( + Path(curdir) / Path("img3.jpg"), + 1.00003, + 1.00003, + 3, + 33, + MAPDeviceMake="Canon", + MAPDeviceModel="EOS R5", + MAPCameraUUID="CAMERA_B_SERIAL", + width=1920, + height=1080, + ), + _make_image_metadata( + Path(curdir) / Path("img4.jpg"), + 1.00004, + 1.00004, + 4, + 44, + MAPDeviceMake="Canon", + MAPDeviceModel="EOS R5", + MAPCameraUUID="CAMERA_B_SERIAL", + width=1920, + height=1080, + ), + # s3 - camera without UUID (should be grouped separately from cameras with UUIDs) + _make_image_metadata( + Path(curdir) / Path("img5.jpg"), + 1.00005, + 1.00005, + 5, + 55, + MAPDeviceMake="Canon", + MAPDeviceModel="EOS R5", + MAPCameraUUID=None, + width=1920, + height=1080, + ), + ] + metadatas = psp.process_sequence_properties( + sequence, + cutoff_distance=1000000, + cutoff_time=10000, + interpolate_directions=False, + duplicate_distance=0, + duplicate_angle=0, + ) + image_metadatas = [d for d in metadatas if isinstance(d, types.ImageMetadata)] + + # Group by sequence UUID to verify the sequences + sequences_by_uuid: T.Dict[str, T.List[types.ImageMetadata]] = {} + for d in image_metadatas: + sequences_by_uuid.setdefault(d.MAPSequenceUUID or "", []).append(d) + + # Should have 3 sequences: CAMERA_A, CAMERA_B, and None + assert len(sequences_by_uuid) == 3 + + # Verify each sequence has images from only one camera + for seq in sequences_by_uuid.values(): + camera_uuids = set(img.MAPCameraUUID for img in seq) + assert len(camera_uuids) == 1, ( + f"Sequence contains images from multiple cameras: {camera_uuids}" + ) + + def test_sequences_sorted(tmpdir: py.path.local): curdir = tmpdir.mkdir("hello1").mkdir("world2") sequence: T.List[types.ImageMetadata] = [ diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index ab84a85df..987b68ec9 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -73,6 +73,15 @@ def test_desc_video(): filetype=types.FileType.CAMM, points=[], ), + types.VideoMetadata( + filename=Path("foo/bar.mp4").resolve(), + md5sum="789", + filetype=types.FileType.GOPRO, + points=[geo.Point(time=456, lat=2.0, lon=3.0, alt=100.0, angle=45)], + make="GoPro", + model="HERO10", + camera_uuid="ABC123_XYZ789", + ), ] for metadata in ds: desc = description.DescriptionJSONSerializer._as_video_desc(metadata)