hass_command_line_scripts/riverconditions.py

#!/usr/bin/env python3
#
# Description:
#
# Retrieves data from waterdata.usgs.gov
# and outputs it in JSON format for use in Home Assistant.
# Designed to be run as a command_line integration.
#
# Requirements:
# - Python 3.9+
# - requests
#
# Usage/installation:
#
# Takes a single argument: the 8-digit code identifying the station.
# This is part of the URL you'd use to view the information on the web,
# and is listed on the web page. For example,
# https://waterdata.usgs.gov/monitoring-location/14339000/
# is for measurement station 14339000.
# The web page title is:
# Rogue River at Dodge Bridge, Near Eagle Point, OR - 14339000
# This means that the command line would be:
# /path/to/binary/riverconditions.py 14339000
#
#
# Update the first line of this script to be the same python3 executable as
# your Home Assistant instance uses.
# 
# To use the integration, add the following to your configuration.yaml file
# (without the comments, obviously!)
# ------------------------
# command_line:
#   - sensor:
#       name: "River conditions"
#       unique_id: river_conditions
#       command: '/home/homeassistant/bin/riverconditions.py 14339000'
#       scan_interval: 1800
#       json_attributes:
#         - data
#       value_template: 'Rogue River conditions at Dodge Bridge'
# ------------------------
# You can use any value you want for value_template.
# Scan interval should be relatively long, since the values aren't updated
# frequently. Minimum interval should be 600 seconds (every 10 minutes).
#
# Next, add one or more sensors corresponding to the conditions you want to
# track in your system. For example,
# ------------------------
# template:
#   - sensors:
#     river_height:
#       friendly_name: "River height"
#       device_class: distance
#       value_template: "{{ state_attr('sensor.river_conditions', 'data')['height'] | round(1) }}"
#     river_flow:
#       friendly_name: "River flow"
#       device_class: volume_flow_rate
#       unit_of_measurement: "cfs"
#       value_template: "{{ state_attr('sensor.river_conditions', 'data')['flow'] | round(0) }}"
# ------------------------
# Note that height and flow are both contained within the 'data' attribute of the river_conditions
# sensor populated by the command.
#
#
#
#==========================================================================
# Copyright 2025 Ethan L. Miller (code@ethanmiller.us)
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

import sys
import json
import requests

mappings = {
    'temperature': 'watertemp',
    'streamflow':  'flow',
    'height':      'height',
}
if __name__ == '__main__':
    exit_code = 1
    for i in range(4):
        try:
            result = dict()
            station = sys.argv[1]
            url = f'https://waterservices.usgs.gov/nwis/iv/?format=json&sites={station}&siteStatus=all'
            req = requests.get (url, timeout=3)
            if req.ok:
                j = req.json()
                for v in j['value']['timeSeries']:
                    variable_name = v['variable']['variableName'].lower()
                    for k in mappings.keys():
                        if k in variable_name:
                            result[mappings[k]] = float(v['values'][0]['value'][0]['value'])
            print (json.dumps ({'data': result}))
            exit_code = 0
            break
        except:
            pass
    sys.exit (exit_code)