support bietigheim bissingen

src/parkapi_sources/converters/__init__.py

@@ -7,6 +7,7 @@
 from .bahn_v2 import BahnV2PullConverter
 from .base_converter import BaseConverter
 from .bfrk_bw import BfrkBwOepnvBikePushConverter, BfrkBwOepnvCarPushConverter, BfrkBwSpnvBikePushConverter, BfrkBwSpnvCarPushConverter
+from .bietigheim_bissingen import BietigheimBissingenPullConverter
 from .ellwangen import EllwangenPushConverter
 from .freiburg import FreiburgPullConverter
 from .heidelberg import HeidelbergPullConverter

src/parkapi_sources/converters/bietigheim_bissingen/__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 BietigheimBissingenPullConverter

src/parkapi_sources/converters/bietigheim_bissingen/converter.py

@@ -0,0 +1,116 @@
+"""
+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 email
+from csv import DictReader
+from email import policy
+from email.message import Message
+from imaplib import IMAP4_SSL
+from io import StringIO
+
+from validataclass.exceptions import ValidationError
+from validataclass.validators import DataclassValidator
+
+from parkapi_sources.converters.base_converter.pull import PullConverter, StaticGeojsonDataMixin
+from parkapi_sources.exceptions import ImportParkingSiteException, ImportSourceException
+from parkapi_sources.models import SourceInfo, RealtimeParkingSiteInput, StaticParkingSiteInput
+from .models import BietigheimBissingenInput
+
+
+class BietigheimBissingenPullConverter(PullConverter, StaticGeojsonDataMixin):
+ _imap_host: str = 'imap.strato.de'
+ required_config_keys = ['PARK_API_BIETIGHEIM_BISSINGEN_USER', 'PARK_API_BIETIGHEIM_BISSINGEN_PASSWORD']
+ bietigheim_bissingen_realtime_update_validator = DataclassValidator(BietigheimBissingenInput)
+ source_info = SourceInfo(
+ uid='bietigheim_bissingen',
+ name='Stadt Bietigheim-Bissingen',
+ public_url='https://www.bietigheim-bissingen.de/wirtschaft-verkehr-einkaufen/mobilitaet/',
+ timezone='Europe/Berlin',
+ 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)
+
+ def get_realtime_parking_sites(self) -> tuple[list[RealtimeParkingSiteInput], list[ImportParkingSiteException]]:
+ realtime_parking_site_inputs: list[RealtimeParkingSiteInput] = []
+ import_parking_site_exceptions: list[ImportParkingSiteException] = []
+
+ for row_dict in self._get_data():
+ try:
+ realtime_input = self.bietigheim_bissingen_realtime_update_validator.validate(row_dict)
+ except ValidationError as e:
+ import_parking_site_exceptions.append(
+ ImportParkingSiteException(
+ source_uid=self.source_info.uid,
+ parking_site_uid=row_dict.get('Name'),
+ message=f'Invalid data at uid {row_dict.get("Name")}: {e.to_dict()}, ' f'data: {row_dict}',
+ ),
+ )
+ continue
+
+ realtime_parking_site_inputs.append(
+ realtime_input.to_realtime_parking_site_input(
+ realtime_data_updated_at=realtime_input.data.updated,
+ ),
+ )
+
+ return realtime_parking_site_inputs, import_parking_site_exceptions
+
+ def _get_data(self) -> list[dict]:
+ with IMAP4_SSL(self._imap_host) as imap_connection:
+ imap_connection.login(
+ self.config_helper.get('PARK_API_BIETIGHEIM_BISSINGEN_USER'),
+ self.config_helper.get('PARK_API_BIETIGHEIM_BISSINGEN_PASSWORD'),
+ )
+ # Select default mailbox and get latest message uid
+ select_status, message_uid_list = imap_connection.select()
+ if len(message_uid_list) == 0:
+ raise ImportSourceException(
+ source_uid=self.source_info.uid,
+ message=f'No last email in inbox: {message_uid_list}.',
+ )
+
+ # Fetch the last mail
+ message_uid = message_uid_list[0]
+ imap_response: type[str, list] = imap_connection.fetch(message_uid, '(RFC822)')
+
+ # Get raw_messages and check if there is one
+ status, raw_messages = imap_response
+ if len(raw_messages) == 0:
+ raise ImportSourceException(
+ source_uid=self.source_info.uid,
+ message=f'No email in imap response although imap server provided status {status}.',
+ )
+
+ # If the raw message is no tuple, it's not a message
+ raw_message: tuple[bytes, bytes] = raw_messages[0]
+ if not isinstance(raw_message, tuple):
+ raise ImportSourceException(
+ source_uid=self.source_info.uid,
+ message=f'No valid email in imap response: {raw_message}.',
+ )
+
+ # The raw message has an envelope and a body, we just need the body
+ mail_envelope, mail_body = raw_message
+ message: Message = email.message_from_bytes(mail_body, policy=policy.default.clone(linesep='\r\n'))
+
+ return self._parse_csv(self._get_csv_bytes_from_message(message))
+
+ @staticmethod
+ def _parse_csv(csv_data: bytes) -> list[dict]:
+ csv_rows = DictReader(StringIO(csv_data.decode('latin1')), delimiter=';')
+
+ return [row for row in csv_rows] # type: ignore
+
+ def _get_csv_bytes_from_message(self, message: Message) -> bytes:
+ for message_part in message.walk():
+ if message_part.get_content_type() == 'application/octet-stream':
+ return message_part.get_payload(decode=True)
+
+ raise ImportSourceException(
+ source_uid=self.source_info.uid,
+ message=f'No valid attachment found in message {message}.',
+ )

src/parkapi_sources/converters/bietigheim_bissingen/models.py

@@ -0,0 +1,48 @@
+"""
+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
+from enum import Enum
+
+from validataclass.dataclasses import validataclass
+from validataclass.exceptions import ValidationError
+from validataclass.validators import StringValidator, IntegerValidator, EnumValidator
+
+from parkapi_sources.models import RealtimeParkingSiteInput
+from parkapi_sources.models.enums import OpeningStatus
+from parkapi_sources.validators import TimestampDateTimeValidator
+
+
+class BietigheimBissingenOpeningStatus(Enum):
+ OPEN = 'Geöffnet'
+ CLOSED = 'Geschlossen'
+
+ def to_realtime_opening_status(self) -> OpeningStatus:
+ return {
+ self.OPEN: OpeningStatus.OPEN,
+ self.CLOSED: OpeningStatus.CLOSED,
+ }.get(self, OpeningStatus.UNKNOWN)
+
+
+@validataclass
+class BietigheimBissingenInput:
+ Name: str = StringValidator()
+ OpeningState: BietigheimBissingenOpeningStatus = EnumValidator(BietigheimBissingenOpeningStatus)
+ Capacity: int = IntegerValidator(allow_strings=True)
+ OccupiedSites: int = IntegerValidator(allow_strings=True)
+ Timestamp: datetime = TimestampDateTimeValidator(allow_strings=True)
+
+ def __post_init__(self):
+ if self.Capacity < self.OccupiedSites:
+ raise ValidationError(reason='More occupied sites than capacity')
+
+ def to_realtime_parking_site_input(self) -> RealtimeParkingSiteInput:
+ return RealtimeParkingSiteInput(
+ uid=self.Name,
+ realtime_opening_status=self.OpeningState.to_realtime_opening_status(),
+ realtime_capacity=self.Capacity,
+ realtime_free_capacity=self.Capacity - self.OccupiedSites,
+ realtime_data_updated_at=self.Timestamp,
+ )

src/parkapi_sources/converters/heidelberg/converter.py

@@ -61,7 +61,7 @@ def get_realtime_parking_sites(self) -> tuple[list[RealtimeParkingSiteInput], li
  ImportParkingSiteException(
  source_uid=self.source_info.uid,
  parking_site_uid=update_dict.get('parkinglocation'),
- message=f'Invallid data at uid {update_dict.get("parkinglocation")}: {e.to_dict()}, ' f'data: {update_dict}',
+ message=f'Invalid data at uid {update_dict.get("parkinglocation")}: {e.to_dict()}, ' f'data: {update_dict}',
  ),
  )
  continue

src/parkapi_sources/parkapi_sources.py

@@ -13,6 +13,7 @@
  BfrkBwOepnvCarPushConverter,
  BfrkBwSpnvBikePushConverter,
  BfrkBwSpnvCarPushConverter,
+ BietigheimBissingenPullConverter,
  BuchenPushConverter,
  EllwangenPushConverter,
  FreiburgPullConverter,
@@ -48,6 +49,7 @@ class ParkAPISources:
  BfrkBwOepnvCarPushConverter,
  BfrkBwSpnvBikePushConverter,
  BfrkBwSpnvCarPushConverter,
+ BietigheimBissingenPullConverter,
  EllwangenPushConverter,
  BuchenPushConverter,
  FreiburgPullConverter,

src/parkapi_sources/validators/__init__.py

@@ -5,7 +5,7 @@
 
 from .boolean_validators import MappedBooleanValidator
 from .date_validator import ParsedDateValidator
-from .datetime_validator import Rfc1123DateTimeValidator, SpacedDateTimeValidator
+from .datetime_validator import Rfc1123DateTimeValidator, SpacedDateTimeValidator, TimestampDateTimeValidator
 from .decimal_validators import GermanDecimalValidator
 from .integer_validators import GermanDurationIntegerValidator
 from .list_validator import PointCoordinateTupleValidator

src/parkapi_sources/validators/datetime_validator.py

@@ -8,7 +8,7 @@
 from typing import Any
 
 from validataclass.exceptions import ValidationError
-from validataclass.validators import DateTimeValidator, StringValidator
+from validataclass.validators import DateTimeValidator, StringValidator, IntegerValidator
 
 
 class Rfc1123DateTimeValidator(StringValidator):
@@ -30,3 +30,13 @@ def validate(self, input_data: Any, **kwargs) -> datetime:
  input_data = f'{input_data[:10]}T{input_data[11:]}'
 
  return super().validate(input_data, **kwargs)
+
+
+class TimestampDateTimeValidator(IntegerValidator):
+ def validate(self, input_data: Any, **kwargs) -> datetime:
+ input_data = super().validate(input_data, **kwargs)
+
+ try:
+ return datetime.fromtimestamp(input_data, tz=timezone.utc)
+ except ValueError as e:
+ raise ValidationError(reason='Invalid timestamp') from e

tests/converters/bietigheim_bissingen_test.py

@@ -0,0 +1,42 @@
+"""
+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 import BietigheimBissingenPullConverter
+from tests.converters.helper import validate_static_parking_site_inputs, validate_realtime_parking_site_inputs
+
+
+@pytest.fixture
+def bietigheim_bissingen_config_helper(mocked_config_helper: Mock):
+ config = {
+ 'PARK_API_BIETIGHEIM_BISSINGEN_USER': '0152d634-9e16-46c0-bfef-20c0b623eaa3',
+ 'PARK_API_BIETIGHEIM_BISSINGEN_PASSWORD': 'eaf7a00c-d0e1-4464-a9dc-f8ef4d01f2cc',
+ }
+ mocked_config_helper.get.side_effect = lambda key, default=None: config.get(key, default)
+ return mocked_config_helper
+
+
+@pytest.fixture
+def bietigheim_bissingen_pull_converter(bietigheim_bissingen_config_helper: Mock) -> BietigheimBissingenPullConverter:
+ return BietigheimBissingenPullConverter(config_helper=bietigheim_bissingen_config_helper)
+
+
+class BietigheimBissingenPullConverterTest:
+ @staticmethod
+ def test_get_static_parking_sites(bietigheim_bissingen_pull_converter: BietigheimBissingenPullConverter):
+
+ static_parking_site_inputs, import_parking_site_exceptions = bietigheim_bissingen_pull_converter.get_static_parking_sites()
+
+ validate_static_parking_site_inputs(static_parking_site_inputs)
+
+ @staticmethod
+ def test_get_realtime_parking_sites(bietigheim_bissingen_pull_converter: BietigheimBissingenPullConverter):
+
+ realtime_parking_site_inputs, import_parking_site_exceptions = bietigheim_bissingen_pull_converter.get_realtime_parking_sites()
+
+ validate_realtime_parking_site_inputs(realtime_parking_site_inputs)