Bart Vestaboard

bart.py

import click
import time
import requests
import xml.etree.ElementTree as ET
from vesta import vesta_layout, send_to_vesta


def get_departures(station_name):
    api_key = "MW9S-E7SL-26DU-VV8V"
    base_url = "https://api.bart.gov/api/etd.aspx"

    # Parameters for the GET request
    params = {"cmd": "etd", "orig": station_name, "key": api_key}

    # Make a GET request to the BART API for departures using the API key and encoded station name
    response = requests.get(base_url, params=params)

    # Parse the XML response into a dictionary
    root = ET.fromstring(response.content)
    departures = {}
    for etd in root.findall(".//etd"):
        destination = etd.find("destination").text
        departures[destination] = []
        for estimate in etd.findall("estimate"):
            minutes = estimate.find("minutes").text
            platform = estimate.find("platform").text
            direction = estimate.find("direction").text
            length = estimate.find("length").text
            color = estimate.find("color").text
            departures[destination].append(
                {
                    "minutes": minutes,
                    "platform": platform,
                    "direction": direction,
                    "length": length,
                    "color": color,
                }
            )

    return departures


def format(departures, station_name):
    # get sorted colors, times to next train
    trains = []
    directions = ["North", "South", "East", "West"]
    for direction in directions:
        for destination, estimates in departures.items():
            for estimate in estimates:
                if estimate["direction"] == direction:
                    trains.append(
                        (estimate["minutes"], estimate["color"], destination, direction)
                    )
    trains.sort()

    def soon(minutes):
        try:
            return int(minutes) < 15
        except ValueError:
            return False

    # format the response
    response = f"CBLUEBARTBLUE @ {station_name[:5]} \n"
    for direction in directions:
        my_trains = [
            (int(minutes), color)
            for minutes, color, _, d in trains
            if d == direction and soon(minutes)
        ]
        my_trains.sort()
        if len(my_trains) > 0:
            formatted = [f"<{color}>{minutes:02d}" for minutes, color, in my_trains]
            response += f"L{direction}:{' '.join(formatted)}\n"
    return response


@click.command()
@click.argument("station_name1")
@click.argument("station_name2")
@click.option(
    "--sleep-interval", default=180, help="Time to wait between updates in seconds."
)
def main(station_name1, station_name2, sleep_interval):
    last_layout = None
    while True:
        layout = vesta_layout(
            format(get_departures(station_name1), station_name1)
            + format(get_departures(station_name2), station_name2)
        )
        if layout != last_layout:
            send_to_vesta(layout)
            last_layout = layout
        time.sleep(sleep_interval)


if __name__ == "__main__":
    main()

vesta.py

import os
import requests
import xml.etree.ElementTree as ET
import json
import datetime
import time

vesta_codes = {
    " ": 0,
    "A": 1,
    "B": 2,
    "C": 3,
    "D": 4,
    "E": 5,
    "F": 6,
    "G": 7,
    "H": 8,
    "I": 9,
    "J": 10,
    "K": 11,
    "L": 12,
    "M": 13,
    "N": 14,
    "O": 15,
    "P": 16,
    "Q": 17,
    "R": 18,
    "S": 19,
    "T": 20,
    "U": 21,
    "V": 22,
    "W": 23,
    "X": 24,
    "Y": 25,
    "Z": 26,
    "1": 27,
    "2": 28,
    "3": 29,
    "4": 30,
    "5": 31,
    "6": 32,
    "7": 33,
    "8": 34,
    "9": 35,
    "0": 36,
    "!": 37,
    "@": 38,
    "#": 39,
    "$": 40,
    "(": 41,
    ")": 42,
    "-": 44,
    "+": 46,
    "&": 47,
    "=": 48,
    ";": 49,
    ":": 50,
    "'": 52,
    '"': 53,
    "%": 54,
    ",": 55,
    ".": 56,
    "/": 59,
    "?": 60,
    "°": 62,
    "<RED>": 63,
    "<ORANGE>": 64,
    "<YELLOW>": 65,
    "<GREEN>": 66,
    "<BLUE>": 67,
    "<VIOLET>": 68,
    "<WHITE>": 69,
    "<BLACK>": 70,
    "<FILLED>": 71,
}


def char_to_code(char):
    return vesta_codes.get(char.upper())


def string_to_codes(s):
    codes = []
    i = 0
    special_keys = list(vesta_codes.keys())[-9:]
    while i < len(s):
        matched = False  # Flag to indicate a keyword match

        # Try matching each keyword in the dictionary
        for key in special_keys:
            key_length = len(key)
            if s[i : i + key_length].upper() == key:  # Check for keyword match
                codes.append(vesta_codes[key])  # Add code for the keyword
                i += key_length  # Move past the keyword
                matched = True
                break  # Exit the loop after matching a keyword

        # If no keyword match, process the character
        if not matched:
            char = s[i].upper()
            if char in vesta_codes:  # Check if the character is in the dictionary
                codes.append(vesta_codes[char])  # Add the code for the character
            i += 1  # Move to the next character

    return codes


def vesta_layout(s):
    start = [
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
    ]
    line_width = len(start[0])
    for i, line in enumerate(s.split("\n")):
        if line == "":
            continue
        align = line[0]
        line = line[1:]
        encoding = string_to_codes(line)
        # trim to fit
        encoding = encoding[:line_width]
        # center the string
        if align == "C":
            start[i][
                int((line_width - len(encoding)) / 2) : int(
                    (line_width - len(encoding)) / 2
                )
                + len(encoding)
            ] = encoding
        elif align == "L":
            start[i][: len(encoding)] = encoding
    return start


def vesta_layout_json(s):
    # strip out the ticks if present
    s = s.split('```json')[-1].split('```')[0].strip()
    data = json.loads(s)
    start = [
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
        [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
    ]
    line_width = len(start[0])
    for i, line in enumerate(data["lines"]):
        if i == len(start):
            break
        align = line["alignment"]
        text = line["text"].strip()
        if text == "":
            continue
        encoding = string_to_codes(text)
        # trim to fit
        encoding = encoding[:line_width]
        # center the string
        if align == "center":
            start[i][
                int((line_width - len(encoding)) / 2) : int(
                    (line_width - len(encoding)) / 2
                )
                + len(encoding)
            ] = encoding
        elif align == "left":
            start[i][: len(encoding)] = encoding
    return start


def send_to_vesta(layout):
    url = "https://rw.vestaboard.com/"
    headers = {
        "X-Vestaboard-Read-Write-Key": os.environ["VESTA_API_KEY"],
        "Content-Type": "application/json",
    }
    response = requests.post(url, headers=headers, json=layout)


def sleep_until_custom_minute_marks(minute_marks):
    # Validate input
    for minute in minute_marks:
        if not (0 <= minute < 60):
            raise ValueError("All minutes in the list must be in the range [0, 59]")

    now = datetime.datetime.now()

    # Sort the minute marks to ensure they are in ascending order
    sorted_minute_marks = sorted(minute_marks)

    # Find the next target minute mark
    target_minute = None
    for minute_mark in sorted_minute_marks:
        if now.minute < minute_mark:
            target_minute = minute_mark
            break

    if target_minute is None:
        # If no future minute mark is found in the current hour, use the first minute mark and move to the next hour
        target_minute = sorted_minute_marks[0]
        target_hour = (
            now.hour + 1 if now.hour < 23 else 0
        )  # Handle wrapping around at midnight
        target = now.replace(
            hour=target_hour, minute=target_minute, second=0, microsecond=0
        )
    else:
        # Set the target to the found minute mark in the current hour
        target = now.replace(minute=target_minute, second=0, microsecond=0)

    sleep_duration = (target - now).total_seconds()

    time.sleep(max(0.1, sleep_duration))