Merge pull request #79 from leonardehrenfried/noi

Add provider for NOI

Makefile

@@ -0,0 +1,15 @@
+MAIN_MODULE=./x2gbfs
+
+.PHONY: lint-fix
+lint-fix:
+	ruff --fix ${MAIN_MODULE}
+	black ${MAIN_MODULE}
+	# mypy has no fix mode, we run it anyway to report (unfixable) errors
+	mypy ${MAIN_MODULE}
+
+.PHONY: lint-check
+lint-check:
+	$(PROXY_RUN) ruff ${MAIN_MODULE}
+	$(PROXY_RUN) black -S --check --diff ${MAIN_MODULE}
+	mypy ${MAIN_MODULE}
+

README.md

@@ -9,6 +9,7 @@ Currently supported providers:
 * Stadtmobil Südbaden (via Cantamen IXSI API, provider id: `stadtmobil_suedbaden`)
 * my-e-car (via Cantamen IXSI API, provider id: `my-e-car`)
 * Lastenvelo Freiburg (via custom CSV, provider id: `lastenvelo_fr`)
+* NOI OpenDataHub (provider id: `noi`)
 
 To generate a feed for e.g. deer network, switch to the `x2gbfs` project basee dir and execute
 
@@ -70,6 +71,11 @@ To generate the voi raumobil GBFS feed, you need to provide the following enviro
 * `VOI_API_URL=https://url.org`
 * `VOI_USER=<your username>`
 * `VOI_PASSWORD=<your password>`
+
+### NOI OpenDataHub
+
+Converts NOI's OpenDataHub carsharing feed to GBFS.
+
 
 ## Implementing a new provider
 To implement a new provider, you should take the following steps:

config/noi.json

@@ -0,0 +1,23 @@
+{
+    "feed_data": {
+        "system_information": {
+            "email": "[email protected]",
+            "feed_contact_email": "[email protected]",
+            "language": "de",
+            "license_url": "https://creativecommons.org/publicdomain/zero/1.0/legalcode",
+            "license_id": "CC0",
+            "attribution_organization_name": "Konsortialgenossenschaft Car Sharing Südtirol Alto Adige",
+            "attribution_url": "https://www.carsharing.bz.it",
+            "name": "Carsharing Südtirol Alto Adige",
+            "operator": "Konsortialgenossenschaft Car Sharing Südtirol Alto Adige",
+            "phone_number": "+39 0471 061319",
+            "privacy_url": "https://www.carsharing.bz.it/de/privacy/",
+            "purchase_url": "https://www.carsharing.bz.it/",
+            "system_id": "carsharing-suedtirol-alto-adige",
+            "terms_last_updated": "2022-02-15",
+            "terms_url": "https://www.carsharing.bz.it/de/impressum/",
+            "timezone": "Europe/Rome",
+            "url": "https://www.carsharing.bz.it/de/"
+        }
+    }
+}

x2gbfs/providers/__init__.py

@@ -4,4 +4,5 @@
 from .fleetster import FleetsterAPI
 from .lastenvelo_fr import LastenVeloFreiburgProvider
 from .moqo import StadtwerkTauberfrankenProvider
+from .noi import NoiProvider
 from .voi_raumobil import VoiRaumobil

x2gbfs/providers/noi.py

@@ -0,0 +1,100 @@
+import logging
+import re
+from typing import Dict, Optional, Tuple
+
+import requests
+
+from x2gbfs.gbfs.base_provider import BaseProvider
+
+logger = logging.getLogger(__name__)
+
+
+class NoiProvider(BaseProvider):
+    STATION_URL = 'https://mobility.api.opendatahub.com/v2/flat%2Cnode/CarsharingStation?limit=500&offset=0&shownull=false&distinct=true'
+    CAR_URL = 'https://mobility.api.opendatahub.com/v2/flat%2Cnode/CarsharingCar?limit=500&offset=0&shownull=false&distinct=true'
+
+    def __init__(self):
+        pass
+
+    def load_vehicles(self, default_last_reported: int) -> Tuple[Optional[Dict], Optional[Dict]]:
+        response = requests.get(self.CAR_URL, timeout=20)
+        raw_cars = response.json()['data']
+        types = {}
+        vehicles = {}
+        for i in raw_cars:
+
+            type_id = self.extract_type_id(i)
+            types[type_id] = {
+                # See https://github.com/MobilityData/gbfs/blob/v2.3/gbfs.md#vehicle_typesjson
+                'vehicle_type_id': type_id,
+                'form_factor': 'car',
+                'propulsion_type': self.extract_propulsion(i),
+                'max_range_meters': 500000,
+                'name': i['smetadata']['brand'].strip(),
+                'wheel_count': 4,
+                'return_constraint': 'roundtrip_station',
+            }
+
+            if i.get('pcode') is not None:
+                id = self.slugify(i['smetadata']['licensePlate'])
+                vehicles[id] = {
+                    'bike_id': id,
+                    'station_id': i['pcode'],
+                    'vehicle_type_id': type_id,
+                    'is_reserved': False,
+                    'is_disabled': False,
+                    'current_range_meters': 500000,
+                }
+
+        return types, vehicles
+
+    def extract_type_id(self, i):
+        return self.slugify(i['smetadata']['brand'])
+
+    def slugify(self, input):
+        output = input.lower().strip().replace('!', '')
+        output = re.sub(r'[^a-z0-9]+', '-', output)
+        return re.sub(r'-$', '', output)
+
+    def extract_propulsion(self, i):
+        if self.extract_type_id(i).__contains__('elektro'):
+            return 'electric'
+        return 'combustion'
+
+    def load_stations(self, default_last_reported: int) -> Tuple[Optional[Dict], Optional[Dict]]:
+
+        response = requests.get(self.STATION_URL, timeout=20)
+        raw_stations = response.json()['data']
+
+        info = {}
+        for i in raw_stations:
+            id = i['scode']
+            coord = i['scoordinate']
+            info[id] = {
+                'station_id': id,
+                'name': i['sname'],
+                'lon': coord['x'],
+                'lat': coord['y'],
+            }
+
+        status = self.load_status(default_last_reported)
+        return info, status
+
+    def load_status(self, last_reported: int) -> Optional[Dict]:
+        response = requests.get(self.CAR_URL, timeout=20)
+        raw_cars = response.json()['data']
+        statuses = {}
+
+        for i in raw_cars:
+
+            if i.get('pcode') is not None:
+
+                station_id = i['pcode']
+                statuses[station_id] = {
+                    'station_id': station_id,
+                    'is_installed': True,
+                    'is_renting': True,
+                    'is_returning': True,
+                    'last_reported': last_reported,
+                }
+        return statuses

x2gbfs/x2gbfs.py

@@ -16,6 +16,7 @@
     ExampleProvider,
     FleetsterAPI,
     LastenVeloFreiburgProvider,
+    NoiProvider,
     StadtwerkTauberfrankenProvider,
     VoiRaumobil,
 )
@@ -47,6 +48,8 @@ def build_extractor(provider: str, feed_config: Dict[str, Any]) -> BaseProvider:
         return CantamenIXSIProvider(feed_config)
     if provider in ['stadtwerk_tauberfranken']:
         return StadtwerkTauberfrankenProvider(feed_config)
+    if provider in ['noi']:
+        return NoiProvider()
 
     raise ValueError(f'Unknown config {provider}')