data_extractor_test.py

import unittest
from datetime import datetime, timedelta

import geopandas as gpd
import pandas as pd
from shapely.geometry import Point

from context import measprocess as mpc


class TestRTRDetailExtractor(unittest.TestCase):
    def setUp(self):
        self._open_test_uuids = [
            "Ob6c34648-54f3-435c-b5f9-5677c8694ad9",
            "Ocf041649-977f-4957-a27f-215069579c18",
        ]

    def test_basics(self):
        details = mpc.rtr.fetch.rtr_details(self._open_test_uuids)
        self.assertTrue(type(details[0]) == dict)


class TestRTROverviewExtractor(unittest.TestCase):
    def setUp(self):
        self._max_time = datetime.fromtimestamp(1618987288)
        self._min_time = self._max_time - timedelta(days=1)

        self._coords = [(48.201914, 16.363859), (48.194170, 16.385466)]

        self._boundaries = gpd.GeoSeries(
            (Point(lon, lat) for lat, lon in self._coords)
        ).set_crs("EPSG:4326")

    def test_time(self):
        results = mpc.rtr.fetch.rtr_overview(
            time_min=self._min_time, time_max=self._max_time, max_results=5000, verbose=True
        )

        self.assertTrue(len(results) > 0)

    def test_cat_tech(self):
        results = mpc.rtr.fetch.rtr_overview(
            time_min=self._min_time,
            time_max=self._max_time,
            cat_technology="4G",
            max_results=5000,
        )

        self.assertTrue(len(results) > 0)

    def test_coords(self):
        results = mpc.rtr.fetch.rtr_overview(
            time_min=self._min_time,
            time_max=self._max_time,
            gps_boundaries=self._boundaries,
            max_results=5000,
        )

        self.assertTrue(len(results) > 0)

    def test_coords_error(self):
        coords = [(48.173759, 16.367682), (48.181447, 16.379107)]
        boundaries = gpd.GeoSeries((Point(lon, lat) for lat, lon in coords)).set_crs(
            "EPSG:4326"
        )

        min_time = self._max_time - timedelta(days=30)

        results = mpc.rtr.fetch.rtr_overview(
            gps_boundaries=boundaries,
            time_min=min_time,
            time_max=self._max_time,
            max_results=5000,
        )

        long_min, lat_min, long_max, lat_max = tuple(boundaries.total_bounds)

        self.assertTrue(
            (results["lat"].min() >= lat_min)
            and (results["long"].min() >= long_min)
            and (results["lat"].max() <= lat_max)
            and (results["long"].max() <= long_max)
        )