# coding: utf-8 # Author: Du Mingzhe (mingzhe@nus.edu.sg) # 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") mrt_cookie = os.getenv("MRT_COOKIE") 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': nextbus_token, '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': mrt_cookie } 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( """ """, 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']))