feature: heidelberg converter

src/parkapi_sources/converters/heidelberg/converter.py

@@ -5,71 +5,84 @@
 
 import requests
 from validataclass.exceptions import ValidationError
-from validataclass.validators import DataclassValidator
+from validataclass.validators import AnythingValidator, DataclassValidator, ListValidator
 
-from parkapi_sources.converters.base_converter.pull import PullConverter, StaticGeojsonDataMixin
+from parkapi_sources.converters.base_converter.pull import PullConverter
 from parkapi_sources.exceptions import ImportParkingSiteException, ImportSourceException
 from parkapi_sources.models import RealtimeParkingSiteInput, SourceInfo, StaticParkingSiteInput
 
-from .validators import HeidelbergRealtimeInput, HeidelbergRealtimeUpdateInput
+from .models import HeidelbergInput
 
 
-class HeidelbergPullConverter(PullConverter, StaticGeojsonDataMixin):
+class HeidelbergPullConverter(PullConverter):
  required_config_keys = ['PARK_API_HEIDELBERG_API_KEY']
- heidelberg_realtime_validator = DataclassValidator(HeidelbergRealtimeInput)
- heidelberg_realtime_update_validator = DataclassValidator(HeidelbergRealtimeUpdateInput)
+ list_validator = ListValidator(AnythingValidator(allowed_types=[dict]))
+ heidelberg_validator = DataclassValidator(HeidelbergInput)
+
  source_info = SourceInfo(
  uid='heidelberg',
  name='Stadt Heidelberg',
  public_url='https://parken.heidelberg.de',
- source_url='https://parken.heidelberg.de/v1',
+ source_url='https://api.datenplattform.heidelberg.de/ckan/or/mobility/main/offstreetparking/v2/entities',
  timezone='Europe/Berlin',
  attribution_contributor='Stadt Heidelberg',
  has_realtime_data=True,
  )
 
  def get_static_parking_sites(self) -> tuple[list[StaticParkingSiteInput], list[ImportParkingSiteException]]:
- return self._get_static_parking_site_inputs_and_exceptions(source_uid=self.source_info.uid)
+ static_parking_site_inputs: list[StaticParkingSiteInput] = []
+
+ heidelberg_inputs, import_parking_site_exceptions = self._get_data()
+
+ for heidelberg_input in heidelberg_inputs:
+ static_parking_site_inputs.append(heidelberg_input.to_static_parking_site())
+
+ return static_parking_site_inputs, import_parking_site_exceptions
 
  def get_realtime_parking_sites(self) -> tuple[list[RealtimeParkingSiteInput], list[ImportParkingSiteException]]:
  realtime_parking_site_inputs: list[RealtimeParkingSiteInput] = []
+
+ heidelberg_inputs, import_parking_site_exceptions = self._get_data()
+
+ for heidelberg_input in heidelberg_inputs:
+ if heidelberg_input.availableSpotNumber is None:
+ continue
+ realtime_parking_site_inputs.append(heidelberg_input.to_realtime_parking_site_input())
+
+ return realtime_parking_site_inputs, import_parking_site_exceptions
+
+ def _get_data(self) -> tuple[list[HeidelbergInput], list[ImportParkingSiteException]]:
+ heidelberg_inputs: list[HeidelbergInput] = []
  import_parking_site_exceptions: list[ImportParkingSiteException] = []
 
  response = requests.get(
- f'{self.source_info.source_url}/parking-update',
- params={'key': self.config_helper.get('PARK_API_HEIDELBERG_API_KEY')},
- headers={
- 'Accept': 'application/json; charset=utf-8',
- 'Referer': 'https://parken.heidelberg.de',
- },
+ self.source_info.source_url,
+ params={'api-key': self.config_helper.get('PARK_API_HEIDELBERG_API_KEY'), 'limit': 50},
+ headers={'X-Gravitee-Api-Key': self.config_helper.get('PARK_API_HEIDELBERG_API_KEY')},
  timeout=30,
  )
  response_data = response.json()
  try:
- realtime_input = self.heidelberg_realtime_validator.validate(response_data)
+ input_dicts = self.list_validator.validate(response_data)
  except ValidationError as e:
  raise ImportSourceException(
  source_uid=self.source_info.uid,
  message=f'Invalid Input at source {self.source_info.uid}: {e.to_dict()}, data: {response_data}',
  ) from e
 
- for update_dict in realtime_input.data.parkingupdates:
+ for input_dict in input_dicts:
  try:
- update_input = self.heidelberg_realtime_update_validator.validate(update_dict)
+ heidelberg_input = self.heidelberg_validator.validate(input_dict)
  except ValidationError as e:
  import_parking_site_exceptions.append(
  ImportParkingSiteException(
  source_uid=self.source_info.uid,
- parking_site_uid=update_dict.get('parkinglocation'),
- message=f'Invalid data at uid {update_dict.get("parkinglocation")}: {e.to_dict()}, ' f'data: {update_dict}',
+ parking_site_uid=input_dict.get('staticParkingSiteId'),
+ message=f'Invalid data at uid {input_dict.get("staticParkingSiteId")}: {e.to_dict()}, ' f'data: {input_dict}',
  ),
  )
  continue
 
- realtime_parking_site_inputs.append(
- update_input.to_realtime_parking_site_input(
- realtime_data_updated_at=realtime_input.data.updated,
- ),
- )
+ heidelberg_inputs.append(heidelberg_input)
 
- return realtime_parking_site_inputs, import_parking_site_exceptions
+ return heidelberg_inputs, import_parking_site_exceptions

src/parkapi_sources/converters/heidelberg/models.py

@@ -0,0 +1,175 @@
+"""
+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 datetime import datetime, time
+from decimal import Decimal
+from enum import Enum
+from typing import Optional
+
+from validataclass.dataclasses import Default, validataclass
+from validataclass.validators import (
+ AnythingValidator,
+ DateTimeValidator,
+ EnumValidator,
+ IntegerValidator,
+ ListValidator,
+ NumericValidator,
+ StringValidator,
+ TimeFormat,
+ TimeValidator,
+ UrlValidator,
+)
+
+from parkapi_sources.models import RealtimeParkingSiteInput, StaticParkingSiteInput
+from parkapi_sources.models.enums import OpeningStatus, ParkAndRideType, ParkingSiteType, SupervisionType
+from parkapi_sources.validators import ExcelNoneable, ReplacingStringValidator, Rfc1123DateTimeValidator
+
+from .validators import NoneableRemoveValueDict, RemoveValueDict
+
+
+class HeidelbergFacilityType(Enum):
+ HANDICAPPED_ACCESSIBLE_PAYING_MASCHINE = 'handicapped accessible paying machine'
+ INTERCOM_AT_EXIT = 'Intercom at Exit'
+ SECURITY_CAMERA = 'Security Camera'
+ ACCESSABLE = 'Accessable'
+ HANDICAPPED_BATHROOM = 'Handicapped Bathroom'
+ BATHROOM = 'Bathroom'
+ STAFF = 'Staff'
+ CHANGING_TABLE = 'Changing Table'
+ BIKE_PARKING = 'BikeParking'
+ ELEVATOR = 'Elevator'
+ DEFIBRILLATOR = 'Defibirlator'
+ COPY_MASCHINE_OR_SERVICE = 'CopyMachineOrService'
+
+
+class HeidelbergPaymentMethodType(Enum):
+ CASH = 'Cash'
+ MONEY_CARD = 'MoneyCard'
+ DEBIT_CART = 'DebitCard'
+ GOOGLE = 'Google'
+ PAY_PAL = 'PayPal'
+ LICENCE_PLATE = 'Licence Plate'
+ CREDIT_CARD = 'CreditCard'
+ INVOICE = 'Invoice'
+ COD = 'COD'
+
+
+class HeidelbergParkingSiteStatus(Enum):
+ OPEN = 'Open'
+ CLOSED = 'Closed'
+ OPEN_DE = 'Offen'
+ CLOSED_DE = 'Geschlossen'
+ BROKEN = 'Stoerung'
+ UNKNOWN = '0'
+
+ def to_opening_status(self) -> OpeningStatus:
+ return {
+ self.OPEN: OpeningStatus.OPEN,
+ self.OPEN_DE: OpeningStatus.OPEN,
+ self.CLOSED: OpeningStatus.CLOSED,
+ self.CLOSED_DE: OpeningStatus.CLOSED,
+ self.UNKNOWN: OpeningStatus.UNKNOWN,
+ self.BROKEN: OpeningStatus.CLOSED,
+ }.get(self, OpeningStatus.UNKNOWN)
+
+
+class HeidelbergParkingType(Enum):
+ OFFSTREET_PARKING = 'OffStreetParking'
+
+
+class HeidelbergParkingSubType(Enum):
+ GARAGE = 'Parking Garage'
+ PARK_AND_RIDE = 'Park and Ride Car Park'
+
+
+@validataclass
+class HeidelbergInput:
+ acceptedPaymentMethod: HeidelbergPaymentMethodType = RemoveValueDict(ListValidator(EnumValidator(HeidelbergPaymentMethodType)))
+ addressLocality: str = RemoveValueDict(StringValidator())
+ availableSpotNumber: Optional[int] = NoneableRemoveValueDict(IntegerValidator()), Default(None)
+ closingHours: time = RemoveValueDict(TimeValidator(time_format=TimeFormat.NO_SECONDS))
+ description: str = RemoveValueDict(ReplacingStringValidator(mapping={'\r': '', '\n': ' ', '\xa0': ' '}))
+ facilities: list[str] = RemoveValueDict(ListValidator(EnumValidator(HeidelbergFacilityType)))
+ familyParkingSpots: int = RemoveValueDict(IntegerValidator())
+ googlePlaceId: str = RemoveValueDict(StringValidator())
+ handicappedParkingSpots: int = RemoveValueDict(IntegerValidator())
+ images: list[str] = RemoveValueDict(ListValidator(UrlValidator()))
+ lat: Decimal = RemoveValueDict(NumericValidator())
+ lon: Decimal = RemoveValueDict(NumericValidator())
+ maximumAllowedHeight: Optional[Decimal] = RemoveValueDict(ExcelNoneable(NumericValidator()))
+ maximumAllowedWidth: Optional[Decimal] = RemoveValueDict(ExcelNoneable((NumericValidator())))
+ observationDateTime: datetime = RemoveValueDict(DateTimeValidator())
+ openingHours: time = RemoveValueDict(TimeValidator(time_format=TimeFormat.NO_SECONDS))
+ type: HeidelbergParkingType = EnumValidator(HeidelbergParkingType)
+ parking_type: HeidelbergParkingSubType = RemoveValueDict(EnumValidator(HeidelbergParkingSubType))
+ postalCode: int = RemoveValueDict(IntegerValidator()) # outsch
+ provider: str = RemoveValueDict(StringValidator())
+ staticName: str = RemoveValueDict(StringValidator())
+ staticParkingSiteId: str = RemoveValueDict(StringValidator())
+ staticStatus: HeidelbergParkingSiteStatus = RemoveValueDict(EnumValidator(HeidelbergParkingSiteStatus))
+ staticTotalSpotNumber: int = RemoveValueDict(IntegerValidator())
+ status: HeidelbergParkingSiteStatus = RemoveValueDict(EnumValidator(HeidelbergParkingSiteStatus))
+ streetAddress: str = RemoveValueDict(StringValidator())
+ streetAddressDriveway: str = RemoveValueDict(StringValidator())
+ streetAddressExit: str = RemoveValueDict(StringValidator())
+ totalSpotNumber: int = RemoveValueDict(IntegerValidator())
+ website: Optional[str] = RemoveValueDict(ExcelNoneable(UrlValidator()))
+ womenParkingSpots: int = RemoveValueDict(IntegerValidator())
+
+ def to_static_parking_site(self) -> StaticParkingSiteInput:
+ if self.parking_type == HeidelbergParkingSubType.GARAGE:
+ parking_site_type = ParkingSiteType.CAR_PARK
+ elif self.type == HeidelbergParkingType.OFFSTREET_PARKING:
+ parking_site_type = ParkingSiteType.OFF_STREET_PARKING_GROUND
+ else:
+ parking_site_type = None
+
+ if self.openingHours == self.closingHours:
+ opening_hours = '24/7'
+ else:
+ opening_hours = f'{self.openingHours.isoformat()[:5]}-{self.closingHours.isoformat()[:5]}'
+
+ return StaticParkingSiteInput(
+ uid=self.staticParkingSiteId,
+ name=self.staticName,
+ description=self.description.replace('\r\n', ' '),
+ lat=self.lat,
+ lon=self.lon,
+ address=f'{self.streetAddress}, {self.postalCode} {self.addressLocality}',
+ operator_name=self.provider,
+ max_height=None if self.maximumAllowedHeight is None else int(self.maximumAllowedHeight * 100),
+ max_width=None if self.maximumAllowedWidth is None else int(self.maximumAllowedWidth * 100),
+ photo_url=self.images[0] if len(self.images) else None,
+ capacity=self.totalSpotNumber,
+ capacity_disabled=self.handicappedParkingSpots,
+ capacity_woman=self.womenParkingSpots,
+ capacity_family=self.familyParkingSpots,
+ opening_hours=opening_hours,
+ static_data_updated_at=self.observationDateTime,
+ type=parking_site_type,
+ park_and_ride_type=[ParkAndRideType.YES] if self.parking_type == HeidelbergParkingSubType.PARK_AND_RIDE else None,
+ # TODO: Maybe STAFF means SupervisionType.ATTENDED?
+ supervision_type=SupervisionType.VIDEO if HeidelbergFacilityType.SECURITY_CAMERA in self.facilities else None,
+ has_realtime_data=self.availableSpotNumber is not None,
+ )
+
+ def to_realtime_parking_site_input(self) -> RealtimeParkingSiteInput:
+ return RealtimeParkingSiteInput(
+ uid=self.staticParkingSiteId,
+ realtime_capacity=self.totalSpotNumber,
+ realtime_capacity_disabled=self.handicappedParkingSpots,
+ realtime_capacity_woman=self.womenParkingSpots,
+ realtime_capacity_family=self.familyParkingSpots,
+ realtime_free_capacity=self.availableSpotNumber,
+ # TODO: most likely broken, as there are realtime open parking sites with static status broken / unknown
+ realtime_opening_status=self.status.to_opening_status(),
+ realtime_data_updated_at=self.observationDateTime,
+ )
+
+
+@validataclass
+class HeidelbergRealtimeDataInput:
+ parkingupdates: list[dict] = ListValidator(AnythingValidator(allowed_types=dict))
+ updated: datetime = Rfc1123DateTimeValidator()

src/parkapi_sources/converters/heidelberg/validators.py

@@ -3,54 +3,32 @@
 Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
 """
 
-from datetime import datetime
-from enum import Enum
+from typing import Any, Optional
 
-from validataclass.dataclasses import validataclass
-from validataclass.validators import AnythingValidator, DataclassValidator, EnumValidator, IntegerValidator, ListValidator
+from validataclass.validators import Validator
 
-from parkapi_sources.models import RealtimeParkingSiteInput
-from parkapi_sources.models.enums import OpeningStatus
-from parkapi_sources.validators import Rfc1123DateTimeValidator
 
+class RemoveValueDict(Validator):
 
-class HeidelbergParkingSiteStatus(Enum):
- available = 'available'
- occupied = 'occupied'
- none = 'none'
+ wrapped_validator: Validator
 
- def to_opening_status(self) -> OpeningStatus:
- return {
- self.available: OpeningStatus.OPEN,
- # For some reason, occupied in API means closed
- self.occupied: OpeningStatus.CLOSED,
- self.none: OpeningStatus.UNKNOWN,
- }.get(self)
+ def __init__(self, validator: Validator):
 
+ # Check parameter validity
+ if not isinstance(validator, Validator):
+ raise TypeError('RemoveValueDict requires a Validator instance.')
 
-@validataclass
-class HeidelbergRealtimeUpdateInput:
- parkinglocation: int = IntegerValidator()
- total: int = IntegerValidator()
- current: int = IntegerValidator()
- status: HeidelbergParkingSiteStatus = EnumValidator(HeidelbergParkingSiteStatus)
+ self.wrapped_validator = validator
 
- def to_realtime_parking_site_input(self, realtime_data_updated_at: datetime) -> RealtimeParkingSiteInput:
- return RealtimeParkingSiteInput(
- uid=str(self.parkinglocation),
- realtime_capacity=self.total,
- realtime_free_capacity=self.current,
- realtime_opening_status=self.status.to_opening_status(),
- realtime_data_updated_at=realtime_data_updated_at,
- )
+ def validate(self, input_data: Any, **kwargs: Any) -> Any:
+ self._ensure_type(input_data, dict)
 
+ return self.wrapped_validator.validate(input_data.get('value'), **kwargs)
 
-@validataclass
-class HeidelbergRealtimeDataInput:
- parkingupdates: list[dict] = ListValidator(AnythingValidator(allowed_types=dict))
- updated: datetime = Rfc1123DateTimeValidator()
 
+class NoneableRemoveValueDict(RemoveValueDict):
+ def validate(self, input_data: Any, **kwargs: Any) -> Optional[Any]:
+ if input_data is None:
+ return None
 
-@validataclass
-class HeidelbergRealtimeInput:
- data: HeidelbergRealtimeDataInput = DataclassValidator(HeidelbergRealtimeDataInput)
+ return super().validate(input_data, **kwargs)

src/parkapi_sources/models/parking_site_inputs.py

@@ -49,6 +49,7 @@ class StaticParkingSiteInput(BaseParkingSiteInput):
 
  max_stay: OptionalUnsetNone[int] = Noneable(IntegerValidator(min_value=0, allow_strings=True)), 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
  has_lighting: OptionalUnsetNone[bool] = Noneable(BooleanValidator()), DefaultUnset
  is_covered: OptionalUnsetNone[bool] = Noneable(BooleanValidator()), DefaultUnset
  fee_description: OptionalUnsetNone[str] = Noneable(StringValidator(max_length=4096)), DefaultUnset

src/parkapi_sources/validators/__init__.py

@@ -8,7 +8,7 @@
 from .datetime_validator import Rfc1123DateTimeValidator, SpacedDateTimeValidator, TimestampDateTimeValidator
 from .decimal_validators import GermanDecimalValidator
 from .integer_validators import GermanDurationIntegerValidator
-from .list_validator import PointCoordinateTupleValidator
+from .list_validator import DumpedListValidator, PointCoordinateTupleValidator
 from .noneable import ExcelNoneable
 from .string_validators import NumberCastingStringValidator, ReplacingStringValidator
 from .time_validators import ExcelTimeValidator

src/parkapi_sources/validators/list_validator.py

@@ -3,7 +3,9 @@
 Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
 """
 
+import json
 import re
+from json import JSONDecodeError
 from typing import Any
 
 from validataclass.exceptions import ValidationError
@@ -23,3 +25,14 @@ def validate(self, input_data: Any, **kwargs) -> list:
  input_data = [input_match.group(1), input_match.group(2)]
 
  return super().validate(input_data, **kwargs)
+
+
+class DumpedListValidator(ListValidator):
+ def validate(self, input_data: Any, **kwargs) -> list:
+ self._ensure_type(input_data, str)
+ try:
+ input_data = json.loads(input_data)
+ except JSONDecodeError as e:
+ raise ValidationError(code='invalid_json_input', reason=f'invalid JSON input: {e}') from e
+
+ return super().validate(input_data, **kwargs)