weather reporting bot for MQTT

weather_report.py

"""
Report weather data via MQTT.
"""

import sys
import os
import requests
import time
import paho.mqtt.client as mqtt


def assert_env(env_var: str) -> str:
    """Get a value from the environment or exit if it's not set."""
    if env_var not in os.environ:
        sys.stdout.write(f"Aborting: {env_var} not set")
        sys.exit(1)
    return str(os.environ.get(env_var))


# Fetch environment variables
WEATHER_API_KEY = assert_env("WEATHER_API_KEY")
WEATHER_LOCATION = assert_env("WEATHER_LOCATION")
MQTT_FQDN = assert_env("MQTT_BROKER_FQDN")
MQTT_TOKEN = assert_env("MQTT_BROKER_TOKEN")
DEVICE_ID = assert_env("DEVICE_ID")


def get_weather_data() -> dict[str, float]:
    """Return current local temperature and humidity data using weatherapi.com."""
    r = requests.get(
        f"https://api.weatherapi.com/v1/current.json?key={WEATHER_API_KEY}&q={WEATHER_LOCATION}&aqi=no"
    )
    data = r.json()
    stats = {
        "temp_f": float(data["current"]["temp_f"]),
        "humidity": float(data["current"]["humidity"]),
    }
    sys.stdout.write(f"Current weather: {stats}\n")
    sys.stdout.flush()
    return stats


def on_connect(client: mqtt.Client, userdata, flags, rc, properties) -> None:
    """Set up weather reporting on connection."""
    sys.stdout.write(f"Connected with result code: {rc}\n")
    publish_weather_data(client)


def publish_weather_data(client: mqtt.Client) -> None:
    """Publish weather data to MQTT broker, repeating every 5 minutes."""
    data = get_weather_data()
    client.publish(f"dt/weather/temp/{DEVICE_ID}", data["temp_f"])
    client.publish(f"dt/weather/humidity/{DEVICE_ID}", data["humidity"])
    time.sleep(60 * 5)
    publish_weather_data(client)


def main() -> None:
    """Connect to MQTT broker and publish weather data."""
    client = mqtt.Client(client_id="", protocol=mqtt.MQTTv5)
    client.username_pw_set(username=DEVICE_ID, password=MQTT_TOKEN.strip('"'))
    client.enable_logger()
    client.tls_set()
    client.on_connect = on_connect
    client.connect(MQTT_FQDN, port=8883)
    client.loop_forever()


if __name__ == "__main__":
    main()