-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #55 from ParkenDD/a81-p-m
Feature: A81 P&M Converter
- Loading branch information
Showing
9 changed files
with
228 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,72 @@ | ||
""" | ||
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] = [] | ||
|
||
a81_p_m_inputs, static_parking_site_errors = self._get_data() | ||
|
||
for a81_p_m_input in a81_p_m_inputs: | ||
static_parking_site_inputs.append(a81_p_m_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] = [] | ||
|
||
a81_p_m_inputs, realtime_parking_site_errors = self._get_data() | ||
|
||
for a81_p_m_input in a81_p_m_inputs: | ||
realtime_parking_site_inputs.append(a81_p_m_input.to_realtime_parking_site()) | ||
|
||
return realtime_parking_site_inputs, realtime_parking_site_errors | ||
|
||
def _get_data(self) -> tuple[list[A81PMInput], list[ImportParkingSiteException]]: | ||
a81_p_m_inputs: list[A81PMInput] = [] | ||
parking_site_errors: list[ImportParkingSiteException] = [] | ||
|
||
response = requests.get( | ||
self.source_info.source_url, | ||
headers={'Authorization': f'Bearer {self.config_helper.get("PARK_API_A81_P_M_TOKEN")}'}, | ||
timeout=60, | ||
) | ||
|
||
for input_dict in self.list_validator.validate(response.json()): | ||
try: | ||
a81_p_m_inputs.append(self.a81_p_m_site_validator.validate(input_dict)) | ||
except ValidationError as e: | ||
parking_site_errors.append( | ||
ImportParkingSiteException( | ||
source_uid=self.source_info.uid, | ||
parking_site_uid=input_dict.get('id'), | ||
message=f'validation error for static data {input_dict}: {e.to_dict()}', | ||
), | ||
) | ||
|
||
return a81_p_m_inputs, parking_site_errors |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,85 @@ | ||
""" | ||
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.models.enums import ParkAndRideType | ||
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) # TODO: what's that? | ||
time: datetime = SpacedDateTimeValidator( | ||
local_timezone=ZoneInfo('Europe/Berlin'), | ||
target_timezone=timezone.utc, | ||
) | ||
location: A81PMLocationInput = DataclassValidator(A81PMLocationInput) | ||
capacity: A81PMCapacityInput = DataclassValidator(A81PMCapacityInput) | ||
category: A81PMCategory = EnumValidator(A81PMCategory) | ||
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, | ||
park_and_ride_type=[ParkAndRideType.YES] if self.category == A81PMCategory.P_M else None, | ||
) | ||
|
||
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, | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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\"}}]}"}] |