Initial commit

dev/test-pull-converter.py

@@ -49,20 +49,20 @@ def main():
  print(f'static_parking_site_errors: {static_import_parking_site_exceptions}') # noqa: T201
  print(f'realtime_parking_site_errors: {realtime_import_parking_site_exceptions}') # noqa: T201
 
- # Additional re-validation if mapping was set up correctly
+ #Additional re-validation if mapping was set up correctly
  for static_parking_site_input in static_parking_site_inputs:
  static_parking_site_dict = json.loads(json.dumps(filter_none(static_parking_site_input.to_dict()), cls=DefaultJSONEncoder))
  try:
  converter.static_parking_site_validator.validate(static_parking_site_dict)
  except ValidationError as e:
  print(f'Invalid mapped static dataset found: {static_parking_site_dict}: {e.to_dict()}')
 
- for realtime_parking_site_input in realtime_parking_site_inputs:
- realtime_parking_site_dict = json.loads(json.dumps(filter_none(realtime_parking_site_input.to_dict()), cls=DefaultJSONEncoder))
- try:
- converter.realtime_parking_site_validator.validate(realtime_parking_site_dict)
- except ValidationError as e:
- print(f'Invalid mapped realtime dataset found: {realtime_parking_site_dict}: {e.to_dict()}')
+ #for realtime_parking_site_input in realtime_parking_site_inputs:
+ # realtime_parking_site_dict = json.loads(json.dumps(filter_none(realtime_parking_site_input.to_dict()), cls=DefaultJSONEncoder))
+ # try:
+ # converter.realtime_parking_site_validator.validate(realtime_parking_site_dict)
+ # except ValidationError as e:
+ # print(f'Invalid mapped realtime dataset found: {realtime_parking_site_dict}: {e.to_dict()}')
 
 
 def filter_none(data: dict) -> dict:

src/parkapi_sources/converters/__init__.py

@@ -26,3 +26,4 @@
 from .stuttgart import StuttgartPushConverter
 from .ulm import UlmPullConverter
 from .vrs_p_r import VrsParkAndRidePushConverter
+from .herrenberg import HerrenbergPullConverter

src/parkapi_sources/converters/herrenberg/__init__.py

@@ -0,0 +1,6 @@
+"""
+Copyright 2024 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from .converter import HerrenbergPullConverter

src/parkapi_sources/converters/herrenberg/converter.py

@@ -0,0 +1,73 @@
+"""
+Copyright 2024 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+import requests
+from validataclass.exceptions import ValidationError
+from validataclass.validators import DataclassValidator
+
+from parkapi_sources.converters.base_converter import BaseConverter
+from parkapi_sources.converters.herrenberg.mapper import Herrenbergmapper
+from parkapi_sources.converters.herrenberg.validation import HerrenbergParkingSiteInput
+from parkapi_sources.exceptions import ImportParkingSiteException
+from parkapi_sources.models import StaticParkingSiteInput, SourceInfo, RealtimeParkingSiteInput
+
+
+class HerrenbergPullConverter(BaseConverter):
+ mapper = Herrenbergmapper()
+ _base_url = 'https://api.stadtnavi.de/herrenberg/parking/parkapi.json'
+ parking_site_validator = DataclassValidator(HerrenbergParkingSiteInput)
+
+ source_info = SourceInfo(
+ uid='herrenberg',
+ name='Stadt Herrenberg',
+ public_url='https://www.herrenberg.de/de/Stadtleben/Erlebnis-Herrenberg/Service/Parkplaetze',
+ source_url='/home/jebus/Arbeit/parkapi-sources-v3-t/tests/converters/data/herrenberg.json',
+ timezone='Europe/Berlin',
+ attribution_contributor='Stadt Herrenberg',
+ has_realtime_data=False,
+ )
+
+ def get_static_parking_sites(self) -> tuple[list[StaticParkingSiteInput], list[ImportParkingSiteException]]:
+ static_parking_site_inputs: list[StaticParkingSiteInput] = []
+ static_parking_site_errors: list[ImportParkingSiteException] = []
+
+ input_data = self._get_remote_data()
+
+ for item in input_data:
+ try:
+ validated: HerrenbergParkingSiteInput = self.parking_site_validator.validate(item)
+ except ValidationError as e:
+ try:
+ static_parking_site_errors.append(
+ ImportParkingSiteException(
+ source_uid=self.source_info.uid,
+ parking_site_uid=item.id,
+ message=f'validation error: {e.to_dict()}',
+ ),
+ )
+ except AttributeError:
+ static_parking_site_errors.append(
+ ImportParkingSiteException(
+ source_uid=self.source_info.uid,
+ parking_site_uid=None,
+ message=f'validation error: {e.to_dict()}',
+ ),
+ )
+ continue
+ static_parking_site_inputs.append(self.mapper.map_static_parking_site(validated))
+ return static_parking_site_inputs, static_parking_site_errors
+
+ def get_realtime_parking_sites(self) -> tuple[list[RealtimeParkingSiteInput], list[ImportParkingSiteException]]:
+ pass
+ # since there is no realtimedata this is just skipped
+
+ def _get_remote_data(self) -> list[dict]:
+ response = requests.get(self._base_url, timeout=60)
+ result_dict: dict = response.json()
+
+ items: list[dict] = []
+ for item in result_dict['lots']:
+ items.append(item)
+ return items

src/parkapi_sources/converters/herrenberg/mapper.py

@@ -0,0 +1,38 @@
+"""
+Copyright 2024 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+
+from parkapi_sources.converters.herrenberg.validation import HerrenbergParkingSiteInput
+from parkapi_sources.models import StaticParkingSiteInput
+from parkapi_sources.models.enums import ParkAndRideType
+
+
+class Herrenbergmapper:
+ def map_static_parking_site(self, herrenbergparkingsiteinput: HerrenbergParkingSiteInput) -> StaticParkingSiteInput:
+ fee = None
+ if herrenbergparkingsiteinput.fee_hours:
+ fee = True
+ herrenbergparkingsiteinput.ride = ParkAndRideType.NO
+
+
+
+ return StaticParkingSiteInput(
+ uid=herrenbergparkingsiteinput.id,
+ name=herrenbergparkingsiteinput.name,
+ lat=herrenbergparkingsiteinput.coords.lat,
+ lon=herrenbergparkingsiteinput.coords.lng,
+ operator_name="Stadt Herrenberg",
+ address="Herrenberg " + herrenbergparkingsiteinput.address,
+ capacity=herrenbergparkingsiteinput.total,
+ description=herrenbergparkingsiteinput.notes.de,
+ type=herrenbergparkingsiteinput.lot_type.to_parking_site_input_type(),
+ park_and_ride_type=herrenbergparkingsiteinput.lot_type.to_parking_site_input_ride(),
+ public_url=herrenbergparkingsiteinput.url,
+ fee_description=herrenbergparkingsiteinput.fee_hours,
+ opening_hours=herrenbergparkingsiteinput.opening_hours,
+ has_fee=fee,
+
+
+ )

src/parkapi_sources/converters/herrenberg/validation.py

@@ -0,0 +1,80 @@
+"""
+Copyright 2024 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from enum import Enum
+
+from validataclass.dataclasses import validataclass, DefaultUnset
+from validataclass.helpers import OptionalUnsetNone
+from validataclass.validators import (
+ DataclassValidator,
+ EnumValidator,
+ IntegerValidator,
+ Noneable,
+ NumericValidator,
+ StringValidator,
+ UrlValidator,
+)
+
+from parkapi_sources.models.enums import ParkingSiteType, ParkAndRideType
+
+
+class HerrenbergParkingSiteType(Enum):
+ Parkplatz = 'Parkplatz'
+ Parkhaus = 'Parkhaus'
+ Wohnmobilparkplatz = 'Wohnmobilparkplatz'
+ Park_Carpool = 'Park-Carpool'
+ Barrierefreier_Parkplatz = 'Barrierefreier-Parkplatz'
+ Tiefgarage = 'Tiefgarage'
+
+ def to_parking_site_input_type(self) -> ParkingSiteType:
+ return {
+ self.Parkplatz: ParkingSiteType.ON_STREET,
+ self.Parkhaus: ParkingSiteType.OFF_STREET_PARKING_GROUND,
+ self.Wohnmobilparkplatz: ParkingSiteType.OFF_STREET_PARKING_GROUND,
+ self.Park_Carpool: ParkingSiteType.OFF_STREET_PARKING_GROUND,
+ self.Barrierefreier_Parkplatz: ParkingSiteType.ON_STREET,
+ self.Tiefgarage: ParkingSiteType.ON_STREET
+ }.get(HerrenbergParkingSiteType)
+
+ def to_parking_site_input_ride(self) -> ParkAndRideType:
+ return {
+ self.Parkhaus: ParkAndRideType.YES,
+ self.Park_Carpool: ParkAndRideType.CARPOOL,
+
+ }.get(HerrenbergParkingSiteType)
+
+
+class HerrenbergParkingRideType(Enum):
+ Parkhaus = 'Parkhaus'
+ Park_Carpool = 'Park-Carpool'
+
+
+@validataclass
+class HerrenbergNotesInput:
+ de: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=512)), DefaultUnset
+ en: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=512)), DefaultUnset
+
+
+@validataclass
+class HerrenbergCoordsInput:
+ lat: OptionalUnsetNone[str] = NumericValidator(), DefaultUnset
+ lng: OptionalUnsetNone[str] = NumericValidator(), DefaultUnset
+
+
+@validataclass
+class HerrenbergParkingSiteInput:
+ id: str = StringValidator(min_length=1, max_length=256)
+ name: str = StringValidator(min_length=1, max_length=256)
+ lot_type: OptionalUnsetNone[HerrenbergParkingSiteType] = Noneable(
+ EnumValidator(HerrenbergParkingSiteType)), DefaultUnset
+ ride: OptionalUnsetNone[HerrenbergParkingSiteType] = Noneable(
+ EnumValidator(HerrenbergParkingRideType)), DefaultUnset,
+ total: OptionalUnsetNone[int] = Noneable(IntegerValidator(min_value=0, allow_strings=True)), DefaultUnset
+ url: OptionalUnsetNone[str] = Noneable(UrlValidator(max_length=4096)), DefaultUnset
+ fee_hours: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=4096)), DefaultUnset
+ opening_hours: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=512)), DefaultUnset
+ address: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=512)), DefaultUnset
+ notes: OptionalUnsetNone[HerrenbergNotesInput] = Noneable(DataclassValidator(HerrenbergNotesInput)), DefaultUnset
+ coords: OptionalUnsetNone[HerrenbergCoordsInput] = Noneable(DataclassValidator(HerrenbergCoordsInput)), DefaultUnset

src/parkapi_sources/parkapi_sources.py

@@ -18,6 +18,7 @@
  EllwangenPushConverter,
  FreiburgPullConverter,
  HeidelbergPullConverter,
+ HerrenbergPullConverter,
  KarlsruheBikePullConverter,
  KarlsruhePullConverter,
  KienzlerPullConverter,
@@ -54,6 +55,7 @@ class ParkAPISources:
  BuchenPushConverter,
  FreiburgPullConverter,
  HeidelbergPullConverter,
+ HerrenbergPullConverter,
  KarlsruheBikePullConverter,
  KarlsruhePullConverter,
  KienzlerPullConverter,

tests/converters/data/herrenberg.json

@@ -0,0 +1,24 @@
+{
+ "data_source": "https://www.herrenberg.de/de/Stadtleben/Erlebnis-Herrenberg/Service/Parkplaetze",
+ "last_downloaded": "2024-06-07T11:34:01.650112",
+ "last_updated": "2024-06-07T11:34:01.650112",
+ "lots": [
+ {
+ "id": "ac230e00-2cea-11eb-93e4-7f8bf428cc7a",
+ "lot_type": "Parkhaus",
+ "address": "71083 Herrenberg",
+ "name": "Altstadtgarage",
+ "forecast": false,
+ "state": "many",
+ "coords": {
+ "lat": 48.593768,
+ "lng": 8.872705
+ },
+ "total": 97,
+ "url": "https://stadtwerke.herrenberg.de/oepnv-parken/parken/sicher-und-zentral-parken/parkhaeuser-und-parkgebuehren/altstadtgarage/",
+ "fee_hours": "Mo-Fr 08:00-20:00; Sa 08:00-14:00; PH off",
+ "notes": {},
+ "free": 69
+ }
+ ]
+}

tests/converters/herrenberg_test.py

@@ -0,0 +1,35 @@
+"""
+Copyright 2024 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+
+from parkapi_sources.converters.herrenberg import HerrenbergPullConverter
+from tests.converters.helper import validate_static_parking_site_inputs
+
+
+@pytest.fixture
+def herrenberg_config_helper(mocked_config_helper: Mock):
+ config = {
+ }
+ mocked_config_helper.get.side_effect = lambda key, default=None: config.get(key, default)
+ return mocked_config_helper
+
+
+@pytest.fixture
+def herrenberg_pull_converter(herrenberg_config_helper: Mock, ) -> HerrenbergPullConverter:
+ return HerrenbergPullConverter(config_helper=herrenberg_config_helper)
+
+
+class HerrenbergPullConverterTest:
+ @staticmethod
+ def test_get_static_parking_sites(herrenberg_pull_converter: HerrenbergPullConverter):
+ static_parking_site_inputs, import_parking_site_exceptions = herrenberg_pull_converter.get_static_parking_sites()
+ assert len(static_parking_site_inputs) < len(
+ import_parking_site_exceptions
+ ), 'There should be more valid then invalid parking sites'
+
+ validate_static_parking_site_inputs(static_parking_site_inputs)