A81 P&M

src/parkapi_sources/converters/__init__.py

@@ -3,6 +3,7 @@
 Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
 """
 
+from .a81_p_m import A81PMPullConverter
 from .bahn_v2 import BahnV2PullConverter
 from .base_converter import BaseConverter
 from .bfrk_bw import BfrkBwOepnvBikePushConverter, BfrkBwOepnvCarPushConverter, BfrkBwSpnvBikePushConverter, BfrkBwSpnvCarPushConverter

src/parkapi_sources/converters/a81_p_m/__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 A81PMPullConverter

src/parkapi_sources/converters/a81_p_m/converter.py

@@ -0,0 +1,82 @@
+"""
+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 AnythingValidator, DataclassValidator, ListValidator
+
+from parkapi_sources.converters.base_converter.pull import PullConverter
+from parkapi_sources.exceptions import ImportParkingSiteException
+from parkapi_sources.models import RealtimeParkingSiteInput, SourceInfo, StaticParkingSiteInput
+
+from .models import A81PMInput
+
+
+class A81PMPullConverter(PullConverter):
+ required_config_keys = ['PARK_API_A81_P_M_TOKEN']
+
+ list_validator = ListValidator(AnythingValidator(allowed_types=[dict]))
+ a81_p_m_site_validator = DataclassValidator(A81PMInput)
+
+ source_info = SourceInfo(
+ uid='a81_p_m',
+ name='A81: P&M',
+ source_url='https://api.cloud-telartec.de/v1/parkings',
+ has_realtime_data=True,
+ )
+
+ def get_static_parking_sites(self) -> tuple[list[StaticParkingSiteInput], list[ImportParkingSiteException]]:
+ static_parking_site_inputs: list[StaticParkingSiteInput] = []
+ static_parking_site_errors: list[ImportParkingSiteException] = []
+
+ parking_site_dicts = self.get_data()
+
+ for parking_site_dict in parking_site_dicts:
+ try:
+ parking_site_input: A81PMInput = self.a81_p_m_site_validator.validate(parking_site_dict)
+ except ValidationError as e:
+ static_parking_site_errors.append(
+ ImportParkingSiteException(
+ source_uid=self.source_info.uid,
+ parking_site_uid=parking_site_dict.get('id'),
+ message=f'validation error for static data {parking_site_dict}: {e.to_dict()}',
+ ),
+ )
+ continue
+
+ static_parking_site_inputs.append(parking_site_input.to_static_parking_site())
+
+ return static_parking_site_inputs, static_parking_site_errors
+
+ def get_realtime_parking_sites(self) -> tuple[list[RealtimeParkingSiteInput], list[ImportParkingSiteException]]:
+ realtime_parking_site_inputs: list[RealtimeParkingSiteInput] = []
+ realtime_parking_site_errors: list[ImportParkingSiteException] = []
+
+ parking_site_dicts = self.get_data()
+
+ for parking_site_dict in parking_site_dicts:
+ try:
+ parking_site_input: A81PMInput = self.a81_p_m_site_validator.validate(parking_site_dict)
+ except ValidationError as e:
+ realtime_parking_site_errors.append(
+ ImportParkingSiteException(
+ source_uid=self.source_info.uid,
+ parking_site_uid=parking_site_dict.get('id'),
+ message=f'validation error for realtime data {parking_site_dict}: {e.to_dict()}',
+ ),
+ )
+ continue
+
+ realtime_parking_site_inputs.append(parking_site_input.to_realtime_parking_site())
+
+ return realtime_parking_site_inputs, realtime_parking_site_errors
+
+ def get_data(self) -> list[dict]:
+ response = requests.get(
+ self.source_info.source_url,
+ headers={'Authorization': f'Bearer {self.config_helper.get("PARK_API_A81_P_M_TOKEN")}'},
+ timeout=60,
+ )
+ return self.list_validator.validate(response.json())

src/parkapi_sources/converters/a81_p_m/models.py

@@ -0,0 +1,82 @@
+"""
+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, timezone
+from decimal import Decimal
+from enum import Enum
+from zoneinfo import ZoneInfo
+
+from validataclass.dataclasses import validataclass
+from validataclass.validators import DataclassValidator, EnumValidator, IntegerValidator, NumericValidator, StringValidator
+
+from parkapi_sources.models import RealtimeParkingSiteInput, StaticParkingSiteInput
+from parkapi_sources.validators import SpacedDateTimeValidator
+
+
+class A81PMConnectionStatus(Enum):
+ OFFLINE = 'OFFLINE'
+ ONLINE = 'ONLINE'
+
+
+class A81PMCategory(Enum):
+ P_M = 'P&M'
+
+
+@validataclass
+class A81PMCapacityInput:
+ bus: int = IntegerValidator()
+ car: int = IntegerValidator()
+ car_charging: int = IntegerValidator()
+ car_handicap: int = IntegerValidator()
+ car_women: int = IntegerValidator()
+ truck: int = IntegerValidator()
+
+
+@validataclass
+class A81PMLocationInput:
+ lat: Decimal = NumericValidator()
+ lng: Decimal = NumericValidator()
+
+
+@validataclass
+class A81PMInput:
+ id: str = StringValidator()
+ long_name: str = StringValidator()
+ name: str = StringValidator()
+ status: A81PMConnectionStatus = EnumValidator(A81PMConnectionStatus)
+ time: datetime = SpacedDateTimeValidator(
+ local_timezone=ZoneInfo('Europe/Berlin'),
+ target_timezone=timezone.utc,
+ )
+ location: A81PMLocationInput = DataclassValidator(A81PMLocationInput)
+ capacity: A81PMCapacityInput = DataclassValidator(A81PMCapacityInput)
+ free_capacity: A81PMCapacityInput = DataclassValidator(A81PMCapacityInput)
+
+ def to_static_parking_site(self) -> StaticParkingSiteInput:
+ return StaticParkingSiteInput(
+ uid=self.id,
+ name=self.long_name,
+ static_data_updated_at=self.time,
+ capacity=self.capacity.car,
+ capacity_charging=self.capacity.car_charging,
+ capacity_disabled=self.capacity.car_handicap,
+ capacity_woman=self.capacity.car_women,
+ lat=self.location.lat,
+ lon=self.location.lng,
+ )
+
+ def to_realtime_parking_site(self) -> RealtimeParkingSiteInput:
+ return RealtimeParkingSiteInput(
+ uid=self.id,
+ realtime_capacity=self.capacity.car,
+ realtime_capacity_charging=self.capacity.car_charging,
+ realtime_capacity_disabled=self.capacity.car_handicap,
+ realtime_capacity_woman=self.capacity.car_women,
+ realtime_free_capacity=self.free_capacity.car,
+ realtime_free_capacity_charging=self.free_capacity.car_charging,
+ realtime_free_capacity_disabled=self.free_capacity.car_handicap,
+ realtime_free_capacity_woman=self.free_capacity.car_women,
+ realtime_data_updated_at=self.time,
+ )

src/parkapi_sources/models/source_info.py

@@ -11,9 +11,9 @@
 class SourceInfo:
  uid: str
  name: str
- public_url: str
  has_realtime_data: Optional[bool]
  timezone: str = 'Europe/Berlin'
+ public_url: Optional[str] = None
  source_url: Optional[str] = None
  attribution_license: Optional[str] = None
  attribution_url: Optional[str] = None

src/parkapi_sources/parkapi_sources.py

@@ -6,6 +6,7 @@
 from typing import Optional, Type
 
 from .converters import (
+ A81PMPullConverter,
  BahnV2PullConverter,
  BaseConverter,
  BfrkBwOepnvBikePushConverter,
@@ -41,6 +42,7 @@
 
 class ParkAPISources:
  converter_classes: list[Type[BaseConverter]] = [
+ A81PMPullConverter,
  BahnV2PullConverter,
  BfrkBwOepnvBikePushConverter,
  BfrkBwOepnvCarPushConverter,

tests/converters/a81_p_m_test.py

@@ -0,0 +1,59 @@
+"""
+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 pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+from parkapi_sources.converters import A81PMPullConverter
+from requests_mock import Mocker
+
+from tests.converters.helper import validate_realtime_parking_site_inputs, validate_static_parking_site_inputs
+
+
+@pytest.fixture
+def a81_p_m_config_helper(mocked_config_helper: Mock):
+ config = {
+ 'PARK_API_A81_P_M_TOKEN': '127d24d7-8262-479c-8e22-c0d7e093b147',
+ }
+ mocked_config_helper.get.side_effect = lambda key, default=None: config.get(key, default)
+ return mocked_config_helper
+
+
+@pytest.fixture
+def a81_p_m_pull_converter(a81_p_m_config_helper: Mock) -> A81PMPullConverter:
+ return A81PMPullConverter(config_helper=a81_p_m_config_helper)
+
+
+class A81PMConverterTest:
+ @staticmethod
+ def test_get_static_parking_sites(a81_p_m_pull_converter: A81PMPullConverter, requests_mock: Mocker):
+ json_path = Path(Path(__file__).parent, 'data', 'a81_p_m.json')
+ with json_path.open() as json_file:
+ json_data = json_file.read()
+
+ requests_mock.get('https://api.cloud-telartec.de/v1/parkings', text=json_data)
+
+ static_parking_site_inputs, import_parking_site_exceptions = a81_p_m_pull_converter.get_static_parking_sites()
+
+ assert len(static_parking_site_inputs) == 2
+ assert len(import_parking_site_exceptions) == 0
+
+ validate_static_parking_site_inputs(static_parking_site_inputs)
+
+ @staticmethod
+ def test_get_realtime_parking_sites(a81_p_m_pull_converter: A81PMPullConverter, requests_mock: Mocker):
+ json_path = Path(Path(__file__).parent, 'data', 'a81_p_m.json')
+ with json_path.open() as json_file:
+ json_data = json_file.read()
+
+ requests_mock.get('https://api.cloud-telartec.de/v1/parkings', text=json_data)
+
+ static_parking_site_inputs, import_parking_site_exceptions = a81_p_m_pull_converter.get_realtime_parking_sites()
+
+ assert len(static_parking_site_inputs) == 2
+ assert len(import_parking_site_exceptions) == 0
+
+ validate_realtime_parking_site_inputs(static_parking_site_inputs)

tests/converters/data/a81_p_m.json

@@ -0,0 +1 @@
+[{"id":"fcde2f18-c277-11ed-b3ce-0050563bb33b","name":"A81-Ergenzingen","long_name":"A81 Ergenzingen","status":"OFFLINE","category":"P&M","street":"A81","free_capacity":{"truck":0,"bus":0,"car":80,"car_women":0,"car_charging":0,"car_handicap":4},"capacity":{"truck":0,"bus":0,"car":111,"car_women":0,"car_charging":0,"car_handicap":4},"time":"2024-03-30 03:19:46","location":{"lat":48.50571,"lng":8.82799},"geometry":"{\"type\":\"FeatureCollection\",\"features\":[{\"type\":\"Feature\",\"properties\":{\"name\":\"P+M Ergenzingen\",\"id\":\"fcde2f18-c277-11ed-b3ce-0050563bb33b\"},\"geometry\":{\"coordinates\":[[[8.828089,48.505406],[8.828942,48.505246],[8.82906,48.505346],[8.829123,48.505457],[8.82914,48.50558],[8.829119,48.50568],[8.828287,48.505851],[8.828089,48.505406]]],\"type\":\"Polygon\"}}]}"},{"id":"ff57e1d7-32c4-11ee-8f99-0050563bb33b","name":"A81 Mundelsheim","long_name":"A81 Mundelsheim","status":"OFFLINE","category":"P&M","street":"A81","free_capacity":{"truck":0,"bus":0,"car":48,"car_women":0,"car_charging":0,"car_handicap":0},"capacity":{"truck":0,"bus":0,"car":134,"car_women":0,"car_charging":0,"car_handicap":0},"time":"2024-03-30 13:59:51","location":{"lat":49.0055564,"lng":9.2372623},"geometry":"{\"type\":\"FeatureCollection\",\"features\":[{\"type\":\"Feature\",\"properties\":{\"name\":\"P+M Mundelsheim\",\"id\":\"ff57e1d7-32c4-11ee-8f99-0050563bb33b\"},\"geometry\":{\"coordinates\":[[[9.237320561279802,49.005705067924794],[9.236095227840138,49.00553651342898],[9.236043243996733,49.00535042080847],[9.23621998906225,49.00522960637886],[9.237431955228317,49.00540011052908],[9.237320561279802,49.005705067924794]]],\"type\":\"Polygon\"}}]}"}]