vincentbernat.i3wm-configuration/bin/polybar-weather

#!/usr/bin/env python3

"""Get current weather condition and forecast for Polybar."""

import requests
import logging
import argparse
import sys
import os
import subprocess
import time
from systemd import journal

logger = logging.getLogger("polybar-weather")


def get_location():
    """Return current location as latitude/longitude tuple."""
    logger.debug("query ip-api.com for location")
    r = requests.get("http://ip-api.com/json")
    r.raise_for_status()
    data = r.json()
    logger.debug("current location data: %s", data)
    logger.info(f'current location: {data["city"]}, {data["country"]}')
    return (data["lat"], data["lon"])


def get_weather(apikey, latitude, longitude, endpoint, count=1):
    """Return forecasts data from openweathermap."""
    logger.debug("query openweathermap for %s, %s", latitude, longitude)
    r = requests.get(
        f"https://api.openweathermap.org/data/2.5/{endpoint}",
        params={
            "appid": apikey,
            "lat": latitude,
            "lon": longitude,
            "units": "metric",
            "cnt": count,
        },
    )
    r.raise_for_status()
    data = r.json()
    logger.debug("%s data: %s", endpoint, data)
    return data


def format_weather(data):
    """Translate OWM icon to WeatherIcons."""
    # https://erikflowers.github.io/weather-icons/
    icon = data["weather"][0]["icon"]
    temperature = data["main"]["temp"]
    if icon == "01d" and temperature > 32:
        icon = "\uf072"
    else:
        icon = {
            "01d": "\uf00d",  # Clear sky - day
            "01n": "\uf02e",  # Clear sky - night
            "02d": "\uf002",  # Few clouds (11-25%) - day
            "02n": "\uf083",  # Few clouds (11-25%) - night
            "03d": "\uf041",  # Scattered clouds (25-50%) - day/night
            "03n": "\uf086",  # Scattered clouds (25-50%) - day/night
            "04d": "\uf013",  # Broken / Overcast clouds (51-84% / 85-100%) - day/night
            "04n": "\uf013",  # Broken / Overcast clouds (51-84% / 85-100%) - day/night
            "09d": "\uf009",  # Shower rain - day
            "09n": "\uf037",  # Shower rain - night
            "10d": "\uf008",  # Moderate / heavy rain - day
            "10n": "\uf036",  # Moderate / heavy rain - night
            "11d": "\uf005",  # Thunderstorm - day
            "11n": "\uf025",  # Thunderstorm - night
            "13d": "\uf00a",  # Snow - day
            "13n": "\uf02a",  # Snow - night
            "50d": "\uf003",  # Fog - day
            "50n": "\uf04a",  # Fog - night
        }.get(icon, "\uf075")
    return "".join(
        [
            "%{Tx}",
            icon,
            "%{T-} ",
            str(int(round(temperature))),
            "°C",
        ]
    )


def update_status(status, output):
    """Update current status."""
    # Write it to file
    with open(output, "w") as f:
        f.write(status)

    # Send it to polybar
    subprocess.run(["polybar-msg", "action", f"#weather.send.{status}"], check=True)


if __name__ == "__main__":
    # Parse
    description = sys.modules[__name__].__doc__
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument(
        "--debug",
        "-d",
        action="store_true",
        default=False,
        help="enable debugging",
    )
    parser.add_argument(
        "--owm-api-key",
        default=os.environ.get("OWM_API_KEY"),
        help="OpenWeatherMap API key",
    )
    parser.add_argument(
        "--forecasts", default=2, type=int, help="Number of forecasts to fetch"
    )
    parser.add_argument(
        "--font-index", default=4, type=int, help="Weather Icons 1-index"
    )
    parser.add_argument(
        "--output",
        default=f"{os.environ['XDG_RUNTIME_DIR']}/i3/weather.txt",
        help="Output destination",
    )
    parser.add_argument(
        "--online-timeout",
        default=30,
        type=int,
        help="Wait up to TIMEOUT minutes to be online",
    )
    options = parser.parse_args()

    # Logging
    root = logging.getLogger("")
    root.setLevel(logging.WARNING)
    logger.setLevel(options.debug and logging.DEBUG or logging.INFO)
    if sys.stderr.isatty():
        ch = logging.StreamHandler()
        ch.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
        root.addHandler(ch)
    else:
        root.addHandler(journal.JournalHandler(SYSLOG_IDENTIFIER=logger.name))

    try:
        # Get location
        try:
            location = get_location()
        except requests.exceptions.ConnectionError:
            # Wait to be online
            time.sleep(2)
            logger.info("not online, waiting")
            update_status("", options.output)
            process = subprocess.run(
                ["nm-online", "-s", "-q", "-t", str(options.online_timeout * 60)]
            )
            if process.returncode != 0:
                logger.warning("not online, exiting")
                sys.exit(1)
            location = get_location()

        # Grab current weather and forecast
        current_weather = get_weather(options.owm_api_key, *location, "weather")
        forecast_weather = get_weather(
            options.owm_api_key, *location, "forecast", count=options.forecasts
        )
        description = current_weather["weather"][0]["description"]
        city = current_weather["name"]
        logger.info(f"current weather at {city}: {description}")

        # Format output
        conditions = [format_weather(data)
                      for data in [current_weather] + forecast_weather["list"]]
        while len(conditions) >= 2:
            if conditions[-1] == conditions[-2]:
                conditions.pop()
            else:
                break
        output = " ".join(conditions).replace("%{Tx}", "%%{T%d}" % options.font_index)
        logger.debug("output: %s", output)

        update_status(output, options.output)

    except Exception as e:
        logger.exception("%s", e)
        sys.exit(1)
    sys.exit(0)