dynamic_time_warping_test.py

import unittest
import geopandas as gpd
from context import measprocess as mpc
import pandas as pd
import numpy as np

from shapely.geometry import LineString, Point

class TestDTW(unittest.TestCase):
    def setUp(self):
        '''
        Note, that the Dummy Example does not use coordinates but x, y positions
        Still we transform it here to be consistent with the default use-case
        '''

        #Prepare Reference Path
        rp = pd.read_csv("tests/example_files/dtw_example/reference_path.csv", index_col=0)
        rp = gpd.GeoSeries(
            gpd.points_from_xy(rp.x_m, rp.y_m)
        ).set_crs("EPSG:31287").to_crs("EPSG:4326")
        self._rp = rp.drop(columns=["x_m, y_m"])

        #Prepare Measurements
        measurement_files = (
            pd.read_csv(f"tests/example_files/dtw_example/measurement_{i}.csv", index_col=0)
            for i in range(3)
        )
        self._measurements = [
            gpd.GeoDataFrame(
                meas,
                geometry=gpd.points_from_xy(meas.x_m, meas.y_m)
            ).set_crs("EPSG:31287").to_crs("EPSG:4326").drop(columns=["x_m", "y_m"]) for meas in measurement_files
        ]

    def test_epsg_exception(self):
        rp = self._rp
        rp.crs = None

        measurements = self._measurements
        for meas in measurements:
            meas.crs = None

        with self.assertRaises(ValueError) as context:
            mpc.dtw.warp_geodataframe(
                rp, measurements
            )

    def test_original_index(self):
        #Check the no measurement was lost
        measurement = self._measurements[0]
        warped_measurement = mpc.dtw.warp_geodataframe(
            self._rp, [measurement],
            resampling_distance=1
        )[0]

        self.assertTrue(
            all(i in warped_measurement.original_index for i in measurement.index)
        )

    def test_basic_example_resampling(self):
        measurement_list = mpc.dtw.warp_geodataframe(
            self._rp, self._measurements,
            resampling_distance=1
        )

        self.assertTrue(
            ((measurement_list[0].geometry == measurement_list[1].geometry) & (measurement_list[1].geometry == measurement_list[2].geometry)).all()
        )

    def test_basic_example(self):
        measurement_list = mpc.dtw.warp_geodataframe(
            self._rp, self._measurements,
            resampling_distance=None
        )

        self.assertTrue(
            ((measurement_list[0].geometry == measurement_list[1].geometry) & (measurement_list[1].geometry == measurement_list[2].geometry)).all()
        )

    def test_linestring_input(self):
        rp = gpd.GeoSeries(LineString(self._rp)).set_crs("EPSG:4326")

        measurement_list = mpc.dtw.warp_geodataframe(
            rp, self._measurements,
            resampling_distance=1
        )

        self.assertTrue(
            ((measurement_list[0].geometry == measurement_list[1].geometry) & (measurement_list[1].geometry == measurement_list[2].geometry)).all()
        )

    def test_linestring_input_warning(self):
        rp = gpd.GeoSeries(LineString(self._rp)).set_crs("EPSG:4326")

        measurement_list = mpc.dtw.warp_geodataframe(
            rp, self._measurements,
            resampling_distance=1
        )

        self.assertTrue(
            ((measurement_list[0].geometry == measurement_list[1].geometry) & (measurement_list[1].geometry == measurement_list[2].geometry)).all()
        )

    def test_index_warning(self):
        rp = gpd.GeoSeries(LineString(self._rp)).set_crs("EPSG:4326")

        with self.assertRaises(IndexError) as context:
            measurement_list = mpc.dtw.warp_geodataframe(
                rp, self._measurements,
                resampling_distance=5
            )