89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import requests
|
|
import re
|
|
import datetime
|
|
import pytz
|
|
from datetime import datetime, timedelta
|
|
import csv
|
|
import pandas as pd
|
|
import numpy as np
|
|
from plotly_calplot import calplot
|
|
import plotly
|
|
|
|
delay_csv = "delays.csv"
|
|
delay_html = "output/delays.html"
|
|
delay_png = "output/delays.png"
|
|
|
|
|
|
def get_delay(date_string: str) -> int:
|
|
"""
|
|
Calculates the delay in minutes between the last bus departure time and the target time.
|
|
|
|
Args:
|
|
date_string (str): The date in the format 'YYYY-MM-DD'.
|
|
|
|
Returns:
|
|
int: The delay in minutes.
|
|
"""
|
|
|
|
|
|
vehicles = requests.get(
|
|
f"https://bustimes.org/services/632-wigan-bus-station-bus-station/vehicles?date={date_string}").text
|
|
|
|
last_joruney = re.findall(
|
|
r'\#journeys\/(\d+)">([\d\:]+)<\/a>\s+<\/td>\s+<td>Chorley Town Centre',
|
|
vehicles,
|
|
)[-1]
|
|
|
|
if last_joruney[1] != "23:20" and last_joruney[1] != "22:45":
|
|
#print(f"No last bus at 23:20 or 22:45, last bus was at {last_joruney[1]}")
|
|
return 60
|
|
else:
|
|
journey_info = requests.get(
|
|
f"https://bustimes.org/services/75994/journeys/{last_joruney[0]}.json").json()
|
|
station_departure = journey_info["stops"][0]["actual_departure_time"]
|
|
|
|
departure_time = datetime.fromisoformat(station_departure.replace("Z", "+00:00")).astimezone(pytz.timezone("UTC"))
|
|
target_time = datetime.strptime(date_string + " " + last_joruney[1], "%Y-%m-%d %H:%M").astimezone(pytz.timezone("Europe/London"))
|
|
|
|
return int((departure_time - target_time).total_seconds() / 60)
|
|
|
|
|
|
def output_csv():
|
|
yesterday = datetime.today() - timedelta(days=1)
|
|
yesterday = yesterday.strftime("%Y-%m-%d")
|
|
|
|
# Check if the file exists
|
|
try:
|
|
with open(delay_csv, "r") as file:
|
|
reader = csv.reader(file)
|
|
for row in reader:
|
|
if row[0] == yesterday:
|
|
break
|
|
# If the entry for yesterday already exists, do nothing
|
|
|
|
else:
|
|
# Add a new entry for yesterday
|
|
with open(delay_csv, "a", newline="") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow([yesterday, get_delay(yesterday)])
|
|
|
|
except FileNotFoundError:
|
|
# Create a new file and add the entry for yesterday
|
|
with open(delay_csv, "w", newline="") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(["Date", "Minutes Late"])
|
|
writer.writerow([yesterday, get_delay(yesterday)])
|
|
|
|
|
|
def make_plot():
|
|
df = pd.read_csv(delay_csv)
|
|
df["Date"] = pd.to_datetime(df["Date"])
|
|
|
|
fig = calplot(df, x="Date", y="Minutes Late", dark_theme=True, colorscale=[(0, "green"), (0.1, "yellow"), (1, "red")])
|
|
fig.write_image(delay_png)
|
|
plotly.offline.plot(fig, filename=delay_html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
output_csv()
|
|
make_plot() |