mirror of https://github.com/home-assistant/core
Create a new NWS Alerts integration
This commit is contained in:
IceBotYT
parent
6f1675944e
commit
38309c5a87
|
|
@ -634,6 +634,8 @@ homeassistant/components/nut/* @bdraco @ollo69
|
|||
tests/components/nut/* @bdraco @ollo69
|
||||
homeassistant/components/nws/* @MatthewFlamm
|
||||
tests/components/nws/* @MatthewFlamm
|
||||
homeassistant/components/nws_alerts/* @IceBotYT
|
||||
tests/components/nws_alerts/* @IceBotYT
|
||||
homeassistant/components/nzbget/* @chriscla
|
||||
tests/components/nzbget/* @chriscla
|
||||
homeassistant/components/obihai/* @dshokouhi
|
||||
|
|
|
|||
|
|
@ -0,0 +1,21 @@
|
|||
"""The NWS Alerts integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
PLATFORMS: list[str] = ["sensor"]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up NWS Alerts from a config entry."""
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
return unload_ok
|
||||
|
|
@ -0,0 +1,144 @@
|
|||
"""Config flow for NWS Alerts integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
from .const import API_ENDPOINT, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required("api_key"): str,
|
||||
vol.Required("friendly_name", default="NWS Alerts"): str,
|
||||
vol.Required("update_interval", default=90): int,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def validate_input(hass: HomeAssistant, data: dict) -> dict:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Return the user input (modified if necessary) or raise a vol.Invalid
|
||||
exception if the data is incorrect.
|
||||
"""
|
||||
endpoint = API_ENDPOINT.format(
|
||||
lat=data["lat"], lon=data["lon"], api_key=data["api_key"]
|
||||
)
|
||||
try:
|
||||
response = await hass.async_add_executor_job(requests.get, endpoint)
|
||||
except requests.exceptions.HTTPError as error:
|
||||
_LOGGER.error("Error connecting to NWS Alerts API: %s", error)
|
||||
raise CannotConnect(
|
||||
"Cannot connect to alerts API. Please try again later"
|
||||
) from error
|
||||
except requests.exceptions.RequestException as error:
|
||||
_LOGGER.error("Error connecting to NWS Alerts API: %s", error)
|
||||
raise CannotConnect(
|
||||
"Cannot connect to alerts API. Please try again later"
|
||||
) from error
|
||||
|
||||
# check if it didn't return code 401
|
||||
if response.status_code == 401:
|
||||
_LOGGER.error("Invalid API key")
|
||||
raise Error401("Invalid API key")
|
||||
|
||||
if response.status_code == 404:
|
||||
_LOGGER.error("Invalid location")
|
||||
raise Error404("Invalid location")
|
||||
|
||||
if response.status_code == 429:
|
||||
_LOGGER.error("Too many requests")
|
||||
raise Error429("Too many requests")
|
||||
|
||||
if (
|
||||
response.status_code == 500
|
||||
or response.status_code == 502
|
||||
or response.status_code == 503
|
||||
or response.status_code == 504
|
||||
):
|
||||
_LOGGER.error("Service Unavailable")
|
||||
raise Error5XX("Service Unavailable")
|
||||
|
||||
if response.status_code == 200:
|
||||
return data
|
||||
else:
|
||||
raise Exception("Unknown error")
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for NWS Alerts."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
lat = self.hass.config.latitude
|
||||
lon = self.hass.config.longitude
|
||||
schema = STEP_USER_DATA_SCHEMA.extend(
|
||||
{
|
||||
vol.Required("lat", default=lat): float,
|
||||
vol.Required("lon", default=lon): float,
|
||||
}
|
||||
)
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=schema,
|
||||
)
|
||||
|
||||
errors = {}
|
||||
|
||||
try:
|
||||
await validate_input(self.hass, user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Error401:
|
||||
errors["api_key"] = "error_401"
|
||||
except Error404:
|
||||
errors["lat"] = "error_404"
|
||||
errors["lon"] = "error_404"
|
||||
except Error429:
|
||||
errors["base"] = "error_429"
|
||||
except Error5XX:
|
||||
errors["base"] = "error_5XX"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=user_input["friendly_name"], data=user_input
|
||||
)
|
||||
|
||||
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
|
||||
|
||||
|
||||
class CannotConnect(HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class Error401(HomeAssistantError):
|
||||
"""Error to indicate error 401."""
|
||||
|
||||
|
||||
class Error404(HomeAssistantError):
|
||||
"""Error to indicate error 404."""
|
||||
|
||||
|
||||
class Error429(HomeAssistantError):
|
||||
"""Error to indicate error 429."""
|
||||
|
||||
|
||||
class Error5XX(HomeAssistantError):
|
||||
"""Error to indicate error 5XX."""
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
"""Constants for the NWS Alerts integration."""
|
||||
|
||||
DOMAIN = "nws_alerts"
|
||||
API_ENDPOINT = "https://api.openweathermap.org/data/2.5/onecall?lat={lat}&lon={lon}&exclude=current,minutely,hourly,daily&appid={api_key}"
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
{
|
||||
"domain": "nws_alerts",
|
||||
"name": "NWS Alerts",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/nws_alerts",
|
||||
"requirements": [],
|
||||
"ssdp": [],
|
||||
"zeroconf": [],
|
||||
"homekit": {},
|
||||
"dependencies": [],
|
||||
"codeowners": [
|
||||
"@IceBotYT"
|
||||
],
|
||||
"iot_class": "cloud_polling"
|
||||
}
|
||||
|
|
@ -0,0 +1,179 @@
|
|||
"""Support for getting weather alerts from NOAA and other alert sources, thanks to the help of OpenWeatherMap."""
|
||||
|
||||
import datetime
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
import pytz
|
||||
import requests
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity, SensorStateClass
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
|
||||
from .const import API_ENDPOINT
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the sensor platform."""
|
||||
|
||||
async def async_update_data():
|
||||
"""Fetch data from OWM."""
|
||||
# Using a data update coordinator so we don't literally get rid of all our requests for the month >.>
|
||||
endpoint = API_ENDPOINT.format(
|
||||
lat=config_entry.data["lat"],
|
||||
lon=config_entry.data["lon"],
|
||||
api_key=config_entry.data["api_key"],
|
||||
)
|
||||
try:
|
||||
response = await hass.async_add_executor_job(requests.get, endpoint)
|
||||
except requests.exceptions.HTTPError as error:
|
||||
raise UpdateFailed(
|
||||
"Cannot connect to alerts API. Please try again later"
|
||||
) from error
|
||||
except requests.exceptions.RequestException as error:
|
||||
raise UpdateFailed(
|
||||
"Cannot connect to alerts API. Please try again later"
|
||||
) from error
|
||||
|
||||
# check if it didn't return code 401
|
||||
if response.status_code == 401:
|
||||
_LOGGER.error("Invalid API key")
|
||||
raise ConfigEntryAuthFailed("Invalid API key")
|
||||
|
||||
if response.status_code == 404:
|
||||
_LOGGER.error("Invalid location")
|
||||
raise ConfigEntryAuthFailed("Invalid location")
|
||||
|
||||
if response.status_code == 429:
|
||||
_LOGGER.error("Too many requests")
|
||||
raise UpdateFailed("Too many requests")
|
||||
|
||||
if (
|
||||
response.status_code == 500
|
||||
or response.status_code == 502
|
||||
or response.status_code == 503
|
||||
or response.status_code == 504
|
||||
):
|
||||
_LOGGER.error("Service Unavailable")
|
||||
raise UpdateFailed("Service Unavailable")
|
||||
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
else:
|
||||
raise UpdateFailed("Unknown error")
|
||||
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name="nws_alerts",
|
||||
update_method=async_update_data,
|
||||
update_interval=timedelta(seconds=config_entry.data.get("update_interval")),
|
||||
)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
sensor = WeatherAlertSensor(hass, config_entry, coordinator)
|
||||
async_add_entities([sensor])
|
||||
|
||||
|
||||
class WeatherAlertSensor(CoordinatorEntity, SensorEntity):
|
||||
"""Weather alert sensor."""
|
||||
|
||||
def __init__(self, hass, config, coordinator):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.hass = hass
|
||||
self._name = config.data.get("friendly_name", "NWS Alerts")
|
||||
self._alert_count = None
|
||||
self._unique_id = (
|
||||
self._name
|
||||
+ "-"
|
||||
+ str(config.data["lat"])
|
||||
+ "-"
|
||||
+ str(config.data["lon"])
|
||||
+ "-"
|
||||
+ config.entry_id
|
||||
)
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID to use for this sensor."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def state_class(self) -> SensorStateClass:
|
||||
"""Return the state class."""
|
||||
return SensorStateClass.MEASUREMENT
|
||||
|
||||
@property
|
||||
def native_value(self) -> int:
|
||||
"""Return the native value of the sensor."""
|
||||
if hasattr(self.coordinator.data, "alerts"):
|
||||
return len(self.coordinator.data["alerts"])
|
||||
else:
|
||||
return 0
|
||||
|
||||
# the property below is the star of the show
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict:
|
||||
"""Return the messages of all the alerts."""
|
||||
# Convert the start and end times from unix UTC to Home Assistant's time zone and format
|
||||
# the alert message
|
||||
attrs = {}
|
||||
if self.coordinator.data is not None:
|
||||
alerts = self.coordinator.data.get("alerts")
|
||||
if alerts is not None:
|
||||
timezone = pytz.timezone(self.hass.config.time_zone)
|
||||
utc = pytz.utc
|
||||
fmt = "%Y-%m-%d %H:%M"
|
||||
alerts = [
|
||||
{
|
||||
"start": datetime.datetime.fromtimestamp(alert["start"], tz=utc)
|
||||
.astimezone(timezone)
|
||||
.strftime(fmt),
|
||||
"end": datetime.datetime.fromtimestamp(alert["end"], tz=utc)
|
||||
.astimezone(timezone)
|
||||
.strftime(fmt),
|
||||
"sender_name": alert.get("sender_name"),
|
||||
"event": alert.get("event"),
|
||||
"description": alert.get("description"),
|
||||
}
|
||||
for alert in alerts
|
||||
]
|
||||
# we cannot have a list of dicts, we can only have strings and ints iirc
|
||||
# let's parse it so that both humans and machines can read it
|
||||
sender_name = " - ".join([alert.get("sender_name") for alert in alerts])
|
||||
event = " - ".join([alert.get("event") for alert in alerts])
|
||||
start = " - ".join([alert.get("start") for alert in alerts])
|
||||
end = " - ".join([alert.get("end") for alert in alerts])
|
||||
description = " - ".join([alert.get("description") for alert in alerts])
|
||||
attrs = {
|
||||
"sender_name": sender_name,
|
||||
"event": event,
|
||||
"start": start,
|
||||
"end": end,
|
||||
"description": description,
|
||||
}
|
||||
|
||||
return attrs
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def attribution(self) -> str:
|
||||
"""Return the attribution."""
|
||||
return "Data provided by the OpenWeatherMap Organization\n© 2012 — 2021 OpenWeather ® All rights reserved" # I don't want to get sued for this, but I can't find a way to get the attribution from the API
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
"""Return the icon."""
|
||||
return "mdi:alert"
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
{
|
||||
"config": {
|
||||
"error": {
|
||||
"error_401": "Invalid API key, or your API key hasn't been activated yet. Check your API key, then wait a few minutes and try again. This is a very common error. See https://openweathermap.org/faq#error401",
|
||||
"error_404": "Invalid location. Check your latitude and longitude values and try again. See https://openweathermap.org/faq#error404",
|
||||
"error_429": "Too many requests. Wait a few minutes and try again. See https://openweathermap.org/faq#error429",
|
||||
"error_5XX": "Something went wrong. Please try again later. See https://openweathermap.org/faq#error500",
|
||||
"unknown": "An unknown error occured. Please raise an issue on the GitHub repository."
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"api_key": "API Key",
|
||||
"friendly_name": "Friendly Name",
|
||||
"lat": "Latitude",
|
||||
"lon": "Longitude",
|
||||
"update_interval": "Update Interval (in seconds, the best for the free plan has been autofilled)"
|
||||
},
|
||||
"description": "Please create a free account at https://home.openweathermap.org/users/sign_up and confirm your account. You should receive an API key in your email. Enter it below.",
|
||||
"title": "NWS Alerts Setup"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
{
|
||||
"config": {
|
||||
"error": {
|
||||
"error_401": "Invalid API key, or your API key hasn't been activated yet. Check your API key, then wait a few minutes and try again. This is a very common error. See https://openweathermap.org/faq#error401",
|
||||
"error_404": "Invalid location. Check your latitude and longitude values and try again. See https://openweathermap.org/faq#error404",
|
||||
"error_429": "Too many requests. Wait a few minutes and try again. See https://openweathermap.org/faq#error429",
|
||||
"error_5XX": "Something went wrong. Please try again later. See https://openweathermap.org/faq#error500",
|
||||
"unknown": "An unknown error occured. Please raise an issue on the GitHub repository."
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"api_key": "API Key",
|
||||
"friendly_name": "Friendly Name",
|
||||
"lat": "Latitude",
|
||||
"lon": "Longitude",
|
||||
"update_interval": "Update Interval (in seconds, the best for the free plan has been autofilled)"
|
||||
},
|
||||
"description": "Please create a free account at https://home.openweathermap.org/users/sign_up and confirm your account. You should receive an API key in your email. Enter it below.",
|
||||
"title": "NWS Alerts Setup"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -213,6 +213,7 @@ FLOWS = [
|
|||
"nuki",
|
||||
"nut",
|
||||
"nws",
|
||||
"nws_alerts",
|
||||
"nzbget",
|
||||
"octoprint",
|
||||
"omnilogic",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,2 @@
|
|||
"""Tests for the NWS Alerts integration."""
|
||||
# No tests will be run, because I have a limited number of requests under my free plan
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
"""Test the NWS Alerts config flow."""
|
||||
# Cannot test a real API key, as I have a limited number of requests under my free plan
|
||||
# The validate_input function actually uses up a request, and if that API key gets out there, people will start spamming it
|
||||
# And then boom, I'll be suspended. :(
|
||||
|
||||
# Sorry :(
|
||||
Loading…
Reference in New Issue