Read gzip-compressed App output

src/analyzer/moving_pandas_trajectory_collection/analyzer.py

@@ -1,8 +1,8 @@
 from geopandas import GeoDataFrame
 
 from src.analyzer.base_analyzer import BaseAnalyzer
+from src.moveapps_pickle import MoveAppsPickle
 
-import pandas as pd
 import geopandas as gpd
 import movingpandas as mpd
 import logging
@@ -23,7 +23,7 @@ def analyze(self, path: str) -> list:
             }]
 
     def __read(self, path: str) -> mpd.TrajectoryCollection:
-        movingpandas = pd.read_pickle(path)
+        movingpandas = MoveAppsPickle.read(path=path)
         logging.info(f'read movingpandas: {movingpandas}')
         return movingpandas
 

src/moveapps_pickle.py

@@ -0,0 +1,43 @@
+import logging
+
+import pandas as pd
+
+GZIP_MAGIC = b'\x1f\x8b'
+
+
+class MoveAppsPickle:
+    """
+    Reads the pickle an App produced.
+
+    MoveApps dictates the file path and it carries no extension, so pandas cannot infer the
+    compression from the file name. It is determined from the file content instead.
+
+    Deliberately read-only: the cargo-agent never writes a pickle, it writes JSON statistics.
+    """
+
+    @staticmethod
+    def read(path: str):
+        """
+        Reads a pickle, compressed or not.
+
+        Apps built against a Python SDK before `v3` write uncompressed pickles, and such output still
+        reaches the cargo-agent. Both flavours have to stay readable.
+
+        :param path: path to the pickle to read
+        :return: the unpickled data
+        """
+        compression = MoveAppsPickle.detect_compression(path=path)
+        logging.info(f'reading pickle \'{path}\' (compression: {compression})')
+        return pd.read_pickle(path, compression=compression)
+
+    @staticmethod
+    def detect_compression(path: str) -> str | None:
+        """
+        Determines the compression of a pickle from its first bytes instead of from its name.
+
+        :param path: path to the pickle to inspect
+        :return: `'gzip'` for a gzip-compressed file, otherwise `None`, which is how pandas is told
+            not to decompress at all
+        """
+        with open(path, 'rb') as probe:
+            return 'gzip' if probe.read(len(GZIP_MAGIC)) == GZIP_MAGIC else None

test/analyzer/moving_pandas_trajectory_collection/test_analyzer.py

@@ -1,13 +1,19 @@
 import os.path
+import tempfile
 from unittest import TestCase
 from src.analyzer.moving_pandas_trajectory_collection.analyzer import MovingPandasAnalyzer
 from test.config.definitions import ROOT_DIR
+from test.config.fixtures import gzip_copy
 
 
 class TestMovingPandasAnalyzer(TestCase):
 
     def setUp(self) -> None:
         self.sut = MovingPandasAnalyzer()
+        self.tmp = tempfile.TemporaryDirectory()
+
+    def tearDown(self) -> None:
+        self.tmp.cleanup()
 
     def test_it_should_handle_empty_input(self):
         # execute
@@ -54,5 +60,23 @@ def test_it_should_analyze_input4_LatLon(self):
         self.assertEqual(actual[11]['data_attributes'], ['sensor_type_id', 'barometric_pressure', 'data_decoding_software', 'eobs_activity', 'eobs_activity_samples', 'eobs_battery_voltage', 'eobs_fix_battery_voltage', 'eobs_horizontal_accuracy_estimate', 'eobs_key_bin_checksum', 'eobs_speed_accuracy_estimate', 'eobs_start_timestamp', 'eobs_status', 'eobs_temperature', 'eobs_type_of_fix', 'eobs_used_time_to_get_fix', 'gps_dop', 'gps_satellite_count', 'ground_speed', 'gt_tx_count', 'heading', 'height_above_ellipsoid', 'height_raw', 'import_marked_outlier', 'manually_marked_outlier', 'tag_voltage', 'timestamp', 'event_id', 'visible', 'individual_name_deployment_id', 'timestamp_tz', 'geometry'])
         self.assertEqual(actual[12]['n'], ['non-empty-result'])
 
+    def test_it_should_analyze_a_gzip_compressed_input4_LatLon(self):
+        # prepare: an App on Python SDK v3 writes gzip-compressed output, under a file name that
+        # carries no extension
+        compressed = gzip_copy(
+            self.__test_file("input4_LatLon.pickle"),
+            os.path.join(self.tmp.name, "output_file")
+        )
+
+        # execute
+        actual = self.sut.analyze(path=compressed)
+
+        # verify: the compression makes no difference to the statistics
+        self.assertEqual(actual[0]['positions_total_number'], 7770)
+        self.assertEqual(actual[1]['timestamps_range'], ['2013-08-07 18:27:02', '2015-02-19 13:42:27'])
+        self.assertEqual(actual[8]['tracks_total_number'], 3)
+        self.assertEqual(actual[10]['number_positions_by_track'], {'X742..deploy_id.56853924.': 3265, 'X746..deploy_id.20813885.': 3010, 'X749..deploy_id.20813892.': 1495})
+        self.assertEqual(actual[12]['n'], ['non-empty-result'])
+
     def __test_file(self, file_name) -> str:
         return os.path.join(ROOT_DIR, 'test', 'resources', 'moving_pandas_trajectory_collection', file_name)

test/config/fixtures.py

@@ -0,0 +1,18 @@
+import gzip
+import shutil
+
+
+def gzip_copy(source: str, target: str) -> str:
+    """
+    Writes a gzip-compressed copy of `source` to `target`.
+
+    Lets a test simulate the output of an App on Python SDK v3 without adding a second set of binary
+    fixtures to the repository. The pickle payload itself is copied byte for byte.
+
+    :param source: path to the uncompressed file
+    :param target: path to write the compressed copy to
+    :return: `target`, for convenience
+    """
+    with open(source, 'rb') as plain, gzip.open(target, 'wb') as compressed:
+        shutil.copyfileobj(plain, compressed)
+    return target

test/test_moveapps_pickle.py

@@ -0,0 +1,67 @@
+import os
+import tempfile
+from unittest import TestCase
+
+from src.moveapps_pickle import MoveAppsPickle
+from test.config.definitions import ROOT_DIR
+from test.config.fixtures import gzip_copy
+
+
+class TestMoveAppsPickle(TestCase):
+    """
+    MoveApps hands the cargo-agent a path without a file extension, so none of these cases may rely
+    on the file name to tell compressed from uncompressed data.
+    """
+
+    def setUp(self) -> None:
+        self.sut = MoveAppsPickle()
+        self.tmp = tempfile.TemporaryDirectory()
+
+    def tearDown(self) -> None:
+        self.tmp.cleanup()
+
+    def test_it_should_detect_an_uncompressed_pickle(self):
+        # prepare
+        legacy = self.__fixture('input4_LatLon.pickle')
+
+        # execute
+        actual = self.sut.detect_compression(path=legacy)
+
+        # verify: `None` is how pandas is told not to decompress at all
+        self.assertIsNone(actual)
+
+    def test_it_should_detect_a_gzip_compressed_pickle(self):
+        # prepare
+        compressed = gzip_copy(self.__fixture('input4_LatLon.pickle'), self.__tmp_file('output_file'))
+
+        # execute
+        actual = self.sut.detect_compression(path=compressed)
+
+        # verify
+        self.assertEqual('gzip', actual)
+
+    def test_it_should_read_an_uncompressed_pickle(self):
+        # prepare
+        legacy = self.__fixture('input4_LatLon.pickle')
+
+        # execute
+        actual = self.sut.read(path=legacy)
+
+        # verify
+        self.assertEqual(3, len(actual.trajectories))
+
+    def test_it_should_read_a_gzip_compressed_pickle_named_without_an_extension(self):
+        # prepare
+        compressed = gzip_copy(self.__fixture('input4_LatLon.pickle'), self.__tmp_file('output_file'))
+
+        # execute
+        actual = self.sut.read(path=compressed)
+
+        # verify
+        self.assertEqual(3, len(actual.trajectories))
+
+    def __fixture(self, file_name: str) -> str:
+        return os.path.join(ROOT_DIR, 'test', 'resources', 'moving_pandas_trajectory_collection', file_name)
+
+    def __tmp_file(self, file_name: str) -> str:
+        return os.path.join(self.tmp.name, file_name)