citibike-jackpot-poller.py

"""
Simple script to loop and report if citibike jackpots are happening.
"""

import datetime
import json
import time

import pandas
import requests

import geopandas
import shapely

# PARAMETERS:

MIN_JACKPOT_VALUE = 5
NEIGHBORHOODS = [
    "North Side-South Side",
    "Williamsburg",
    "Greenpoint",
    "East Williamsburg",
]

# INPUTS:
SLACK_URL = (
    "<fill this in for yourself>"
)
NEIGHBORHOOD_URL = "https://data.cityofnewyork.us/api/geospatial/cpf4-rkhq?method=export&format=GeoJSON"
STATION_URL = "https://layer.bicyclesharing.net/map/v1/nyc/stations"

#
# Neighborhood manipulation:
#


def get_hood_geom(data, name):
    return data[data["ntaname"] == name]["geometry"].values[0]


def combine_neighborhoods(url=NEIGHBORHOOD_URL, names=NEIGHBORHOODS):
    hood_data = geopandas.read_file(NEIGHBORHOOD_URL)
    combined_area = shapely.ops.cascaded_union(
        [get_hood_geom(hood_data, n) for n in names]
    )
    return combined_area


def get_jackpots(area_filter, jackpot_value=MIN_JACKPOT_VALUE):
    station_data = geopandas.read_file(STATION_URL)
    return station_data[
        (station_data["geometry"].within(area_filter))
        & (station_data["bike_angels_points"] >= jackpot_value)
    ]


def format_slack_message(jackpots):
    return {
        "text": f'Citibike jackpots at: {datetime.datetime.now().isoformat(timespec="minutes")}',
        "attachments": [
            {
                "color": "#00ff00" if row.bike_angels_action == "take" else "#ff0000",
                "title": f'{row.name}: {"+" if row.bike_angels_action=="take" else "-"}{row.bike_angels_points}',
                "fallback": f'{row.name}: {"+" if row.bike_angels_action=="take" else "-"}{row.bike_angels_points}',
            }
            for row in jackpots.itertuples()
        ],
    }


def poll_delay_in_seconds():
    now = datetime.datetime.now()
    next_poll_time = now.replace(second=5, microsecond=0) + datetime.timedelta(
        minutes=15 - (now.minute % 15)
    )
    return (next_poll_time - now).seconds


if __name__ == "__main__":
    area_filter = combine_neighborhoods()
    while True:
        iso_now = datetime.datetime.now().isoformat(timespec="minutes")
        print(f"[{iso_now}] Polling... ")
        current_jackpots = get_jackpots(area_filter)
        if not current_jackpots.empty:
            print(
                f'[{iso_now}] Found {len(current_jackpots)} jackpot{"s" if len(current_jackpots) > 1 else ""}'
            )
            requests.post(SLACK_URL, json=format_slack_message(current_jackpots))
        time.sleep(poll_delay_in_seconds())