Ensure positions if vessels in multiple sources are not discarded

GlobalFishingWatch · Jul 8, 2024 · 5f9dd6d · 5f9dd6d
1 parent 6c16b8f
commit 5f9dd6d
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 7 deletions.
diff --git a/.../pipe-vms-ingestion/tests/vms_ingestion/normalization/transforms/test_deduplicate_msgs.py b/.../pipe-vms-ingestion/tests/vms_ingestion/normalization/transforms/test_deduplicate_msgs.py
@@ -36,6 +36,7 @@ def test_convert_speed_kph_to_kt(self):
                         "lon": 10,
                         "shipname": "",
                         "received_at": None,
+                        "source": "country_fleet_1",
                         "ssvid": "1",
                     },
                     {
@@ -46,6 +47,7 @@ def test_convert_speed_kph_to_kt(self):
                         "lon": 10,
                         "shipname": "SANTA MARIA",
                         "received_at": None,
+                        "source": "country_fleet_1",
                         "ssvid": "1",
                     },
                     {
@@ -58,6 +60,7 @@ def test_convert_speed_kph_to_kt(self):
                         "received_at": datetime.strptime(
                             "2024-01-01T00:00:01", "%Y-%m-%dT%H:%M:%S"
                         ),
+                        "source": "country_fleet_1",
                         "ssvid": "1",
                     },
                     {
@@ -70,6 +73,20 @@ def test_convert_speed_kph_to_kt(self):
                         "received_at": datetime.strptime(
                             "2024-01-01T00:10:00", "%Y-%m-%dT%H:%M:%S"
                         ),
+                        "source": "country_fleet_1",
+                        "ssvid": "1",
+                    },
+                    {
+                        "timestamp": datetime.strptime(
+                            "2024-01-01T00:00:00", "%Y-%m-%dT%H:%M:%S"
+                        ),
+                        "lat": 10,
+                        "lon": 10,
+                        "shipname": "SANTA MARIA",
+                        "received_at": datetime.strptime(
+                            "2024-01-01T00:10:00", "%Y-%m-%dT%H:%M:%S"
+                        ),
+                        "source": "country_fleet_2",
                         "ssvid": "1",
                     },
                 ]
@@ -84,17 +101,23 @@ def test_convert_speed_kph_to_kt(self):
                 pcol_equal_to(
                     [
                         {
-                            "timestamp": datetime.strptime(
-                                "2024-01-01T00:00:00", "%Y-%m-%dT%H:%M:%S"
-                            ),
                             "lat": 10,
                             "lon": 10,
+                            "received_at": datetime(2024, 1, 1, 0, 10),
+                            "shipname": "SANTA MARIA",
+                            "source": "country_fleet_1",
+                            "ssvid": "1",
+                            "timestamp": datetime(2024, 1, 1, 0, 0),
+                        },
+                        {
+                            "lat": 10,
+                            "lon": 10,
+                            "received_at": datetime(2024, 1, 1, 0, 10),
                             "shipname": "SANTA MARIA",
-                            "received_at": datetime.strptime(
-                                "2024-01-01T00:10:00", "%Y-%m-%dT%H:%M:%S"
-                            ),
+                            "source": "country_fleet_2",
                             "ssvid": "1",
-                        }
+                            "timestamp": datetime(2024, 1, 1, 0, 0),
+                        },
                     ]
                 ),
                 label="CheckOutput",

diff --git a/packages/pipe-vms-ingestion/vms_ingestion/normalization/transforms/deduplicate_msgs.py b/packages/pipe-vms-ingestion/vms_ingestion/normalization/transforms/deduplicate_msgs.py
@@ -34,6 +34,7 @@ def group_by_ssvid_timestamp(self):
         return GroupBy(
             id=lambda message: message["ssvid"],
             timestamp=lambda message: message["timestamp"].isoformat(),
+            source=lambda message: message["source"],
         )
 
     def prioritize_msg_with_shipname(self):