kmb.py

from urllib import request
from json import load
from datetime import datetime, timezone, timedelta
from os import system
from time import sleep


routes = [
    '32', '36', '40E', '40P', '40X', '46P', '46X', '47A',
    '47X', '48X', '73D', '73P', '73X', '936', 'N36', 'N48',
    '234C', '273C', '273P', '278A', '278P', '278X', '936A',
    'R936'
]

def update():
    url = 'https://data.etabus.gov.hk/v1/transport/kmb/stop-eta/'
    directions = [{
        'type': 'inbound',
        'desc': '往荃灣方向',
        'stops': [{
            "id": '97BB9CE7A058E809',
        }, {
            "id": '0A1DAEAFFDFF0B66',
        }],
        "data": []
    }, {
        'type': 'outbound',
        'desc': '往沙田方向',
        "stops": [{
            'id': 'DEDD6247441C8C05'
        }],
        "data": []
    }]

    # print('\n\033[0;30mUpdating...\033[0m')

    for direction in directions:
        stops = direction['stops']
        for stop in stops:
            id = stop['id']
            response = request.urlopen(f'{url}{id}')
            json = load(response)

            for route in routes:
                route = route.upper()
                for value in json['data']:
                    if value['route'] == route and value['service_type'] == 1:
                        if value['eta'] is None:
                            continue
                        eta = datetime.fromisoformat(value['eta'])
                        direction['data'].append({"route": route, "eta": eta})
            # result += '\n\n'

    tz = timezone.utc
    now = datetime.now(tz)
    updatedAt = now + timedelta(hours=8)
    result = "\033[0;33m"
    result += f'{datetime.strftime(updatedAt, "%Y年%-m月%-d日 (更新於%H:%M:%S)")}'
    result += "\033[0m"

    for direction in directions:
        desc = direction['desc']
        result += f'\n\n\033[0;36m\033[4m{desc}:\033[0m'
        data = sorted(direction['data'], key=lambda x: x.get('eta'))
        marked_routes = {}

        padding = 10
        for datum in data:
            route = datum['route']

            if route not in marked_routes:
                marked_routes[route] = '\n\033[0;32m' + \
                    f'{route}'.rjust(padding, " ") + '\033[0m'

            eta = datum['eta']
            diff = round((eta - now).seconds / 60)
            if (now > eta):
                marked_routes[route] += '\033[0;31m' + \
                    '已到達'.rjust(padding + 1, ' ') + '\033[0m'
            elif (diff < 1):
                marked_routes[route] += '\033[0;93m' + \
                    '快將到達'.rjust(padding, ' ') + '\033[0m'
            else:
                marked_routes[route] += f'{diff} 分鐘'.rjust(padding + 2, ' ')
        keys = marked_routes.keys()
        length = min(10, len(keys))
        for key in list(keys)[:length]:
            result += f"{marked_routes[key]}"
        if length < len(keys):
            result +='\n' + '...'.rjust(padding, ' ')
    system('clear')
    print(result)


while True:
    try:
        update()
    except:
        pass
    sleep(5)