Check your daily Boston Water (BWSC) usage and alert about potential leaks

bwsc-usage.py

#!/usr/bin/env python3

"""

Check your daily Boston Water (BWSC) usage and alert about potential leaks

This script logs into the Boston Water and Sewer Commission customer portal
using the Chrome Selenium driver, downloads your daily usage for the past 30
days, and prints a warning if the most recent daily usage number is more than
three standard deviations higher than the mean of all the usage numbers. This
threshold was determined empirically, i.e., I looked at my own usage data for
the past month and there was one legitimate usage day that was more than two
standard deviations above mean, but none above three.

You can specify --headless for the obvious purpose.

You can specify the username and password on stdin or put them in one or two
files. For example:

bwsc-usage.py --username USERNAME --password PASSWORD
bwsc-usage.py --username-from ~/.bwsc_username --password-from ~/.bwsc_password
bwsc-usage.py --username-from - --password-from - < ~/.bwsc_credentials
[some command that prints username and password] | \
    bwsc-usage.py --username-from - --password-from -

You're probably going to get an alert if you fill a swimming pool. Just sayin'.

Copyright 2023 Jonathan Kamens <[email protected]>.

This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later
version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY
WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
PARTICULAR PURPOSE. See the GNU General Public License for more details.

For a copy of the GNU General Public License, see
<https://www.gnu.org/licenses/>.

"""

import argparse
from selenium import webdriver
from selenium.webdriver.common.by import By
import statistics
import time

portal_url = 'https://customerportal.bwsc.org/'


def find_one_of_several(driver, locators, timeout=5):
    end_at = time.time() + timeout
    first = True
    while first or time.time() < end_at:
        if first:
            first = False
        else:
            time.sleep(1)
        for locator in locators:
            try:
                elt = driver.find_element(locator[0], locator[1])
                return (locator, elt)
            except Exception:
                continue
    else:
        raise TimeoutError(f'Could not find any of {locators}')


def parse_args():
    parser = argparse.ArgumentParser(description='Check BWSC daily usage')
    group = parser.add_mutually_exclusive_group()
    group.add_argument('--username', action='store')
    group.add_argument('--username-from', metavar='PATH',
                       type=argparse.FileType('r'),
                       help='File to read username from, or "-" for stdin')
    group = parser.add_mutually_exclusive_group()
    group.add_argument('--password', action='store')
    group.add_argument('--password-from',  metavar='PATH',
                       type=argparse.FileType('r'),
                       help='File to read password from, or "-" for stdin')
    parser.add_argument('--headless', action='store_true', default=False)
    args = parser.parse_args()

    if args.username_from:
        args.username = args.username_from.readline().strip()
    if args.password_from:
        args.password = args.password_from.readline().strip()

    if not args.username:
        parser.error('Username is required')
    if not args.password:
        parser.error('Password is required')
    return args


def main():
    args = parse_args()
    options = webdriver.ChromeOptions()
    if args.headless:
        options.add_argument('headless')
    driver = webdriver.Chrome(options=options)
    driver.get(portal_url)
    locator, elt = find_one_of_several(driver, ((By.ID, 'logonIdentifier'),))
    elt.send_keys(args.username)
    driver.find_element(By.ID, 'password').send_keys(args.password)
    driver.find_element(By.ID, 'next').click()
    locator, elt = find_one_of_several(
        driver, ((By.XPATH, '//*/a[contains(text(),"View daily usage")]'),
                 (By.XPATH, '//*/a[contains(text(),"View Daily Usage")]')),
        timeout=10)
    elt.click()
    locator, elt = find_one_of_several(
        driver, ((By.XPATH, '//*/span[contains(text(), "Table")]'),),
        timeout=30)
    elt.click()
    usage_numbers = []
    for cell in driver.find_elements(By.XPATH, '//*/table/tbody/tr/td[5]'):
        usage_numbers.append(int(cell.get_attribute('innerText')))
    mean = statistics.mean(usage_numbers)
    stdev = statistics.pstdev(usage_numbers)
    if usage_numbers[0] > mean + 3 * stdev:
        print(f'WARNING! Most recent daily water usage {usage_numbers[0]} '
              f'seems very high')


if __name__ == '__main__':
    main()