app.py

# coding: utf-8

# Author: Du Mingzhe ([email protected])
# Date: 2025/03/22

import os
import re
import json
import requests
import streamlit as st
from datetime import datetime
from bs4 import BeautifulSoup
from streamlit_autorefresh import st_autorefresh

nextbus_token = os.getenv("NEXTBUS_TOKEN")
datamall_token = os.getenv("DATAMALL_TOKEN")
mrt_token = os.getenv("MRT_TOKEN")

def wide_space_default():
    st.set_page_config(layout='wide')
wide_space_default()

count = st_autorefresh(interval=30000)

def get_all_stops():
    url = "https://nnextbus.nus.edu.sg/BusStops"

    payload = {}
    headers = {
        'Host': 'nnextbus.nus.edu.sg',
        'Content-Type': 'application/json',
        'Connection': 'keep-alive',
        'Accept': 'application/json',
        'User-Agent': 'nusnextbusv2/1 CFNetwork/978.0.7 Darwin/18.7.0',
        'Authorization': nextbus_token,
        'Accept-Language': 'en-us',
        'Accept-Encoding': 'br, gzip, deflate'
    }

    response = requests.request("GET", url, headers=headers, data=payload)
    return response.json()["BusStopsResult"]["busstops"]

def get_nus_bus_arrival(bus_stop_code):
    url = f"https://nnextbus.nus.edu.sg/ShuttleService?busstopname={bus_stop_code}"

    payload = {}
    headers = {
        'Host': 'nnextbus.nus.edu.sg',
        'Content-Type': 'application/json',
        'Connection': 'keep-alive',
        'Accept': 'application/json',
        'User-Agent': 'nusnextbusv2/1 CFNetwork/978.0.7 Darwin/18.7.0',
        'Authorization': 'Basic TlVTbmV4dGJ1czoxM2RMP3pZLDNmZVdSXiJU',
        'Accept-Language': 'en-us',
        'Accept-Encoding': 'br, gzip, deflate'
    }

    response = requests.request("GET", url, headers=headers, data=payload)

    return response.json()['ShuttleServiceResult']['shuttles']

def get_lta_bus_arrival(bus_stop_code):
    url = f"https://datamall2.mytransport.sg/ltaodataservice/v3/BusArrival?BusStopCode={bus_stop_code}"
    payload = {}
    headers = {
        'AccountKey': datamall_token
    }
    response = requests.request("GET", url, headers=headers, data=payload)
    return response.json()

def get_smrt_train_arrival(station_code):
    url = "https://trainarrivalweb.smrt.com.sg/"

    payload = f"ScriptManager1=UP1%7CddlStation&stnCode=&stnName=&ddlStation={station_code}&{mrt_token}"
    headers = {
        'Accept': '*/*',
        'Accept-Language': 'en-GB,en-US;q=0.9,en;q=0.8,zh-CN;q=0.7,zh-HK;q=0.6,zh-TW;q=0.5,zh;q=0.4',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
        'Origin': 'https://trainarrivalweb.smrt.com.sg',
        'Referer': 'https://trainarrivalweb.smrt.com.sg/',
        'Sec-Fetch-Dest': 'empty',
        'Sec-Fetch-Mode': 'cors',
        'Sec-Fetch-Site': 'same-origin',
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36',
        'X-MicrosoftAjax': 'Delta=true',
        'X-Requested-With': 'XMLHttpRequest',
        'sec-ch-ua': '"Chromium";v="134", "Not:A-Brand";v="24", "Google Chrome";v="134"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"macOS"',
        'Cookie': 'ASP.NET_SessionId=bwmeirfwzq30nj0ddk0qzyav; _ga=GA1.1.974289719.1742578510; _ga_78C66TP2RM=GS1.1.1742578509.1.1.1742578584.0.0.0'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    raw_html = response.text    
    soup = BeautifulSoup(raw_html, 'html.parser')

    tables = soup.find_all("table", id="gvTime")
    
    trains = list()
    
    for table in tables:
        time_row = table.find_all("tr")[1]
        time_cells = [td.get_text(strip=True) for td in time_row.find_all("td")]

        direction_row = table.find_all("tr")[2]
        direction_cells = [td.get_text(strip=True) for td in direction_row.find_all("td")]
        
        for eta_text, direction in zip(time_cells, direction_cells):
            eta = re.findall(r'\d+', eta_text)
            if eta:
                trains.append({
                    "direction": direction,
                    "eta": eta[0]
                })
    
    return trains


# stops = get_all_stops()
nus_bus_stops = [
    {
        "caption": "COM 3",
        "name": "COM3",
        "LongName": "COM 3",
        "ShortName": "COM 3",
        "latitude": 1.294431,
        "longitude": 103.775217
    },
    {
        "caption": "Opp TCOMS",
        "name": "TCOMS-OPP",
        "LongName": "Opp TCOMS",
        "ShortName": "Opp TCOMS",
        "latitude": 1.293789,
        "longitude": 103.776715
    },
    {
        "caption": "TCOMS",
        "name": "TCOMS",
        "LongName": "TCOMS",
        "ShortName": "TCOMS",
        "latitude": 1.293654,
        "longitude": 103.776898
    },
    {
        "caption": "Prince George's Park Foyer",
        "name": "PGPR",
        "LongName": "Prince George's Park Foyer",
        "ShortName": "PGP Foyer",
        "latitude": 1.290994,
        "longitude": 103.781153
    },
]

public_bus_stops = [
    {
        'name': "Opp HMK",
        'code': "16061"
    },
    {
        'name': "HMK",
        'code': "16069"
    },
]

public_mrt_station = [
    {
        'name': "KR MRT",
        'code': 'CKRG'
    }
]

# Hack the CSS to hide the delta icon
st.write(
    """
    <style>
    [data-testid="stMetricDelta"] svg {
        display: none;
    }
    </style>
    """,
    unsafe_allow_html=True,
)

# Layout
coloumns = [2,1,1,1,1,1,1,1,1,1,1]
number_of_coloumns = len(coloumns)

# NUS Stops
for stop_info in nus_bus_stops:
    shuttle_info = get_nus_bus_arrival(stop_info['name'])
    
    buses = list()
    
    for shuttle in shuttle_info:
        if "_etas" not in shuttle: continue
        for bus in shuttle["_etas"]:
            plate = bus["plate"]
            eta = bus["eta"]
            shuttle_name = shuttle['name']
            buses.append({
                "shuttle_name": shuttle_name,
                "plate": plate,
                "eta": eta
            })
    
    buses.sort(key=lambda x: x["eta"])
    
    with st.container(border=True):
        cols = st.columns(coloumns)
        cols[0].metric("NUS Stop", stop_info['name'])
        
        for i, bus in enumerate(buses[:number_of_coloumns-1]):
            cols[i+1].metric(bus['plate'], bus["shuttle_name"], str(bus["eta"]))
            
# Public Bus
for stop_info in public_bus_stops:
    bus_info = get_lta_bus_arrival(stop_info['code'])
    buses = list()
    
    for shuttle in bus_info["Services"]:
        try:
            service_no = shuttle["ServiceNo"]
            for bus_seq in ['NextBus', 'NextBus2', 'NextBus3']:
                bus_type = shuttle[bus_seq]['Type']
                bus_load = shuttle[bus_seq]['Load']
                arrival_time = datetime.fromisoformat(shuttle[bus_seq]['EstimatedArrival'])
                now = datetime.now(arrival_time.tzinfo)
                time_diff = arrival_time - now
                eta = int(time_diff.total_seconds() / 60)
                
                buses.append({
                    "service": service_no,
                    "eta": eta,
                    "type": f'{bus_type} - {bus_load}'
                })
        except Exception as e:
            pass
    
    buses.sort(key=lambda x: x["eta"])
    
    with st.container(border=True):
        cols = st.columns(coloumns)
        cols[0].metric("Public Stop", stop_info['name'])
        
        for i, bus in enumerate(buses[:number_of_coloumns-1]):
            cols[i+1].metric(bus["type"], bus['service'], bus["eta"])

# SMRT
for station in public_mrt_station:
    smrt_data = get_smrt_train_arrival(station['code'])
    trains = list()
    
    for train in smrt_data:
        trains.append({
            "direction": train["direction"],
            "eta": int(train["eta"])
        })
        
    trains.sort(key=lambda x: x["eta"])
    
with st.container(border=True):
    cols = st.columns(coloumns)
    cols[0].metric("MRT Station", station['name'])
    
    for i, train in enumerate(trains[:number_of_coloumns-1]):
        cols[i+1].metric(train['direction'], 'CC', str(train['eta']))