first datex2 approach

dev/api_tests/public_api/datex2.http

@@ -0,0 +1,2 @@
+### List datex2 parking sites
+GET {{api_host}}/api/public/datex2/json

webapp/common/dataclass.py

@@ -0,0 +1,31 @@
+"""
+Copyright 2023 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from dataclasses import is_dataclass, asdict
+from typing import Any
+
+from validataclass.helpers import UnsetValue
+
+
+def recursive_to_dict(data: Any):
+    if is_dataclass(data):
+        return {key: recursive_to_dict(value) for key, value in asdict(data).items()}
+    if isinstance(data, list):
+        return [recursive_to_dict(item) for item in data]
+    return data
+
+
+def filter_unset_value(data: Any) -> Any:
+    if isinstance(data, dict):
+        return {key: filter_unset_value(value) for key, value in data.items() if value is not UnsetValue}
+    if isinstance(data, list):
+        return [filter_unset_value(item) for item in data if item is not UnsetValue]
+    return data
+
+
+class DataclassMixin:
+
+    def to_dict(self):
+        return filter_unset_value(recursive_to_dict(self))

webapp/public_rest_api/datex2/datex2_handler.py

@@ -0,0 +1,35 @@
+"""
+Copyright 2023 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from webapp.public_rest_api.datex2.datex2_mapper import Datex2Mapper
+from webapp.public_rest_api.datex2.datex2_models import Datex2Publication
+from webapp.repositories import SourceRepository
+from webapp.shared.parking_site import GenericParkingSiteHandler
+from webapp.shared.parking_site.parking_site_search_query import ParkingSiteSearchInput
+
+
+class Datex2Handler(GenericParkingSiteHandler):
+    source_repository: SourceRepository
+    datex2_mapper: Datex2Mapper = Datex2Mapper()
+
+    def __init__(self, *args, source_repository: SourceRepository, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.source_repository = source_repository
+
+    def get_parking_sites(self, search_query: ParkingSiteSearchInput) -> Datex2Publication:
+        parking_sites = self.get_parking_site_list(search_query)
+
+        if search_query.source_uid is None:
+            name = 'Aggregated parking sites'
+        else:
+            source = self.source_repository.fetch_source_by_uid(search_query.source_uid)
+            name = source.name
+
+        datex_2_publication = self.datex2_mapper.map_parking_sites(
+            parking_sites=parking_sites,
+            name=name,
+        )
+
+        return datex_2_publication

webapp/public_rest_api/datex2/datex2_mapper.py

@@ -0,0 +1,70 @@
+"""
+Copyright 2023 binary butterfly GmbH
+Use of this source code is governed by an MIT-style license that can be found in the LICENSE.txt.
+"""
+
+from webapp.models import ParkingSite
+from webapp.models.parking_site import OpeningStatus, ParkingSiteType
+from webapp.public_rest_api.datex2.datex2_models import Datex2ParkingSite, Datex2LocationAndDimension, Datex2Coordinate, \
+    Datex2ParkingSiteType, Datex2Publication, Datex2ParkingPublicationLight
+
+
+class Datex2Mapper:
+
+    def map_parking_sites(self, name: str, parking_sites: list[ParkingSite]) -> Datex2Publication:
+        publication = Datex2Publication(
+            parkingPublicationLight=Datex2ParkingPublicationLight(
+                name=name,
+                parkingSite=[self.map_parking_site(parking_site) for parking_site in parking_sites],
+            ),
+        )
+
+        return publication
+
+    def map_parking_site(self, parking_site: ParkingSite) -> Datex2ParkingSite:
+        datex2_parking_site = Datex2ParkingSite(
+            uid=str(parking_site.id),
+            type=self.map_parking_site_type(parking_site.type),
+            locationAndDimension=Datex2LocationAndDimension(
+                coordinatesForDisplay=Datex2Coordinate(
+                    latitude=float(parking_site.lat),
+                    longitude=float(parking_site.lon),
+                ),
+            ),
+        )
+
+        if parking_site.has_realtime_data:
+            if parking_site.realtime_capacity is not None:
+                datex2_parking_site.availableSpaces = parking_site.realtime_capacity
+
+            if parking_site.realtime_opening_status == OpeningStatus.OPEN:
+                datex2_parking_site.isOpenNow = True
+            elif parking_site.realtime_opening_status == OpeningStatus.CLOSED:
+                datex2_parking_site.isOpenNow = False
+
+        field_mapping = {
+            'description': 'description',
+            'modified': 'lastUpdate',
+            'max_stay': 'maximumParkingDuration',
+            'name': 'name',
+            'capacity': 'numberOfSpaces',
+            'fee_description': 'tariffDescription',
+            'public_url': 'urlLinkAddress',
+        }
+
+        for source, target in field_mapping.items():
+            if getattr(parking_site, source, None) is None:
+                continue
+            setattr(datex2_parking_site, target, getattr(parking_site, source))
+
+        return datex2_parking_site
+
+    @staticmethod
+    def map_parking_site_type(parking_site_type: ParkingSiteType) -> Datex2ParkingSiteType:
+        return {
+            ParkingSiteType.CAR_PARK: Datex2ParkingSiteType.carPark,
+            ParkingSiteType.ON_STREET: Datex2ParkingSiteType.onStreet,
+            ParkingSiteType.UNDERGROUND: Datex2ParkingSiteType.carPark,
+            ParkingSiteType.OFF_STREET_PARKING_GROUND: Datex2ParkingSiteType.offStreetParkingGround,
+
+        }.get(parking_site_type, Datex2ParkingSiteType.other)