Mapa rowerów Mevo w Pythonie – warsztaty

Na początku lipca, w ramach współpracy z infoShare Academy, miałem przyjemność poprowadzić otwarty warsztat „Python od podstaw”. Zainteresowanie tematem okazało się ogromne – na zajęcia przyszło ponad 100 osób. Celem warsztatu było przekazanie „w pigułce” podstawowych informacji o języku oraz praktyczne sprawdzenie się w boju, podczas implementacji własnej aplikacji. Spotkanie trwało 3 godziny i zmieszczenie w tym limicie wstępu teoretycznego oraz pracy warsztatowej okazało się nie lada wyzwaniem. Po krótkim wprowadzeniu, wspólnie zmierzyliśmy się z zadaniem jakim była implementacja skryptu generującego mapę rowerów Mevo.

Aby nakreślić kontekst warsztatu pozwolę sobie na kilka słów wprowadzenia na temat Mevo ;). Jest to system rowerów miejskich funkcjonujący na obszarze Trójmiasta oraz okolicznych gmin. System nietypowy, bo wszystkie rowery zostały wyposażone we wspomaganie elektryczne. Jest to największy tego typu projekt w całej Europie. Aktualnie trwa realizacja drugiego etapu wdrożenia i dostępnych jest ok 1/3 wszystkich planowanych rowerów (łącznie ma być ich ponad 4 tys). System spotkał się z bardzo dużym zainteresowaniem mieszkańców, co w połączeniu z opóźnieniami w realizacji kolejny etapów powoduje, że czasami nie łatwo jest znaleźć wolny rower 😉

Aktualizacja [2020]
Oryginalny warsztat dotyczył rowerów Mevo. Niestety, w międzyczasie projekt ten został wstrzymany. Na moment obecny rowery Mevo nie są dostępne w ogóle. W związku z tym, przygotowałem aktualizację. Na końcu posta zamieściłem zmiany, które trzeba wprowadzić w omawianym przykładzie, żeby otrzymać mapę poznańskiego roweru.

Na oficjalnej stronie Mevo dostępna jest mapka stacji, wraz z informacją o wolnych rowerach. Planem na warsztaty było uzyskanie bardzo podobnej mapki pokazującej jednak na widoku oddalonym zagregowaną liczbę rowerów, a nie stacji (na których może nie być żadnego roweru). Dodatkowo kolor znacznika mógłby pokazać nam informacje o poziomie naładowania baterii.

Zadanie było dość wymagające, ale mając do dyspozycji bogaty zestaw Pythonowych bibliotek okazało się jak najbardziej wykonalne. Do implementacji wykorzystaliśmy bibliotekę folium, która w bardzo łatwy sposób pozwala nam na wizualizację danych z wykorzystaniem interaktywnej mapy. Za pomocą czterech linijek możemy wygenerować plik HTML-owy z mapą świata:

import folium

GDANSK_CENTER_POSITION = [54.346320, 18.649246]
bikes_map = folium.Map(location=GDANSK_CENTER_POSITION, zoom_start=10)
bikes_map.save('bikes_map.html')

W pierwszej wersji rozwiązania, dane o położeniu rowerów wczytywaliśmy z pobranego wcześniej ze strony Mevo pliku csv zawierającego współrzędne stacji i poszczególnych pojazdów. Użycie wbudowanego w Pythona modułu csv bardzo ułatwiło to zadanie:

import csv
...
with open(PATH_TO_LOCATIONS, mode='r') as locations_file:
    locations_reader = csv.DictReader(locations_file)

    for station_row in locations_reader:
        available_bikes = int(station_row['DOSTĘPNE ROWERY'])
        if available_bikes > 0:

            coordinates_str = station_row['WSPÓŁRZĘDNE']

Potem wystarczyło już tylko sparsować współrzędne zamieniąjąc rozdzielone przecinkiem długość i szerokość geograficzną na listę floatów oczekiwaną przez interfejs folium:

coordinates = coordinates_str.split(', ')
latitude = float(coordinates[0])
longitude = float(coordinates[1])
coordinates = [latitude, longitude]

I nanieść wyniki na mapę:

bike_info = f'{available_bikes} rowerów jest dostępnych'
bike_marker = folium.Marker(location=coordinates, popup=bike_info)
bikes_map.add_child(bike_marker)

W kolejnym etapie zajęliśmy się agregacją rowerów. Do tego celu wykorzystaliśmy klaster znaczników.

from folium.plugins import MarkerCluster
...
markers_cluster = MarkerCluster()

Dodając poszczególne rowery do klastra, a następnie cały klaster do mapy, uzyskujemy ładny efekt agregacji przy zmianie zooma:

for bike_id in available_bikes_ids:
    bike_info = f'ID: {bike_id}'
    bike_marker = folium.Marker(location=coordinates, popup=bike_info)
    markers_cluster.add_child(bike_marker)

bikes_map.add_child(markers_cluster)
bikes_map.save('bikes_map.html')

Identyfikatory poszczególnych rowerów wykorzystywane w dymku nad znacznikiem również pozyskaliśmy z tego samego pliku csv:

available_bikes_ids_str = station_row['NUMERY DOSTĘPNYCH ROWERÓW']
available_bikes_ids = available_bikes_ids_str.split(',')

Zmieniliśmy również ikonkę na ładniejszą – w końcu pokazujemy położenie rowerów, warto żeby znacznik był bardziej „rowerowy” 😉

bike_icon = folium.Icon(icon='bicycle', prefix='fa', color='green')
bike_marker = folium.Marker(
    location=coordinates, popup=bike_info, icon=bike_icon
)

Skąd wiemy jakie parametry powinniśmy podać, żeby otrzymać ulubioną ikonę? Otóż za pomocą prefixu określamy zestaw ikon (‚fa’ oznacza ‚Font Awesome’), a parametr icon wskazuje na konkretny obrazek (listę ikon ze zbioru Font Awesome znajdziemy tutaj)

Kolejnym krokiem było przedstawienie informacji o poziomie naładowania baterii za pomocą koloru znacznika z rowerem. W tym celu napisaliśmy funkcję określającą oczekiwany kolor znacznika na podstawie poziomu baterii (% pozostałej mocy):

def get_icon_color(battery_level):
    if battery_level is None:
        return 'gray'

    if battery_level > 50:
        return 'green'

    if battery_level > 20:
        return 'orange'

    return 'red'

Ponieważ dane o bateriach pobieramy z sieci, może okazać się, że informacje dotyczące danego roweru nie są aktualnie dostępne. Taką sytuację ogrywamy kolorem szarym, za pomocą pierwszego warunku w tej funkcji. Otrzymany z funkcji kolor wykorzystujemy do zabarwienia znacznika z rowerem:

icon_color = get_icon_color(battery_level)
bike_icon = folium.Icon(icon='bicycle', prefix='fa', color=icon_color)

Ok, ale skąd wziąć informacje o poziomie baterii? Niestety Mevo nie dostarcza sensownego API i konieczne jest scrapowanie strony w celu wyciągnięcia potrzebnych informacji. Ze względu na potrzebę takiej kombinacji ten moduł przygotowałem dla uczestników warsztatów wcześniej. Dane o poziomie baterii znajdują się w pliku locations.js, który można pobrać z serwisu jednak wymaga on od nas przesłania w żądaniu odpowiednich nagłówków oraz klucza. Klucz jest ważny tylko przez pewien czas, zaś aktualny można „wyciągnąć” z kodu strony serwisu Mevo. Do implementacji tej części wykorzystałem pakiet requests, który świetnie sprawdza się przy wysyłaniu żądań HTTP/HTTPS.

import requests

SERVICE_URL = 'https://rowermevo.pl/mapa-stacji/'
LOCATIONS_JS_KEY_PATTERN = 'src="/locations.js?key='
LOCATIONS_JS_URL_BASE = 'https://rowermevo.pl/locations.js?key='
HEADERS = {
    'cookie': 'cookies-info=1',
    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36'
}

service_page_content = requests.get(
    SERVICE_URL, headers=HEADERS
).content.decode('utf-8')
locations_js_key = get_key_to_resource_from_service_page(
    service_page_content, LOCATIONS_JS_KEY_PATTERN
)
locations_js_url = f'{LOCATIONS_JS_URL_BASE}{locations_js_key}'
locations_response = requests.get(locations_js_url, headers=HEADERS)
batteries_data = parse_response_to_batteries_data(locations_response)

Absolutnie nie jest to „kulooporny” kod, który poradziłby sobie z wyzwaniami stawianymi tego typu aplikacjom w trybie produkcyjnym. Zarówno sam sposób parsowania treści strony jak i przetwarzanie odpowiedzi są bardzo wrażliwe na wszelkie zmiany struktury i możliwe niepowodzenia. W tym przypadku nie komplikujemy dodatkowo tematu i zakładamy, że wszystko pójdzie dobrze 😉 Wyszukiwanie klucza w treści strony oraz parsowanie danych o poziomie baterii wygląda następująco:

import json
...

def get_key_to_resource_from_service_page(
    service_page_content, key_pattern
):
    src_with_key_index = service_page_content.find(key_pattern)
    key_part_str = service_page_content[
        src_with_key_index + len(key_pattern):
    ]
    return key_part_str.split('"')[0]

def parse_response_to_batteries_data(locations_response):
    locations_data = locations_response.content.decode('utf-8')
    batteries_data_line = locations_data.split(';')[1]
    batteries_data_str = batteries_data_line.split("'")[1]
    batteries_data_str = '{"data":' + batteries_data_str + '}'
    return json.loads(batteries_data_str)

Pozyskane informacje zapisujemy w postaci słownika, aby łatwo móc zwrócić informacje o poziomie baterii dla danego identyfikatora roweru:

self.parsed_batteries_info = {}
    for bike_data in batteries_data['data']:
        self.parsed_batteries_info[
            bike_data['bike']
        ] = bike_data['battery']

Ponieważ może zdarzyć się, że nie posiadamy informacji o danym rowerze, warto wykorzystać konstrukcję pozwalającą na zwrócenie wartości domyślnej, kiedy podany klucz nie występuje w słowniku:

def battery_info_for_bike(self, bike_id):
    return self.parsed_batteries_info.get(bike_id, None)

Zapisane dane wykorzystujemy w przygotowanej już wcześniej funkcji określającej kolor znacznika:

from bike_service_proxy import BikeServiceProxy
...
bikes_proxy = BikeServiceProxy()
...
battery_level = bikes_proxy.battery_info_for_bike(bike_id)
icon_color = get_icon_color(battery_level)

Ostatnim krokiem było dodanie pobierania danych o położeniu rowerów w czasie rzeczywistym (również scrapując wcześniej stronę portalu w poszukiwaniu klucza):

bikes_file = bikes_proxy.current_locations_file
bikes_reader = csv.DictReader(bikes_file)

Implementacja takiego zadania podczas krótkiego warsztatu z pewnością nie byłaby możliwa gdyby nie niewyczerpana energia Łukasza Falkowicza, który bez chwili wytchnienia biegał pomiędzy stolikami i rozwiązywał poważniejsze problemy. Oczywiście największe brawa i wyrazy uznania należą się wszystkim uczestnikom warszatu. Dzięki wielkiemu zaangażowaniu i determinacji rozwiązali napotkane wyzwania i stworzyli naprawdę fajną i funkcjonalną aplikację w Pythonie. Dla mnie prowadzenie tego warsztatu i współpraca z tyloma świetnymi i pełnymi pasji ludźmi była niesamowicie energetyzującym wydarzeniem. Uznanie należy się również całej ekipie zaangażowanej w organizację, bo mimo bardzo dużej liczby uczestników wszystko poszło naprawdę sprawnie.

Tak jak pisałem na początku artykułu, aktualnie projekt Mevo został wstrzymany. Na szczęście, po dokonaniu kilku niedużych zmian, nasz program może działać z innym systemem roweru miejskiego (np. poznańskim).

Najpierw przestawimy naszą mapę na centrum Poznania:

POZNAN_CENTER_POSITION = [52.405783, 16.917085]
bikes_map = folium.Map(location=POZNAN_CENTER_POSITION, zoom_start=10)

Następnie zastąpimy wczytywanie danych z pliku – czyli poniższe 3 linijki:

bikes_file = bikes_proxy.current_locations_file
bikes_reader = csv.DictReader(bikes_file)

for station_row in bikes_reader:
    ...

Danymi pobranymi ze strony poznańskiego systemu:

for station_row in bikes_proxy.station_rows:
    ...

Teraz pozostaje nam już tylko zmodyfikować BikesProxy, aby dostosować kod do nowego formatu danych o rowerach. Informacje o stacjach zapiszemy w polu station_rows dokładnie w takiej samej postaci w jakiej wcześniej wczytywaliśmy je z pliku. Dzięki temu nie będziemy musieli modyfikować kodu przetwarzającego liczbę rowerów, dodającego znaczniki do mapy itp.

Poznański system nie udostępnia pliku csv z danymi, jednak na jego stronie znajduje się tabelka, z której te same dane możemy wyciągnąć. W tym celu zastosujemy bibliotekę BeautifulSoup, która bardzo dobrze radzi sobie z parsowaniem dokumentów HTML.

Najpierw pobieramy dane i inicjalizujemy parser:

self.station_rows = []

service_page_content = requests.get(SERVICE_URL).content.decode('utf-8')
bikes_data_parser = BeautifulSoup(service_page_content, features="lxml")

Aby wyszukać wszystkie wiersze z danymi o stacjach rowerowych posłużymy się faktem, iż każdy z nich zawarty jest w znaczniku <tr> i posiada specyficzną klasę css:

def is_station_row(css_class):
    return css_class and "place-number" in css_class

...

station_rows = bikes_data_parser.find_all('tr', class_=is_station_row)

Następnie przeszukamy wszystkie kolumny (komórki) z danego wiersza i wybierzemy z nich te, które zawierają interesujące nas informacje:

for row in station_rows:
    self.station_rows.append({})
    for column_number, column in enumerate(row.find_all('td')):
        if column_number == 1:
            self.station_rows[-1]["DOSTĘPNE ROWERY"] = self._value_or_default_when_none(column.string)
        elif column_number == 2:
            self.station_rows[-1]["WSPÓŁRZĘDNE"] = column.string
        elif column_number == 4:
            self.station_rows[-1]["NUMERY DOSTĘPNYCH ROWERÓW"] = column.string

Zdarza się, że kolumna z informacją o dostępnych rowerach zawiera niepoprawne dane. W celu obsłużenia tej sytuacji, korzystamy z prostej funkcji pomocniczej:

def _value_or_default_when_none(self, value):
    if value is not None:
        return value
    return '0'

A co z informacją o poziomie baterii? Rowery w Poznaniu to rozwiązanie tradycyjne zasilane jedynie za pomocą nóg. Na potrzeby naszego programu możemy założyć, że mniej więcej co piąty rower będzie miał niski poziom baterii, zaś co trzeci średni:

def battery_info_for_bike(self, bike_id):
    if int(bike_id) % 5 == 0:
        return 10
    if int(bike_id) % 3 == 0:
        return 35
    return 85

Tym sposobem otrzymaliśmy program generujący mapę z zaznaczonym rowerami z Poznania 🙂

Kod z przykładowymi rozwiązaniami kolejnych etapów warsztatu można pobrać z mojego repozytorium albo ze strony infoShare Academy, a prezentacja dostępna jest tutaj.

Do zobaczenia na blogu 🙂 a może i na kolejnych warsztatach? Hej!