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)