filter STPC's API to expose only user-friendly fields

main.py

@@ -1,19 +1,18 @@
 from stcp.api import get_stop_real_times, get_lines, get_line_directions, get_stop_data, get_line_stops
 
 
-def follow_line(line_code: str, direction: str) -> None:
+def follow_line(line_code: str, direction_code: str) -> None:
     """
     Print the current times for a given line
     :param line_code: the line to get the current times for
-    :param direction: line direction, usually '0' or '1', can be found using get_line_directions
+    :param direction_code: line direction, usually '0' or '1', can be found using get_line_directions
     :return: None
     """
-    stops = get_line_stops(line_code, direction)
+    stops = get_line_stops(line_code, direction_code)
     for stop in stops:
-        stop_code = stop['code']
-        stop_data = [bus for bus in get_stop_real_times(stop_code) if bus[0] == line_code]
+        stop_data = [bus for bus in get_stop_real_times(stop['stop_code']) if bus['line_code'] == line_code]
 
-        print(f'{stop["name"]: <25} {stop_data[0][1] if len(stop_data) > 0 else ""}')
+        print(f'{stop["name"]: <25} {stop_data[0]["time"] if len(stop_data) > 0 else ""}')
 
 
 # usage examples

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name = 'stcp_api',
-    version = '0.1.1',
+    version = '1.0.0',
     author = 'Guilherme Borges',
     author_email = 'g@guilhermeborges.net',
     description = 'Unofficial API to retrieve STCP information for public transit buses in Porto, Portugal',

stcp/_primitives.py

@@ -1,8 +1,6 @@
 import json
-import os
 import requests
 
-from bs4 import BeautifulSoup
 
 # TODO should not be needed to disable SSL, but after several tries seems to be a problem with STCP's certificate,
 # as all environments tested worked with several sites but this one
@@ -11,6 +9,8 @@
 
 
 def get_stop_hash(stop_code):
+    from bs4 import BeautifulSoup
+
     r = requests.get(f'https://www.stcp.pt/pt/viajar/horarios/?paragem={stop_code}&t=smsbus', verify=False)
     getter_script = BeautifulSoup(r.content.decode(), 'html.parser').find('table').find('script').string
     code = getter_script.split(',')[2].split('\'')[1]
@@ -23,13 +23,13 @@ def get_lines():
     return json.loads(r.content.decode())['records']
 
 
-def get_line_directions(line):
-    r = requests.get(f'https://www.stcp.pt/pt/itinerarium/callservice.php?action=linedirslist&lcode={line}', verify=False)
+def get_line_directions(internal_line_code):
+    r = requests.get(f'https://www.stcp.pt/pt/itinerarium/callservice.php?action=linedirslist&lcode={internal_line_code}', verify=False)
     return json.loads(r.content.decode())['records']
 
 
-def get_line_stops(line, direction):
-    r = requests.get(f'https://www.stcp.pt/pt/itinerarium/callservice.php?action=linestops&lcode={line}&ldir={direction}', verify=False)
+def get_line_stops(internal_line_code, direction_code):
+    r = requests.get(f'https://www.stcp.pt/pt/itinerarium/callservice.php?action=linestops&lcode={internal_line_code}&ldir={direction_code}', verify=False)
     return json.loads(r.content.decode())['records']
 
 
@@ -38,30 +38,22 @@ def get_stop_data(stop_code):
     return json.loads(r.content.decode())[0]  # there should be always one dictionary only
 
 
-def get_stop_real_times(stop_code, use_hash_cache):
-    # use cache to avoid making a request to STCP
-    if use_hash_cache:
-        # create the cache if it doesn't exist yet...
-        if not os.path.isfile('hash.tmp'):
-            from stcp._hash_cache import write_hash_file
-            write_hash_file()
-
-        from stcp._hash_cache import read_hash_file
-        hash_code = read_hash_file()[stop_code]
-    else:
-        hash_code = get_stop_hash(stop_code)
+def get_stop_real_times(stop_code, hash_code):
+    from bs4 import BeautifulSoup
 
     r = requests.get(f'https://www.stcp.pt/pt/itinerarium/soapclient.php?codigo={stop_code}&linha=0&hash123={hash_code}', verify=False)
     parsed_page = BeautifulSoup(r.content.decode(), 'html.parser')
+
     if parsed_page.find(class_='msgBox warning'):
         # TODO check for an occasion where the cached hash might not work; in that case invalidate it
         return []
 
     buses = []
     for bus in parsed_page.find(id='smsBusResults').find_all('tr')[1:]:
         elements = bus.find_all('td')
-        line = elements[0].find('a').text.strip()
-        time = elements[1].text.strip()
-        buses.append((line, time))
+        buses.append({
+            'line_code': elements[0].find('a').text.strip(),
+            'time': elements[1].text.strip()
+        })
 
     return buses

stcp/_stop_cache.py

@@ -4,28 +4,10 @@
 
 
 def write_stops_file(filename='stops.json'):
-    from stcp._primitives import get_stop_data
-
-    all_stops = get_all_stops()
+    from stcp.api import get_stop_data
 
     with open(filename, 'w') as file:
-        stop_data = []
-
-        for stop_code in all_stops:
-            # get stop data to store
-            data = get_stop_data(stop_code)
-            coordinates = json.loads(data['geomdesc'])['coordinates']
-            lines = [{'line_code': line['code'], 'dir': line['dir'], 'description': line['description']} for line in data['lines']]
-
-            stop_data.append({
-                'stop_code': stop_code,
-                'name': data['name'],
-                'address': data['address'],
-                'lon': coordinates[0],
-                'lat': coordinates[1],
-                'lines': lines
-            })
-
+        stop_data = [get_stop_data(stop_code) for stop_code in get_all_stops()]
         json.dump(stop_data, file)
 
 

stcp/_util.py

@@ -3,12 +3,28 @@ def get_all_stops():
     Returns a list of all STCP stops across all lines.
     :return: a list of all STCP stops
     """
-    from stcp._primitives import get_lines, get_line_stops
+    from stcp.api import get_lines, get_line_stops, get_line_directions
 
     all_stops = set()
 
     for line in get_lines():
-        stops = get_line_stops(line['code'], 0) + get_line_stops(line['code'], 1)
-        all_stops.update([stop['code'] for stop in stops])  # TODO there is a stop in Maia called . with code .
+        for direction in get_line_directions(line['line_code']):
+            stops = get_line_stops(line['line_code'], direction['direction_code'])
+            all_stops.update([stop['stop_code'] for stop in stops])  # TODO there is a stop in Maia called . with code .
 
     return all_stops
+
+
+def get_internal_line_code(line_code: str):
+    """
+    Converts the human-readable line code into the internal one (e.g. ZC -> 107)
+    :param line_code: the human-readable line code
+    :return: the internal line code
+    """
+    from stcp._primitives import get_lines
+
+    for line in get_lines():
+        if line['pubcode'] == line_code:
+            return line['code']
+
+    raise Exception('Invalid line code')

stcp/api.py

@@ -1,4 +1,4 @@
-from typing import List, Dict, Tuple
+from typing import List, Dict
 
 
 def get_lines() -> List[Dict]:
@@ -8,30 +8,53 @@ def get_lines() -> List[Dict]:
     :return: list of all STCP lines
     """
     from stcp._primitives import get_lines
-    return get_lines()
 
+    return [{
+        'line_code': line['pubcode'],
+        'accessibility': line['accessibility'],
+        'description': line['description']
+    } for line in get_lines()]
 
-def get_line_directions(line: str) -> List[Dict]:
+
+def get_line_directions(line_code: str) -> List[Dict]:
     """
     Get a list of directions (usually 2) of a line.
 
-    :param line: code of the line
+    :param line_code: code of the line
     :return: list of _line_'s directions
     """
     from stcp._primitives import get_line_directions
-    return get_line_directions(line)
+    from stcp._util import get_internal_line_code
+
+    internal_line_code = get_internal_line_code(line_code)
+
+    return [{
+        'direction_code': direction['dir'],
+        'description': direction['descr_dir'],
+        'readable': direction['descr']
+    } for direction in get_line_directions(internal_line_code)]
 
 
-def get_line_stops(line: str, direction: str) -> List[Dict]:
+def get_line_stops(line_code: str, direction_code: str) -> List[Dict]:
     """
     Get a list of all stops of a line. Direction is important as not all lines are "symmetrical".
 
-    :param line: code of the line
-    :param direction: direction of the line
+    :param line_code: code of the line (use the line_code, instead of the display_code)
+    :param direction_code: direction of the line
     :return: list of stops of that line, in that direction
     """
     from stcp._primitives import get_line_stops
-    return get_line_stops(line, direction)
+    from stcp._util import get_internal_line_code
+
+    internal_line_code = get_internal_line_code(line_code)
+
+    return [{
+        'stop_code': stop['code'],
+        'name': stop['name'],
+        'zone': stop['zone'],
+        'address': stop['address'],
+        'seq': stop['sequence']
+    } for stop in get_line_stops(internal_line_code, direction_code)]
 
 
 def get_stop_data(stop_code) -> Dict:
@@ -42,16 +65,52 @@ def get_stop_data(stop_code) -> Dict:
     :return: a dictionary containg data such as stop name, address, and a list of lines that pass through the stop
     """
     from stcp._primitives import get_stop_data
-    return get_stop_data(stop_code)
+    import json
+
+    stop = get_stop_data(stop_code)
 
+    coordinates = json.loads(stop['geomdesc'])['coordinates']
 
-def get_stop_real_times(stop_code: str, use_hash_cache=True) -> List[Tuple[str, str]]:
+    lines = [{
+        'line_code': line['pubcode'],
+        'direction_code': line['dir'],
+        'accessibility': line['accessibility'],
+        'description': line['description']
+    } for line in stop['lines']]
+
+    return {
+        'stop_code': stop_code,
+        'name': stop['name'],
+        'zone': stop['zone'],
+        'address': stop['address'],
+        'mode': stop['mode'],
+        'lon': coordinates[0],
+        'lat': coordinates[1],
+        'lines': lines
+    }
+
+
+def get_stop_real_times(stop_code: str, use_hash_cache=True) -> List[Dict]:
     """
     Get a real-time list of buses passing through a stop soon (up to one hour from the current time).
 
     :param stop_code: code of the stop
     :param use_hash_cache: use a local cache to avoid doing two requests per invocation
     :return: list of buses passing through the stop soon
     """
-    from stcp._primitives import get_stop_real_times
-    return get_stop_real_times(stop_code, use_hash_cache)
+    from stcp._primitives import get_stop_real_times, get_stop_hash
+    import os
+
+    # use cache to avoid making a request to STCP
+    if use_hash_cache:
+        # create the cache if it doesn't exist yet...
+        if not os.path.isfile('hash.tmp'):
+            from stcp._hash_cache import write_hash_file
+            write_hash_file()
+
+        from stcp._hash_cache import read_hash_file
+        hash_code = read_hash_file()[stop_code]
+    else:
+        hash_code = get_stop_hash(stop_code)
+
+    return get_stop_real_times(stop_code, hash_code)