Client code for https://www.metrolisboa.pt/ web api

metro.py

import requests
import re
import json
import logging

BEARER_RX = re.compile(r"Bearer \w{8}\-\w{4}\-\w{4}\-\w{4}\-\w{12}")
STATIONS_RX = re.compile(
    r"{ id: \"(\w\w)\", nome: \"([^\"]*)\", coord: \"(-?\d*\.\d*),(-?\d*\.\d*)\" }")
DESTINATIONS_RX = re.compile(
    r"case \"(\d*)\":.*\n.*return \[\"([^\"]*)\",\s*\"([^\"]*)\",\s*\"(\w\w)\"\];")

AUTH_URL = 'https://www.metrolisboa.pt/viajar/proximoscomboios/'
AUTH_HEADERS = {'User-Agent': 'WanderingTraveler'}

ENDPOINT_URL = "https://api.metrolisboa.pt:8243/estadoServicoML/1.0.0/tempoEspera/Estacao/%s"

class Metro:
    def __init__(self, auth = None):
        self.auth, self.stations, self.destinations = self.getAuth()
        
        if not auth is None :
            self.auth = auth

        logging.info("Using token: %s" % self.auth)

    def getAuth(self):
        req = requests.get(AUTH_URL, headers = AUTH_HEADERS)
        
        # get authentication token
        m = re.search(BEARER_RX, req.text)
        auth = m.group()
        
        # get stations id, name and coordinates
        stations = []
        for s in re.finditer(STATIONS_RX, req.text):
            sid, sname = s.group(1), s.group(2)
            lat, lon = float(s.group(3)), float(s.group(4))
            
            logging.debug("%s %s %f,%f" % (sid, sname, lat, lon))
            
            stations.append((sid, sname, (lat, lon)))

        # get destinations id, name, number, line
        destinations = []
        for d in re.finditer(DESTINATIONS_RX, req.text):
            dnum, dname = d.group(1), d.group(2)
            line, did = d.group(3), d.group(4)
            
            logging.debug("%s %s %s %s" % (did, dname, dnum, line))
            
            destinations.append((did, dname, dnum, line))

        return (auth, stations, destinations)

    def getWaitingTimes(self, stationid):
        u = ENDPOINT_URL % stationid
        h = {'Authorization': self.auth}

        req = requests.get(u, headers = h)
        # print(req.text)
        j = json.loads(req.text)
        
        if not j['codigo'] is '200' :
            raise Exception('erro %s %s' % (j['codigo'], j['resposta']))
        
        return j['resposta']

    def colectTrains(self):
        trains = set()
        
        for s, _, _ in self.stations:
            t = self.getWaitingTimes(s)
            for c in t:
                print(c)
                trains.add(c['comboio'])
                trains.add(c['comboio2'])
                trains.add(c['comboio3'])

        print(trains)
        print(len(trains))
        
    def colectPeers(self):
        peers = dict()
        
        for s, sn, _ in self.stations:
            t = self.getWaitingTimes(s)
            peers[(s, sn)] = []
            for c in t:
                peers[(s, sn)].append((c['cais'], c['destino'], c['sairServico']))
            
        print(peers)


if __name__== "__main__":
    #auth, stations, destinations = getAuth()
    # colectTrains(auth, stations)
    # colectPeers(auth, stations)

    #auth, stations, destinations = getAuth() 
    #print(destinations)
    logging.basicConfig(level=logging.DEBUG)
    m = Metro("Bearer ec1d2bf5-f586-3403-a977-1b539c87ae75")
    m.colectTrains()