Nextbike: vehicle_types_available fill-in

app/converters/gbfs_nextbike_vehicle_availabilities.py

@@ -0,0 +1,67 @@
+"""
+MobiData BW Proxy
+Copyright (c) 2023, binary butterfly GmbH
+All rights reserved.
+"""
+
+from typing import Any
+
+from app.base_converter import BaseConverter
+from app.utils.gbfs_util import update_stations_availability_status
+
+
+class GbfsNextbikeVehicleAvailabilityConverter(BaseConverter):
+    """
+    Nextbike's station_status feeds currently don't provide the vehicle_types_available property.
+    But their vehicles at stations have a station_id assigned.
+
+    This Converter counts the number of vehicles per vehicle_type_id at each station and
+    constructs the vehicle_types_available from this information.
+
+    In cases not a single vehicle is assigned to a station, we add
+    a single vehicle_types_id with count == 0 as vehicle_types_available.
+
+    Note that this is a workaround and might be a vehicle_type that will never be
+    available at this station, or a vehicle_type which could be available sometimes
+    will not appear in vehicle_types_available.
+    """
+
+    hostnames = ['gbfs.nextbike.net']
+
+    free_vehicles_cache_per_system: dict[str, list[dict]] = {}
+
+    def convert(self, data: dict | list, path: str) -> dict | list:
+        if (
+            not isinstance(data, dict)
+            or not path.startswith('/maps/gbfs/v2/')
+            or not (path.endswith('/station_status.json') or path.endswith('/free_bike_status.json'))
+        ):
+            return data
+
+        system_id = self._get_system_id_from_path(path)
+
+        if path.endswith('/free_bike_status.json'):
+            return self._convert_free_vehicle_status(system_id, data, path)
+
+        return self._convert_station_status(system_id, data, path)
+
+    @staticmethod
+    def _get_system_id_from_path(path: str) -> str:
+        return path.split('/')[-3:-2][0]
+
+    def _convert_free_vehicle_status(self, system_id: str, data: dict, path: str) -> dict:
+        # cache vehicles per feed
+        vehicles = data.get('data', {}).get('bikes', [])
+        if isinstance(vehicles, list):
+            self.free_vehicles_cache_per_system[system_id] = vehicles
+        return data
+
+    def _convert_station_status(self, system_id: str, data: dict, path: str) -> dict:
+        if not data.get('data', {}).get('stations'):
+            return data
+
+        vehicles = self.free_vehicles_cache_per_system.get(system_id, [])
+
+        update_stations_availability_status(data['data']['stations'], vehicles)
+
+        return data

app/utils/gbfs_util.py

@@ -0,0 +1,93 @@
+"""
+MobiData BW Proxy
+Copyright (c) 2023, systect Holger Bruch
+All rights reserved.
+"""
+import logging
+from collections import Counter
+from typing import Any, Callable, Dict, List, Optional
+
+
+def update_stations_availability_status(station_status: List[Dict], vehicles: List[Dict]) -> None:
+    """
+    Updates station_status' vehicle_types_available and num_bikes_available.
+    A vehicle_type is available at a station, when any vehicle of it's type
+    is assigned to this station. However, for the availabilty count,
+    only those vehicles not reserved and not disabled are taken into account.
+    """
+
+    status_map = {station['station_id']: station for station in station_status}
+    station_vehicle_type_free_cnt = _count_vehicle_types_at_station(
+        vehicles, lambda v: not v['is_reserved'] and not v['is_disabled'] and 'station_id' in v
+    )
+    station_vehicle_type_cnt = _count_vehicle_types_at_station(vehicles, lambda v: 'station_id' in v)
+
+    vehicle_types_per_station: Dict[str, list] = {}
+    for station_vehicle_type in station_vehicle_type_cnt:
+        station_id = station_vehicle_type[0]
+
+        if station_id not in vehicle_types_per_station:
+            vehicle_types_per_station[station_id] = []
+
+        vehicle_types_per_station[station_id].append(
+            {
+                'vehicle_type_id': station_vehicle_type[1],
+                'count': station_vehicle_type_free_cnt.get(station_vehicle_type, 0),
+            }
+        )
+
+    for station_id in vehicle_types_per_station.keys():
+        if station_id in status_map:
+            _update_station_availability_status(vehicle_types_per_station[station_id], status_map[station_id])
+
+    default_vehicle_types_available = [{'vehicle_type_id': vehicles[0]['vehicle_type_id'], 'count': 0}]
+    for station in station_status:
+        if 'vehicle_types_available' not in station:
+            station['vehicle_types_available'] = default_vehicle_types_available
+
+
+def _count_vehicle_types_at_station(vehicles: list[dict[str, Any]], filter: Callable[[dict], bool]) -> Counter:
+    """
+    Count vehicle's per vehicle_type and station, which fulfill the filter critera.
+    """
+    filtered_vehicle_map = {v['bike_id']: v for v in vehicles if filter(v)}
+    station_vehicle_type_arr = [(v['station_id'], v['vehicle_type_id']) for v in filtered_vehicle_map.values()]
+
+    return Counter(station_vehicle_type_arr)
+
+
+def _update_station_availability_status(vt_available: List[Dict[str, Any]], station_status: Dict[str, Any]) -> None:
+    """
+    Sets station_status.vehicle_types_available and
+    calculates num_bikes_available as the sum of all vehicle_types_available.
+    Retains pre-existing vehicle_types_available (usually having count 0)
+    for vehicle_type_ids without available vehicles,
+    as this is the only way to find out, if vehicles are for rent at this station.
+    """
+    num_bikes_available = sum([vt['count'] for vt in vt_available])
+    if 'num_bikes_available' in station_status:
+        if num_bikes_available != station_status['num_bikes_available']:
+            logging.warn(
+                f"Official num_bikes_available ({station_status['num_bikes_available']}) does not match count deduced "
+                + f" from vehicle_types_available ({num_bikes_available}) at stationn {station_status['station_id']}"
+            )
+    else:
+        station_status['num_bikes_available'] = num_bikes_available
+
+    station_status['vehicle_types_available'] = _merge_vehicle_types_available(vt_available, station_status.get('vehicle_types_available'))
+
+
+def _merge_vehicle_types_available(
+    vt_available: List[Dict[str, Any]], pre_existing_vt: Optional[List[Dict[str, Any]]]
+) -> List[Dict[str, Any]]:
+    """
+    Merges vehicle_types_available lists.
+    """
+    if not pre_existing_vt:
+        return vt_available
+
+    # convert both to map, merge, reconvert to list
+    vt_map = {vt['vehicle_type_id']: vt for vt in vt_available}
+    vt_map_fallback = {vt['vehicle_type_id']: vt for vt in pre_existing_vt}
+    vt_merged = {**vt_map_fallback, **vt_map}
+    return list(vt_merged.values())