Initial commit
This commit is contained in:
82
last_bus_times.py
Normal file
82
last_bus_times.py
Normal file
@@ -0,0 +1,82 @@
|
||||
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
|
||||
|
||||
|
||||
date_format = "%Y-%m-%d"
|
||||
yesterday = datetime.today() - timedelta(days=1)
|
||||
yesterday = yesterday.strftime(date_format)
|
||||
delay_csv = "delays.csv"
|
||||
|
||||
|
||||
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():
|
||||
# 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
|
||||
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("delays.png")
|
||||
plotly.offline.plot(fig, filename="delays.html")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
output_csv()
|
||||
make_plot()
|
||||
Reference in New Issue
Block a user