circuitbodge/circuit-laundry-notifier/web.py

import os
import uuid
from dataclasses import dataclass
from threading import Lock
from typing import *

import flask
import requests

from circuit_scraper import (
    CircuitScraper,
    Machine,
    ScraperError,
    MachineState,
    MachineType,
)


def new(debug=False) -> flask.Flask:
    app = flask.Flask(__name__)

    app.add_url_rule("/", view_func=_web_get_index, methods=["GET"])
    app.add_url_rule(
        "/api/v1/machines/<site_id>", view_func=_api_get_machines, methods=["GET"]
    )
    app.add_url_rule(
        "/api/v1/machines/<site_id>/watch",
        view_func=_api_register_watcher,
        methods=["POST"],
    )
    app.add_url_rule(
        "/api/v1/watcher/run", view_func=_api_run_watcher, methods=["POST"]
    )

    if debug:
        app.add_url_rule(
            "/api/v1/watcher", view_func=_api_list_watched, methods=["GET"]
        )

    return app


def _web_get_index() -> Union[any, Tuple[any, int]]:
    return flask.render_template("index.html")


def _api_get_machines(site_id: str) -> Union[any, Tuple[any, int]]:
    try:
        site = CircuitScraper.get_site_machine_states(site_id)
    except ScraperError:
        return (
            flask.jsonify(
                {
                    "ok": False,
                    "message": "This laundry room is unavailable via CircuitView.",
                }
            ),
            400,
        )

    machines = site.machines

    return flask.jsonify({
        "name": site.name,
        "machines": [
            machine.to_dict()
            for machine in sorted(
                machines,
                # It was annoying me that the machines weren't sorted by number. This sorts the machines by number,
                # taking into account the fact that some machine numbers include letters, which are quietly ignored.
                key=lambda x: int("".join(filter(lambda y: y.isdigit(), x.number))),
            )
        ],
    })


@dataclass
class WatchState:
    site_id: str
    ntfy_topic: str
    machine_number: str


jobs: List[WatchState] = []
job_lock = Lock()


def _api_register_watcher(site_id: str) -> Union[any, Tuple[any, int]]:
    ntfy_topic = flask.request.form.get("ntfy_topic", str(uuid.uuid4()))
    machine_number = flask.request.form.get("machine_number")

    if machine_number is None:
        return flask.jsonify({"ok": False, "message": "missing machine_number"}), 400

    site = CircuitScraper.get_site_machine_states(site_id)
    machine = site.get_machine(machine_number)
    if machine is None:
        return (
            flask.jsonify(
                {
                    "ok": False,
                    "message": "invalid site_id and machine_number combination",
                }
            ),
            400,
        )

    if machine.state != MachineState.InUse:
        return (
            flask.jsonify(
                {
                    "ok": False,
                    "message": "Machine not in use",
                }
            ),
            400,
        )

    job_lock.acquire()
    ws = WatchState(site_id, ntfy_topic, machine_number)
    jobs.append(ws)
    job_lock.release()

    time_remaining_text: str
    if machine.minutes_remaining < 0:
        time_remaining_text = (
            f"It was due to finish {machine.minutes_remaining} minutes ago."
        )
    else:
        time_remaining_text = (
            f"It's due to finish in {machine.minutes_remaining} minutes."
        )

    requests.post(
        f"https://ntfy.sh/{ws.ntfy_topic}",
        data=f"Registered to {machine.type.value.lower()} {ws.machine_number} at {site.name if site.name is not None else site_id} ({site_id}). "
             + time_remaining_text,
        headers={
            "Title": "Watching machine",
            "Tags": "white_check_mark",
        },
    )

    return flask.jsonify(ws), 200


def _api_list_watched() -> Union[any, Tuple[any, int]]:
    job_lock.acquire()
    res = flask.jsonify(jobs)
    job_lock.release()
    return res


WATCHER_BEARER_TOKEN = os.environ.get("CIRCUIT__WATCHER_TOKEN")


def _api_run_watcher() -> Union[any, Tuple[any, int]]:
    if WATCHER_BEARER_TOKEN is not None:
        auth_header = flask.request.headers.get("Authorization")
        if auth_header is None:
            return (
                flask.jsonify({"ok": False, "message": "missing Authorization header"}),
                401,
            )
        if auth_header != f"Bearer {WATCHER_BEARER_TOKEN}":
            return (
                flask.jsonify({"ok": False, "message": "bad Authorization token"}),
                401,
            )

    global jobs, job_lock

    job_lock.acquire()

    seen_sites: Dict[str, List[Machine]] = {}
    completed_items: List[int] = []

    for i, ws in enumerate(jobs):
        site_machines = seen_sites.get(ws.site_id)
        if site_machines is None:
            try:
                site_machines = CircuitScraper.get_site_machine_states(ws.site_id).machines
                seen_sites[ws.site_id] = site_machines
            except ScraperError:
                completed_items.append(i)
                continue

        target_machine: Optional[Machine] = None
        for machine in site_machines:
            if machine.number.lower() == ws.machine_number.lower():
                target_machine = machine
                break

        if target_machine is None:
            completed_items.append(i)
            continue

        if target_machine.state == MachineState.Completed or target_machine.state == MachineState.Available:
            requests.post(
                f"https://ntfy.sh/{ws.ntfy_topic}",
                data=f"{_first_letter_upper(target_machine.type.value.lower())} {target_machine.number} is finished!",
                headers={
                    "Title": (
                                 "Washing"
                                 if target_machine.type == MachineType.Washer
                                 else "Drying"
                             )
                             + " completed!",
                    "Priority": "urgent",
                    "Tags": "tada",
                },
            )
            completed_items.append(i)
            continue

    for x in list(reversed(sorted(completed_items))):
        jobs.pop(x)

    job_lock.release()

    return "", 204


def _first_letter_upper(x: str) -> str:
    if len(x) == 0:
        return ""
    return x[0].upper() + x[1:]