mirror of
https://github.com/RichieCahill/dotfiles.git
synced 2026-04-17 04:58:19 -04:00
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""Van weather service - fetches weather with masked GPS for privacy."""
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Annotated, Any
|
|
|
|
import requests
|
|
import typer
|
|
from apscheduler.schedulers.blocking import BlockingScheduler
|
|
from tenacity import before_sleep_log, retry, stop_after_attempt, wait_fixed
|
|
|
|
from python.common import configure_logger
|
|
from python.van_weather.models import Config, DailyForecast, HourlyForecast, Weather
|
|
|
|
# Map Pirate Weather icons to Home Assistant conditions
|
|
CONDITION_MAP = {
|
|
"clear-day": "sunny",
|
|
"clear-night": "clear-night",
|
|
"rain": "rainy",
|
|
"snow": "snowy",
|
|
"sleet": "snowy-rainy",
|
|
"wind": "windy",
|
|
"fog": "fog",
|
|
"cloudy": "cloudy",
|
|
"partly-cloudy-day": "partlycloudy",
|
|
"partly-cloudy-night": "partlycloudy",
|
|
}
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@retry(
|
|
stop=stop_after_attempt(3),
|
|
wait=wait_fixed(5),
|
|
before_sleep=before_sleep_log(logger, logging.WARNING),
|
|
reraise=True,
|
|
)
|
|
def get_ha_state(url: str, token: str, entity_id: str) -> float:
|
|
"""Get numeric state from Home Asasistant entity."""
|
|
response = requests.get(
|
|
f"{url}/api/states/{entity_id}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
state = response.json()["state"]
|
|
if state in ("unavailable", "unknown"):
|
|
error = f"{entity_id} is {state}"
|
|
raise ValueError(error)
|
|
return float(state)
|
|
|
|
|
|
def parse_daily_forecast(data: dict[str, dict[str, Any]]) -> list[DailyForecast]:
|
|
"""Parse daily forecast from Pirate Weather API."""
|
|
daily = data.get("daily", {}).get("data", [])
|
|
daily_forecasts = []
|
|
for day in daily[:8]: # Up to 8 days
|
|
time_stamp = day.get("time")
|
|
if time_stamp:
|
|
date_time = datetime.fromtimestamp(time_stamp, tz=UTC).isoformat()
|
|
daily_forecasts.append(
|
|
DailyForecast(
|
|
date_time=date_time,
|
|
condition=CONDITION_MAP.get(day.get("icon", ""), "cloudy"),
|
|
temperature=day.get("temperatureHigh"),
|
|
templow=day.get("temperatureLow"),
|
|
precipitation_probability=day.get("precipProbability"),
|
|
moon_phase=day.get("moonPhase"),
|
|
wind_gust=day.get("windGust"),
|
|
cloud_cover=day.get("cloudCover"),
|
|
)
|
|
)
|
|
|
|
return daily_forecasts
|
|
|
|
|
|
def parse_hourly_forecast(data: dict[str, dict[str, Any]]) -> list[HourlyForecast]:
|
|
"""Parse hourly forecast from Pirate Weather API."""
|
|
hourly = data.get("hourly", {}).get("data", [])
|
|
hourly_forecasts = []
|
|
for hour in hourly[:48]: # Up to 48 hours
|
|
time_stamp = hour.get("time")
|
|
if time_stamp:
|
|
date_time = datetime.fromtimestamp(time_stamp, tz=UTC).isoformat()
|
|
hourly_forecasts.append(
|
|
HourlyForecast(
|
|
date_time=date_time,
|
|
condition=CONDITION_MAP.get(hour.get("icon", ""), "cloudy"),
|
|
temperature=hour.get("temperature"),
|
|
precipitation_probability=hour.get("precipProbability"),
|
|
)
|
|
)
|
|
return hourly_forecasts
|
|
|
|
|
|
@retry(
|
|
stop=stop_after_attempt(3),
|
|
wait=wait_fixed(5),
|
|
before_sleep=before_sleep_log(logger, logging.WARNING),
|
|
reraise=True,
|
|
)
|
|
def fetch_weather(api_key: str, lat: float, lon: float) -> Weather:
|
|
"""Fetch weather from Pirate Weather API."""
|
|
url = f"https://api.pirateweather.net/forecast/{api_key}/{lat},{lon}"
|
|
response = requests.get(url, params={"units": "us"}, timeout=30)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
daily_forecasts = parse_daily_forecast(data)
|
|
hourly_forecasts = parse_hourly_forecast(data)
|
|
|
|
current = data.get("currently", {})
|
|
icon = current.get("icon", "")
|
|
return Weather(
|
|
temperature=current.get("temperature"),
|
|
feels_like=current.get("apparentTemperature"),
|
|
humidity=current.get("humidity"),
|
|
wind_speed=current.get("windSpeed"),
|
|
wind_bearing=current.get("windBearing"),
|
|
condition=CONDITION_MAP.get(icon, "cloudy"),
|
|
summary=current.get("summary"),
|
|
pressure=current.get("pressure"),
|
|
visibility=current.get("visibility"),
|
|
uv_index=current.get("uvIndex"),
|
|
ozone=current.get("ozone"),
|
|
nearest_storm_distance=current.get("nearestStormDistance"),
|
|
nearest_storm_bearing=current.get("nearestStormBearing"),
|
|
precip_probability=current.get("precipProbability"),
|
|
cloud_cover=current.get("cloudCover"),
|
|
daily_forecasts=daily_forecasts,
|
|
hourly_forecasts=hourly_forecasts,
|
|
)
|
|
|
|
|
|
@retry(
|
|
stop=stop_after_attempt(3),
|
|
wait=wait_fixed(5),
|
|
before_sleep=before_sleep_log(logger, logging.WARNING),
|
|
reraise=True,
|
|
)
|
|
def post_to_ha(url: str, token: str, weather: Weather) -> None:
|
|
"""Post weather data to Home Assistant as sensor entities."""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Post current weather as individual sensors
|
|
sensors = {
|
|
"sensor.van_weather_condition": {
|
|
"state": weather.condition or "unknown",
|
|
"attributes": {"friendly_name": "Van Weather Condition"},
|
|
},
|
|
"sensor.van_weather_temperature": {
|
|
"state": weather.temperature,
|
|
"attributes": {"unit_of_measurement": "°F", "device_class": "temperature"},
|
|
},
|
|
"sensor.van_weather_apparent_temperature": {
|
|
"state": weather.feels_like,
|
|
"attributes": {"unit_of_measurement": "°F", "device_class": "temperature"},
|
|
},
|
|
"sensor.van_weather_humidity": {
|
|
"state": int((weather.humidity or 0) * 100),
|
|
"attributes": {"unit_of_measurement": "%", "device_class": "humidity"},
|
|
},
|
|
"sensor.van_weather_pressure": {
|
|
"state": weather.pressure,
|
|
"attributes": {"unit_of_measurement": "mbar", "device_class": "pressure"},
|
|
},
|
|
"sensor.van_weather_wind_speed": {
|
|
"state": weather.wind_speed,
|
|
"attributes": {"unit_of_measurement": "mph", "device_class": "wind_speed"},
|
|
},
|
|
"sensor.van_weather_wind_bearing": {
|
|
"state": weather.wind_bearing,
|
|
"attributes": {"unit_of_measurement": "°"},
|
|
},
|
|
"sensor.van_weather_visibility": {
|
|
"state": weather.visibility,
|
|
"attributes": {"unit_of_measurement": "mi"},
|
|
},
|
|
"sensor.van_weather_uv_index": {
|
|
"state": weather.uv_index,
|
|
"attributes": {"friendly_name": "Van Weather UV Index", "icon": "mdi:sun-wireless"},
|
|
},
|
|
"sensor.van_weather_ozone": {
|
|
"state": weather.ozone,
|
|
"attributes": {"unit_of_measurement": "DU", "icon": "mdi:earth"},
|
|
},
|
|
"sensor.van_weather_nearest_storm_distance": {
|
|
"state": weather.nearest_storm_distance,
|
|
"attributes": {"unit_of_measurement": "mi", "icon": "mdi:weather-lightning"},
|
|
},
|
|
"sensor.van_weather_nearest_storm_bearing": {
|
|
"state": weather.nearest_storm_bearing,
|
|
"attributes": {"unit_of_measurement": "°", "icon": "mdi:weather-lightning"},
|
|
},
|
|
"sensor.van_weather_precip_probability": {
|
|
"state": int((weather.precip_probability or 0) * 100),
|
|
"attributes": {"unit_of_measurement": "%", "icon": "mdi:weather-rainy"},
|
|
},
|
|
"sensor.van_weather_cloud_cover": {
|
|
"state": int((weather.cloud_cover or 0) * 100),
|
|
"attributes": {"unit_of_measurement": "%", "icon": "mdi:weather-cloudy"},
|
|
},
|
|
}
|
|
|
|
for entity_id, data in sensors.items():
|
|
if data["state"] is not None:
|
|
response = requests.post(f"{url}/api/states/{entity_id}", headers=headers, json=data, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
# Post daily forecast as JSON attribute sensor
|
|
daily_forecast = [
|
|
{
|
|
"datetime": daily_forecast.date_time.isoformat(),
|
|
"condition": daily_forecast.condition,
|
|
"temperature": daily_forecast.temperature,
|
|
"templow": daily_forecast.templow,
|
|
"precipitation_probability": int((daily_forecast.precipitation_probability or 0) * 100),
|
|
}
|
|
for daily_forecast in weather.daily_forecasts
|
|
]
|
|
|
|
response = requests.post(
|
|
f"{url}/api/states/sensor.van_weather_forecast_daily",
|
|
headers=headers,
|
|
json={"state": len(daily_forecast), "attributes": {"forecast": daily_forecast}},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# Post hourly forecast as JSON attribute sensor
|
|
hourly_forecast = [
|
|
{
|
|
"datetime": hourly_forecast.date_time.isoformat(),
|
|
"condition": hourly_forecast.condition,
|
|
"temperature": hourly_forecast.temperature,
|
|
"precipitation_probability": int((hourly_forecast.precipitation_probability or 0) * 100),
|
|
}
|
|
for hourly_forecast in weather.hourly_forecasts
|
|
]
|
|
|
|
response = requests.post(
|
|
f"{url}/api/states/sensor.van_weather_forecast_hourly",
|
|
headers=headers,
|
|
json={"state": len(hourly_forecast), "attributes": {"forecast": hourly_forecast}},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
|
|
def update_weather(config: Config) -> None:
|
|
"""Fetch weather using last-known location, post to HA."""
|
|
lat = get_ha_state(config.ha_url, config.ha_token, config.lat_entity)
|
|
lon = get_ha_state(config.ha_url, config.ha_token, config.lon_entity)
|
|
|
|
masked_lat = round(lat, config.mask_decimals)
|
|
masked_lon = round(lon, config.mask_decimals)
|
|
|
|
logger.info(f"Masked location: {masked_lat}, {masked_lon}")
|
|
|
|
weather = fetch_weather(config.pirate_weather_api_key, lat, lon)
|
|
logger.info(f"Weather: {weather.temperature}°F, {weather.condition}")
|
|
|
|
post_to_ha(config.ha_url, config.ha_token, weather)
|
|
logger.info("Posted weather to HA")
|
|
|
|
|
|
def main(
|
|
ha_url: Annotated[str, typer.Option(envvar="HA_URL")],
|
|
ha_token: Annotated[str, typer.Option(envvar="HA_TOKEN")],
|
|
api_key: Annotated[str, typer.Option(envvar="PIRATE_WEATHER_API_KEY")],
|
|
interval: Annotated[int, typer.Option(help="Poll interval in seconds")] = 900,
|
|
log_level: Annotated[str, typer.Option()] = "INFO",
|
|
) -> None:
|
|
"""Fetch weather for van using masked GPS location."""
|
|
configure_logger(log_level)
|
|
|
|
config = Config(ha_url=ha_url, ha_token=ha_token, pirate_weather_api_key=api_key)
|
|
|
|
logger.info(f"Starting van weather service, polling every {interval}s")
|
|
|
|
scheduler = BlockingScheduler()
|
|
scheduler.add_job(
|
|
update_weather,
|
|
"interval",
|
|
seconds=interval,
|
|
args=[config],
|
|
next_run_time=datetime.now(UTC),
|
|
)
|
|
scheduler.start()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
typer.run(main)
|