St Andrews Gym Occupancy Bot

occupancy.py

import requests
import re
import logging
import time
from datetime import datetime

# Create a custom logger
logger = logging.getLogger(__name__)

# Create handlers
c_handler = logging.StreamHandler()
f_handler = logging.FileHandler('occupancy.log')
c_handler.setLevel(logging.WARNING)
f_handler.setLevel(logging.INFO)

# Create formatters and add it to handlers
c_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
f_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
c_handler.setFormatter(c_format)
f_handler.setFormatter(f_format)

# Add handlers to the logger
logger.addHandler(c_handler)
logger.addHandler(f_handler)

MAX_OCCUPANCY = 1000


class OccupancyChecker:

    def __init__(self):
        self.URL = "https://sport.wp.st-andrews.ac.uk/"
        self.THRESHOLD = 80
        self.COOLDOWN_SECONDS = 20
        self.WAIT_TIME = 5

        self.last_notified_occupancy = MAX_OCCUPANCY
        self.last_notification = datetime(2000, 1, 1)


    def get_occupancy(self):
        r = requests.get(self.URL, headers={'User-Agent': 'Occupancy Bot'})
        if r.status_code != 200:
            return MAX_OCCUPANCY

        match = re.search('Occupancy:\s(\d{0,2})%', r.text)
        if match is None:
            logger.error("Regex did not match anything.")
            return MAX_OCCUPANCY

        digit = match.group(1)
        if not digit.isdigit():
            logger.error(f"Regex matched a non-digit: {digit}")
            return MAX_OCCUPANCY

        return int(digit)


    def should_go(self, occ):
        return occ < self.THRESHOLD


    def has_cooled_down(self):
        delta = datetime.now() - self.last_notification
        return delta.total_seconds() > self.COOLDOWN_SECONDS


    def has_decreased(self, occ):
        return occ < self.last_notified_occupancy


    def should_notify(self, occ):
        if not self.should_go(occ):
            return False
        cooled_down = self.has_cooled_down()
        decreased = self.has_decreased(occ)

        # if cooldown period is over, we reset the decrease counter
        if cooled_down and not decreased:
            self.last_notified_occupancy = MAX_OCCUPANCY

        return cooled_down or decreased


    def notify(self, occ):
        ntfy_msg = f"Gym occupancy is at {occ}%"
        logger.warning(f"Notifying: {ntfy_msg}")
        requests.post("https://ntfy.sh/gym-occupancy", data=ntfy_msg)
        self.last_notified_occupancy = occ
        self.last_notification = datetime.now()


    def handle(self, occ):
        if self.should_notify(occ):
            self.notify(occ)


    def run(self):
        while True:
            occ = self.get_occupancy()
            logger.info(occ)
            self.handle(occ)
            time.sleep(self.WAIT_TIME)


if __name__ == '__main__':
    checker = OccupancyChecker()
    checker.run()