diff --git a/src/analyzer/moving_pandas_trajectory_collection/analyzer.py b/src/analyzer/moving_pandas_trajectory_collection/analyzer.py index b71ea0b..0690872 100644 --- a/src/analyzer/moving_pandas_trajectory_collection/analyzer.py +++ b/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 diff --git a/src/moveapps_pickle.py b/src/moveapps_pickle.py new file mode 100644 index 0000000..d4f1356 --- /dev/null +++ b/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 diff --git a/test/analyzer/moving_pandas_trajectory_collection/test_analyzer.py b/test/analyzer/moving_pandas_trajectory_collection/test_analyzer.py index 5970871..42d58eb 100644 --- a/test/analyzer/moving_pandas_trajectory_collection/test_analyzer.py +++ b/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) diff --git a/test/config/fixtures.py b/test/config/fixtures.py new file mode 100644 index 0000000..8c0ae53 --- /dev/null +++ b/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 diff --git a/test/test_moveapps_pickle.py b/test/test_moveapps_pickle.py new file mode 100644 index 0000000..ad5e201 --- /dev/null +++ b/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)