161 use static data as additional data input for kienzler

src/parkapi_sources/converters/kienzler/converter.py

@@ -9,36 +9,59 @@
 from validataclass.exceptions import ValidationError
 from validataclass.validators import AnythingValidator, DataclassValidator, ListValidator
 
-from parkapi_sources.converters.base_converter.pull import PullConverter
+from parkapi_sources.converters.base_converter.pull import PullConverter, StaticGeojsonDataMixin
 from parkapi_sources.exceptions import ImportParkingSiteException
 from parkapi_sources.models import RealtimeParkingSiteInput, SourceInfo, StaticParkingSiteInput
 
-from .models import KienzlerInput
+from .models import KienzlerGeojsonFeatureInput, KienzlerInput
 
 
-class KienzlerBasePullConverter(PullConverter):
+class KienzlerBasePullConverter(PullConverter, StaticGeojsonDataMixin):
     kienzler_list_validator = ListValidator(AnythingValidator(allowed_types=[dict]))
     kienzler_item_validator = DataclassValidator(KienzlerInput)
+    kienzler_geojson_item_validator = DataclassValidator(KienzlerInput)
+    use_geojson = False
 
     @property
     @abstractmethod
     def config_prefix(self):
         pass
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, use_geojson=False, **kwargs):
         super().__init__(*args, **kwargs)
         self.required_config_keys = [
             f'PARK_API_KIENZLER_{self.config_prefix}_USER',
             f'PARK_API_KIENZLER_{self.config_prefix}_PASSWORD',
             f'PARK_API_KIENZLER_{self.config_prefix}_IDS',
         ]
+        self.use_geojson = use_geojson
+        self.geojson_feature_validator = DataclassValidator(KienzlerGeojsonFeatureInput)
 
     def get_static_parking_sites(self) -> tuple[list[StaticParkingSiteInput], list[ImportParkingSiteException]]:
-        static_parking_site_inputs: list[StaticParkingSiteInput] = []
-
         kienzler_parking_sites, static_parking_site_errors = self._get_kienzler_parking_sites()
-        for kienzler_parking_site in kienzler_parking_sites:
-            static_parking_site_inputs.append(kienzler_parking_site.to_static_parking_site(self.source_info.public_url))
+        static_parking_site_inputs = [
+            kienzler_parking_site.to_static_parking_site(self.source_info.public_url)
+            for kienzler_parking_site in kienzler_parking_sites
+        ]
+
+        if self.use_geojson:
+            geojson_parking_sites, geojson_parking_site_errors = self._get_static_geojson_parking_sites()
+
+            static_geojson_parking_site_inputs_by_uid: dict[str, KienzlerGeojsonFeatureInput] = {}
+            for geojson_parking_site in geojson_parking_sites:
+                static_geojson_parking_site_inputs_by_uid[geojson_parking_site.uid] = geojson_parking_site
+
+            static_parking_site_errors += geojson_parking_site_errors
+
+            # For each GeoJSON entry, extend its correspondent static data
+            for kienzler_parking_site in static_parking_site_inputs:
+                # If the uid is not known in our static data: ignore the GeoJSON data
+                parking_site_uid = str(kienzler_parking_site.uid)
+                if parking_site_uid not in static_geojson_parking_site_inputs_by_uid:
+                    continue
+
+                # Extend static data with GeoJSON data
+                kienzler_parking_site.from_dict(static_geojson_parking_site_inputs_by_uid[parking_site_uid].to_dict())
 
         return static_parking_site_inputs, static_parking_site_errors
 
@@ -88,6 +111,16 @@ def _request(self) -> list[dict]:
 
         return result_dicts
 
+    def _get_static_geojson_parking_sites(
+        self,
+    ) -> tuple[list[KienzlerGeojsonFeatureInput], list[ImportParkingSiteException]]:
+        static_parking_site_inputs, import_parking_site_exceptions = (
+            self._get_static_parking_site_inputs_and_exceptions(
+                source_uid=self.source_info.uid,
+            )
+        )
+        return static_parking_site_inputs, import_parking_site_exceptions
+
 
 class KienzlerBikeAndRidePullConverter(KienzlerBasePullConverter):
     config_prefix = 'BIKE_AND_RIDE'

src/parkapi_sources/converters/kienzler/models.py

@@ -5,12 +5,24 @@
 
 from datetime import datetime, timezone
 from decimal import Decimal
+from typing import Optional
 
-from validataclass.dataclasses import validataclass
-from validataclass.validators import IntegerValidator, NumericValidator, StringValidator
+from validataclass.dataclasses import Default, DefaultUnset, ValidataclassMixin, validataclass
+from validataclass.helpers import OptionalUnsetNone
+from validataclass.validators import (
+    DataclassValidator,
+    EnumValidator,
+    IntegerValidator,
+    ListValidator,
+    Noneable,
+    NumericValidator,
+    StringValidator,
+)
 
+from parkapi_sources.converters.base_converter.pull.static_geojson_data_mixin.models import GeojsonFeatureInput
 from parkapi_sources.models import RealtimeParkingSiteInput, StaticParkingSiteInput
-from parkapi_sources.models.enums import ParkingSiteType, PurposeType
+from parkapi_sources.models.enums import ExternalIdentifierType, ParkAndRideType, ParkingSiteType, PurposeType
+from parkapi_sources.models.parking_site_inputs import BaseParkingSiteInput, ExternalIdentifierInput
 
 
 @validataclass
@@ -45,3 +57,60 @@ def to_realtime_parking_site(self) -> RealtimeParkingSiteInput:
             realtime_capacity=self.sum_boxes,
             realtime_free_capacity=self.bookable,
         )
+
+
+@validataclass
+class ExternalIdentifier(ValidataclassMixin):
+    type: ExternalIdentifierType = EnumValidator(ExternalIdentifierType)
+    value: str = StringValidator()
+
+
+@validataclass
+class KienzlerGeojsonFeaturePropertiesInput(ValidataclassMixin):
+    uid: str = StringValidator(min_length=1, max_length=256)
+    address: Optional[str] = StringValidator(max_length=512), Default(None)
+    type: Optional[ParkingSiteType] = EnumValidator(ParkingSiteType), Default(None)
+    max_height: int = IntegerValidator(min_value=0)
+    max_width: int = IntegerValidator(min_value=0)
+    max_depth: int = IntegerValidator(min_value=0)
+    park_and_ride_type: list[ParkAndRideType] = ListValidator(EnumValidator(ParkAndRideType))
+    external_identifiers: Optional[list[ExternalIdentifier]] = ListValidator(DataclassValidator(ExternalIdentifier))
+
+
+@validataclass
+class KienzlerStaticParkingSiteInput(BaseParkingSiteInput):
+    address: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=512)), DefaultUnset
+    type: OptionalUnsetNone[ParkingSiteType] = Noneable(EnumValidator(ParkingSiteType)), DefaultUnset
+
+    max_height: OptionalUnsetNone[int] = Noneable(IntegerValidator(min_value=0, allow_strings=True)), DefaultUnset
+    max_width: OptionalUnsetNone[int] = Noneable(IntegerValidator(min_value=0, allow_strings=True)), DefaultUnset
+    park_and_ride_type: OptionalUnsetNone[list[ParkAndRideType]] = (
+        Noneable(
+            ListValidator(EnumValidator(ParkAndRideType)),
+        ),
+        DefaultUnset,
+    )
+
+    # Set min/max to Europe borders
+    lat: Decimal = NumericValidator(min_value=34, max_value=72)
+    lon: Decimal = NumericValidator(min_value=-27, max_value=43)
+
+    external_identifiers: OptionalUnsetNone[list[ExternalIdentifierInput]] = (
+        Noneable(ListValidator(DataclassValidator(ExternalIdentifierInput))),
+        DefaultUnset,
+    )
+
+
+@validataclass
+class KienzlerGeojsonFeatureInput(GeojsonFeatureInput):
+    properties: KienzlerGeojsonFeaturePropertiesInput = DataclassValidator(KienzlerGeojsonFeaturePropertiesInput)
+
+    def to_static_parking_site_input(self, static_data_updated_at: datetime) -> KienzlerStaticParkingSiteInput:
+        properties_dict = self.properties.to_dict()
+        # TODO: this property should eventually be added to the StaticParkingSiteInput class.
+        properties_dict.pop('max_depth', None)
+        return KienzlerStaticParkingSiteInput(
+            lat=self.geometry.coordinates[1],
+            lon=self.geometry.coordinates[0],
+            **properties_dict,
+        )

src/parkapi_sources/models/parking_site_inputs.py

@@ -128,6 +128,11 @@ def __post_init__(self):
                     ),
                 )
 
+    def from_dict(self, data: dict):
+        for field in data.keys():
+            setattr(self, field, data[field])
+        return self
+
 
 @validataclass
 class RealtimeParkingSiteInput(BaseParkingSiteInput):

tests/converters/data/kienzler.geojson

@@ -0,0 +1,31 @@
+{
+    "type": "FeatureCollection",
+    "features": [
+        {
+            "type": "Feature",
+            "properties": {
+                "uid": "unit1676",
+                "type": "TWO_TIER",
+                "max_height": 1250,
+                "max_width": 800,
+                "max_depth": 1970,
+                "park_and_ride_type": [
+                    "TRAIN"
+                ],
+                "external_identifiers": [
+                    {
+                        "type": "DHID",
+                        "value": "de:08317:14500_Parent"
+                    }
+                ]
+            },
+            "geometry": {
+                "type": "Point",
+                "coordinates": [
+                    7.947474,
+                    48.475546
+                ]
+            }
+        }
+    ]
+}

tests/converters/kienzler_test.py

@@ -3,13 +3,15 @@
 Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
 """
 
+from decimal import Decimal
 from pathlib import Path
 from unittest.mock import Mock
 
 import pytest
 from requests_mock import Mocker
 
 from parkapi_sources.converters import KienzlerBikeAndRidePullConverter
+from parkapi_sources.models.enums import ParkAndRideType
 from tests.converters.helper import validate_realtime_parking_site_inputs, validate_static_parking_site_inputs
 
 
@@ -24,12 +26,27 @@ def requests_mock_kienzler(requests_mock: Mocker) -> Mocker:
     return requests_mock
 
 
+@pytest.fixture
+def requests_mock_kienzler_with_geojson(requests_mock: Mocker) -> Mocker:
+    json_path = Path(Path(__file__).parent, 'data', 'kienzler.geojson')
+    with json_path.open() as json_file:
+        json_data = json_file.read()
+
+    requests_mock.get(
+        'https://raw.githubusercontent.com/ParkenDD/parkapi-static-data/refs/heads/main/sources/kienzler_bike_and_ride.geojson',
+        text=json_data,
+    )
+
+    return requests_mock
+
+
 @pytest.fixture
 def kienzler_config_helper(mocked_config_helper: Mock):
     config = {
         'PARK_API_KIENZLER_BIKE_AND_RIDE_USER': '01275925-742c-460b-8778-eca90eb114bc',
         'PARK_API_KIENZLER_BIKE_AND_RIDE_PASSWORD': '626027f2-66e9-40bd-8ff2-4c010f5eca05',
         'PARK_API_KIENZLER_BIKE_AND_RIDE_IDS': 'id1,id2,id3',
+        'STATIC_GEOJSON_BASE_URL': 'https://raw.githubusercontent.com/ParkenDD/parkapi-static-data/refs/heads/main/sources',
     }
     mocked_config_helper.get.side_effect = lambda key, default=None: config.get(key, default)
     return mocked_config_helper
@@ -40,6 +57,11 @@ def kienzler_pull_converter(kienzler_config_helper: Mock) -> KienzlerBikeAndRide
     return KienzlerBikeAndRidePullConverter(config_helper=kienzler_config_helper)
 
 
+@pytest.fixture
+def kienzler_pull_converter_with_geojson(kienzler_config_helper: Mock) -> KienzlerBikeAndRidePullConverter:
+    return KienzlerBikeAndRidePullConverter(config_helper=kienzler_config_helper, use_geojson=True)
+
+
 class KienzlerPullConverterTest:
     @staticmethod
     def test_get_static_parking_sites(
@@ -53,6 +75,37 @@ def test_get_static_parking_sites(
 
         validate_static_parking_site_inputs(static_parking_site_inputs)
 
+    @staticmethod
+    def test_get_static_parking_sites_with_geojson(
+        kienzler_pull_converter_with_geojson: KienzlerBikeAndRidePullConverter,
+        requests_mock_kienzler: Mocker,
+        requests_mock_kienzler_with_geojson: Mocker,
+    ):
+        static_parking_site_inputs, import_parking_site_exceptions = (
+            kienzler_pull_converter_with_geojson.get_static_parking_sites()
+        )
+
+        assert len(static_parking_site_inputs) == 5
+        assert len(import_parking_site_exceptions) == 0
+
+        # Check that the data has been updated
+        aux = [
+            realtime_parking_site_input
+            for realtime_parking_site_input in static_parking_site_inputs
+            if realtime_parking_site_input.uid == 'unit1676'
+        ][0]
+        assert aux.uid == 'unit1676'
+        assert aux.type.value == 'TWO_TIER'
+        assert aux.max_height == 1250
+        assert aux.max_width == 800
+        assert aux.park_and_ride_type == [ParkAndRideType.TRAIN]
+        assert aux.external_identifiers[0]['type'].value == 'DHID'
+        assert aux.external_identifiers[0]['value'] == 'de:08317:14500_Parent'
+        assert aux.lat == Decimal('48.475546')
+        assert aux.lon == Decimal('7.947474')
+
+        validate_static_parking_site_inputs(static_parking_site_inputs)
+
     @staticmethod
     def test_get_realtime_parking_sites(
         kienzler_pull_converter: KienzlerBikeAndRidePullConverter,