Add nws weather.

This commit is contained in:
ktdad
2019-05-03 06:49:01 -04:00
parent e13e4376f8
commit 86fc312352
5 changed files with 626 additions and 0 deletions

View File

@@ -0,0 +1 @@
"""NWS Integration."""

View File

@@ -0,0 +1,8 @@
{
"domain": "nws",
"name": "National Weather Service",
"documentation": "https://www.home-assistant.io/components/nws",
"dependencies": [],
"codeowners": ["@MatthewFlamm"],
"requirements": ["pynws==0.6"]
}

View File

@@ -0,0 +1,306 @@
"""Support for NWS weather service."""
from collections import OrderedDict
from datetime import timedelta
import logging
from statistics import mean
import async_timeout
import voluptuous as vol
from homeassistant.components.weather import (
WeatherEntity, PLATFORM_SCHEMA, ATTR_FORECAST_CONDITION,
ATTR_FORECAST_TEMP, ATTR_FORECAST_TIME,
ATTR_FORECAST_WIND_SPEED, ATTR_FORECAST_WIND_BEARING)
from homeassistant.const import (
CONF_API_KEY, CONF_NAME, CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE,
LENGTH_KILOMETERS, LENGTH_METERS, LENGTH_MILES, PRESSURE_HPA, PRESSURE_PA,
PRESSURE_INHG, TEMP_CELSIUS, TEMP_FAHRENHEIT)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers import config_validation as cv
from homeassistant.util import Throttle
from homeassistant.util.distance import convert as convert_distance
from homeassistant.util.pressure import convert as convert_pressure
from homeassistant.util.temperature import convert as convert_temperature
REQUIREMENTS = ['pynws==0.6']
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = 'Data from National Weather Service/NOAA'
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
CONF_STATION = 'station'
ATTR_FORECAST_DETAIL_DESCRIPTION = 'detailed_description'
ATTR_FORECAST_PRECIP_PROB = 'precipitation_probability'
ATTR_FORECAST_DAYTIME = 'daytime'
# Ordered so that a single condition can be chosen from multiple weather codes.
# Known NWS conditions that do not map: cold
CONDITION_CLASSES = OrderedDict([
('snowy', ['snow', 'snow_sleet', 'sleet', 'blizzard']),
('snowy-rainy', ['rain_snow', 'rain_sleet', 'fzra',
'rain_fzra', 'snow_fzra']),
('hail', []),
('lightning-rainy', ['tsra', 'tsra_sct', 'tsra_hi']),
('lightning', []),
('pouring', []),
('rainy', ['rain', 'rain_showers', 'rain_showers_hi']),
('windy-variant', ['wind_bkn', 'wind_ovc']),
('windy', ['wind_skc', 'wind_few', 'wind_sct']),
('fog', ['fog']),
('clear', ['skc']), # sunny and clear-night
('cloudy', ['bkn', 'ovc']),
('partlycloudy', ['few', 'sct'])
])
FORECAST_CLASSES = {
ATTR_FORECAST_DETAIL_DESCRIPTION: 'detailedForecast',
ATTR_FORECAST_TEMP: 'temperature',
ATTR_FORECAST_TIME: 'startTime',
}
FORECAST_MODE = ['daynight', 'hourly']
WIND_DIRECTIONS = ['N', 'NNE', 'NE', 'ENE',
'E', 'ESE', 'SE', 'SSE',
'S', 'SSW', 'SW', 'WSW',
'W', 'WNW', 'NW', 'NNW']
WIND = {name: idx * 360 / 16 for idx, name in enumerate(WIND_DIRECTIONS)}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_MODE, default='daynight'): vol.In(FORECAST_MODE),
vol.Optional(CONF_STATION, default=''): cv.string,
vol.Required(CONF_API_KEY): cv.string
})
def parse_icon(icon):
"""
Parse icon url to NWS weather codes.
Example:
https://api.weather.gov/icons/land/day/skc/tsra,40/ovc?size=medium
Example return:
('day', (('skc', 0), ('tsra', 40),))
"""
icon_list = icon.split('/')
time = icon_list[5]
weather = [i.split('?')[0] for i in icon_list[6:]]
code = [w.split(',')[0] for w in weather]
chance = [int(w.split(',')[1]) if len(w.split(',')) == 2 else 0
for w in weather]
return time, tuple(zip(code, chance))
def convert_condition(time, weather):
"""
Convert NWS codes to HA condition.
Choose first condition in CONDITION_CLASSES that exists in weather code.
If no match is found, return fitst condition from NWS
"""
conditions = [w[0] for w in weather]
prec_prob = [w[1] for w in weather]
# Choose condition with highest priority.
cond = next((key for key, value in CONDITION_CLASSES.items()
if any(condition in value for condition in conditions)),
conditions[0])
if cond == 'clear':
if time == 'day':
return 'sunny', max(prec_prob)
if time == 'night':
return 'clear-night', max(prec_prob)
return cond, max(prec_prob)
async def async_setup_platform(hass, config, async_add_entities,
discovery_info=None):
"""Set up the nws platform."""
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
station = config.get(CONF_STATION)
api_key = config.get(CONF_API_KEY)
if None in (latitude, longitude):
_LOGGER.error("Latitude/longitude not set in Home Assistant config")
return
from pynws import Nws
websession = async_get_clientsession(hass)
# ID request as being from HA, pynws prepends the api_key in addition
api_key_ha = [api_key + 'homeassistant']
nws = Nws(websession, latlon=(float(latitude), float(longitude)),
userid=api_key_ha)
_LOGGER.debug("Setting up station: %s", station)
if station == '':
with async_timeout.timeout(10, loop=hass.loop):
stations = await nws.stations()
_LOGGER.info("Station list: %s", stations)
nws.station = stations[0]
_LOGGER.debug("Initialized for coordinates %s, %s -> station %s",
latitude, longitude, stations[0])
else:
nws.station = station
_LOGGER.debug("Initialized station %s", station[0])
async_add_entities([NWSWeather(nws, hass.config.units, config)], True)
class NWSWeather(WeatherEntity):
"""Representation of a weather condition."""
def __init__(self, nws, units, config):
"""Initialise the platform with a data instance and station name."""
self._nws = nws
self._station_name = config.get(CONF_NAME, self._nws.station)
self._observation = None
self._forecast = None
self._description = None
self._is_metric = units.is_metric
self._mode = config[CONF_MODE]
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Update Condition."""
with async_timeout.timeout(10, loop=self.hass.loop):
_LOGGER.debug("Updating station observations %s",
self._nws.station)
self._observation = await self._nws.observations()
_LOGGER.debug("Updating forecast")
if self._mode == 'daynight':
self._forecast = await self._nws.forecast()
elif self._mode == 'hourly':
self._forecast = await self._nws.forecast_hourly()
else:
_LOGGER.error("Invalid Forecast Mode")
_LOGGER.debug("Observations: %s", self._observation)
_LOGGER.debug("Forecasts: %s", self._forecast)
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def name(self):
"""Return the name of the station."""
return self._station_name
@property
def temperature(self):
"""Return the current temperature."""
temp_c = self._observation[0]['temperature']['value']
if temp_c is not None:
return convert_temperature(temp_c, TEMP_CELSIUS, TEMP_FAHRENHEIT)
return None
@property
def pressure(self):
"""Return the current pressure."""
pressure_pa = self._observation[0]['seaLevelPressure']['value']
# convert Pa to in Hg
if pressure_pa is None:
return None
if self._is_metric:
pressure = convert_pressure(pressure_pa, PRESSURE_PA, PRESSURE_HPA)
pressure = round(pressure)
else:
pressure = convert_pressure(pressure_pa,
PRESSURE_PA, PRESSURE_INHG)
pressure = round(pressure, 2)
return pressure
@property
def humidity(self):
"""Return the name of the sensor."""
return self._observation[0]['relativeHumidity']['value']
@property
def wind_speed(self):
"""Return the current windspeed."""
wind_m_s = self._observation[0]['windSpeed']['value']
if wind_m_s is None:
return None
wind_m_hr = wind_m_s * 3600
if self._is_metric:
wind = convert_distance(wind_m_hr,
LENGTH_METERS, LENGTH_KILOMETERS)
else:
wind = convert_distance(wind_m_hr, LENGTH_METERS, LENGTH_MILES)
return round(wind)
@property
def wind_bearing(self):
"""Return the current wind bearing (degrees)."""
return self._observation[0]['windDirection']['value']
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return TEMP_FAHRENHEIT
@property
def condition(self):
"""Return current condition."""
time, weather = parse_icon(self._observation[0]['icon'])
cond, _ = convert_condition(time, weather)
return cond
@property
def visibility(self):
"""Return visibility."""
vis_m = self._observation[0]['visibility']['value']
if vis_m is None:
return None
if self._is_metric:
vis = convert_distance(vis_m, LENGTH_METERS, LENGTH_KILOMETERS)
else:
vis = convert_distance(vis_m, LENGTH_METERS, LENGTH_MILES)
return round(vis, 0)
@property
def forecast(self):
"""Return forecast."""
forecast = []
for forecast_entry in self._forecast:
data = {attr: forecast_entry[name]
for attr, name in FORECAST_CLASSES.items()}
if self._mode == 'daynight':
data[ATTR_FORECAST_DAYTIME] = forecast_entry['isDaytime']
time, weather = parse_icon(forecast_entry['icon'])
cond, precip = convert_condition(time, weather)
data[ATTR_FORECAST_CONDITION] = cond
if precip > 0:
data[ATTR_FORECAST_PRECIP_PROB] = precip
else:
data[ATTR_FORECAST_PRECIP_PROB] = None
data[ATTR_FORECAST_WIND_BEARING] = \
WIND[forecast_entry['windDirection']]
# wind speed reported as '7 mph' or '7 to 10 mph'
# if range, take average
wind_speed = forecast_entry['windSpeed'].split(' ')[0::2]
wind_speed_avg = mean(int(w) for w in wind_speed)
if self._is_metric:
data[ATTR_FORECAST_WIND_SPEED] = round(
convert_distance(wind_speed_avg,
LENGTH_MILES, LENGTH_KILOMETERS))
else:
data[ATTR_FORECAST_WIND_SPEED] = round(wind_speed_avg)
forecast.append(data)
return forecast

View File

@@ -1288,6 +1288,9 @@ pynuki==1.3.3
# homeassistant.components.nut # homeassistant.components.nut
pynut2==2.1.2 pynut2==2.1.2
# homeassistant.components.nws
pynws==0.6
# homeassistant.components.nx584 # homeassistant.components.nx584
pynx584==0.4 pynx584==0.4

View File

@@ -0,0 +1,308 @@
"""Tests for the NWS weather component."""
import unittest
from unittest.mock import patch
from homeassistant.components import weather
from homeassistant.components.nws.weather import ATTR_FORECAST_PRECIP_PROB
from homeassistant.components.weather import (
ATTR_WEATHER_HUMIDITY, ATTR_WEATHER_PRESSURE, ATTR_WEATHER_TEMPERATURE,
ATTR_WEATHER_VISIBILITY, ATTR_WEATHER_WIND_BEARING,
ATTR_WEATHER_WIND_SPEED)
from homeassistant.components.weather import (
ATTR_FORECAST, ATTR_FORECAST_CONDITION, ATTR_FORECAST_TEMP,
ATTR_FORECAST_TIME, ATTR_FORECAST_WIND_BEARING, ATTR_FORECAST_WIND_SPEED)
from homeassistant.const import (
LENGTH_KILOMETERS, LENGTH_METERS, LENGTH_MILES, PRECISION_WHOLE,
PRESSURE_INHG, PRESSURE_PA, PRESSURE_HPA, TEMP_CELSIUS, TEMP_FAHRENHEIT)
from homeassistant.helpers.temperature import display_temp
from homeassistant.util.pressure import convert as convert_pressure
from homeassistant.util.distance import convert as convert_distance
from homeassistant.util.unit_system import IMPERIAL_SYSTEM, METRIC_SYSTEM
from homeassistant.util.temperature import convert as convert_temperature
from homeassistant.setup import setup_component
from tests.common import get_test_home_assistant, MockDependency
OBS = [{
'temperature': {'value': 7, 'qualityControl': 'qc:V'},
'relativeHumidity': {'value': 10, 'qualityControl': 'qc:V'},
'windChill': {'value': 10, 'qualityControl': 'qc:V'},
'heatIndex': {'value': 10, 'qualityControl': 'qc:V'},
'windDirection': {'value': 180, 'qualityControl': 'qc:V'},
'visibility': {'value': 10000, 'qualityControl': 'qc:V'},
'windSpeed': {'value': 10, 'qualityControl': 'qc:V'},
'seaLevelPressure': {'value': 30000, 'qualityControl': 'qc:V'},
'windGust': {'value': 10, 'qualityControl': 'qc:V'},
'dewpoint': {'value': 10, 'qualityControl': 'qc:V'},
'icon': 'https://api.weather.gov/icons/land/day/skc?size=medium',
'textDescription': 'Sunny'
}]
FORE = [{
'endTime': '2018-12-21T18:00:00-05:00',
'windSpeed': '8 to 10 mph',
'windDirection': 'S',
'shortForecast': 'Chance Showers And Thunderstorms',
'isDaytime': True,
'startTime': '2018-12-21T15:00:00-05:00',
'temperatureTrend': None,
'temperature': 41,
'temperatureUnit': 'F',
'detailedForecast': 'A detailed description',
'name': 'This Afternoon',
'number': 1,
'icon': 'https://api.weather.gov/icons/land/day/skc/tsra,40?size=medium'
}]
HOURLY_FORE = [{
'endTime': '2018-12-22T05:00:00-05:00',
'windSpeed': '4 mph',
'windDirection': 'N',
'shortForecast': 'Chance Showers And Thunderstorms',
'startTime': '2018-12-22T04:00:00-05:00',
'temperatureTrend': None,
'temperature': 32,
'temperatureUnit': 'F',
'detailedForecast': '',
'number': 2,
'icon': 'https://api.weather.gov/icons/land/night/skc?size=medium'
}]
STN = 'STNA'
class MockNws():
"""Mock Station from pynws."""
def __init__(self, websession, latlon, userid):
"""Init mock nws."""
pass
async def observations(self):
"""Mock Observation."""
return OBS
async def forecast(self):
"""Mock Forecast."""
return FORE
async def forecast_hourly(self):
"""Mock Hourly Forecast."""
return HOURLY_FORE
async def stations(self):
"""Mock stations."""
return [STN]
class TestNWS(unittest.TestCase):
"""Test the NWS weather component."""
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.hass.config.units = IMPERIAL_SYSTEM
self.lat = self.hass.config.latitude = 40.00
self.lon = self.hass.config.longitude = -8.00
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_w_name(self, mock_pynws):
"""Test for successfully setting up the NWS platform with name."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'name': 'HomeWeather',
'platform': 'nws',
'api_key': 'test_email',
}
})
state = self.hass.states.get('weather.homeweather')
assert state.state == 'sunny'
data = state.attributes
temp_f = convert_temperature(7, TEMP_CELSIUS, TEMP_FAHRENHEIT)
assert data.get(ATTR_WEATHER_TEMPERATURE) == \
display_temp(self.hass, temp_f, TEMP_FAHRENHEIT, PRECISION_WHOLE)
assert data.get(ATTR_WEATHER_HUMIDITY) == 10
assert data.get(ATTR_WEATHER_PRESSURE) == round(
convert_pressure(30000, PRESSURE_PA, PRESSURE_INHG), 2)
assert data.get(ATTR_WEATHER_WIND_SPEED) == round(10 * 2.237)
assert data.get(ATTR_WEATHER_WIND_BEARING) == 180
assert data.get(ATTR_WEATHER_VISIBILITY) == round(
convert_distance(10000, LENGTH_METERS, LENGTH_MILES))
assert state.attributes.get('friendly_name') == 'HomeWeather'
forecast = data.get(ATTR_FORECAST)
assert forecast[0].get(ATTR_FORECAST_CONDITION) == 'lightning-rainy'
assert forecast[0].get(ATTR_FORECAST_PRECIP_PROB) == 40
assert forecast[0].get(ATTR_FORECAST_TEMP) == 41
assert forecast[0].get(ATTR_FORECAST_TIME) == \
'2018-12-21T15:00:00-05:00'
assert forecast[0].get(ATTR_FORECAST_WIND_BEARING) == 180
assert forecast[0].get(ATTR_FORECAST_WIND_SPEED) == 9
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_w_station(self, mock_pynws):
"""Test for successfully setting up the NWS platform with station."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
'station': 'STNB',
'api_key': 'test_email',
}
})
assert self.hass.states.get('weather.stnb')
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_w_no_name(self, mock_pynws):
"""Test for successfully setting up the NWS platform w no name."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
'api_key': 'test_email',
}
})
assert self.hass.states.get('weather.' + STN)
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test__hourly(self, mock_pynws):
"""Test for successfully setting up hourly forecast."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'name': 'HourlyWeather',
'platform': 'nws',
'api_key': 'test_email',
'mode': 'hourly',
}
})
state = self.hass.states.get('weather.hourlyweather')
data = state.attributes
forecast = data.get(ATTR_FORECAST)
assert forecast[0].get(ATTR_FORECAST_CONDITION) == 'clear-night'
assert forecast[0].get(ATTR_FORECAST_PRECIP_PROB) is None
assert forecast[0].get(ATTR_FORECAST_TEMP) == 32
assert forecast[0].get(ATTR_FORECAST_TIME) == \
'2018-12-22T04:00:00-05:00'
assert forecast[0].get(ATTR_FORECAST_WIND_BEARING) == 0
assert forecast[0].get(ATTR_FORECAST_WIND_SPEED) == 4
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_daynight(self, mock_pynws):
"""Test for successfully setting up daynight forecast."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
'api_key': 'test_email',
'mode': 'daynight',
}
})
assert self.hass.states.get('weather.' + STN)
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_latlon(self, mock_pynws):
"""Test for successfully setting up the NWS platform with lat/lon."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
'api_key': 'test_email',
'latitude': self.lat,
'longitude': self.lon,
}
})
assert self.hass.states.get('weather.' + STN)
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_setup_failure_mode(self, mock_pynws):
"""Test for unsuccessfully setting up incorrect mode."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
'api_key': 'test_email',
'mode': 'abc',
}
})
assert self.hass.states.get('weather.' + STN) is None
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_setup_failure_no_apikey(self, mock_pynws):
"""Test for unsuccessfully setting up without api_key."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'platform': 'nws',
}
})
assert self.hass.states.get('weather.' + STN) is None
class TestNwsMetric(unittest.TestCase):
"""Test the NWS weather component using metric units."""
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.hass.config.units = METRIC_SYSTEM
self.lat = self.hass.config.latitude = 40.00
self.lon = self.hass.config.longitude = -8.00
def tearDown(self):
"""Stop down everything that was started."""
self.hass.stop()
@MockDependency("pynws")
@patch("pynws.Nws", new=MockNws)
def test_metric(self, mock_pynws):
"""Test for successfully setting up the NWS platform with name."""
assert setup_component(self.hass, weather.DOMAIN, {
'weather': {
'name': 'HomeWeather',
'platform': 'nws',
'api_key': 'test_email',
}
})
state = self.hass.states.get('weather.homeweather')
assert state.state == 'sunny'
data = state.attributes
assert data.get(ATTR_WEATHER_TEMPERATURE) == \
display_temp(self.hass, 7, TEMP_CELSIUS, PRECISION_WHOLE)
assert data.get(ATTR_WEATHER_HUMIDITY) == 10
assert data.get(ATTR_WEATHER_PRESSURE) == round(
convert_pressure(30000, PRESSURE_PA, PRESSURE_HPA))
# m/s to km/hr
assert data.get(ATTR_WEATHER_WIND_SPEED) == round(10 * 3.6)
assert data.get(ATTR_WEATHER_WIND_BEARING) == 180
assert data.get(ATTR_WEATHER_VISIBILITY) == round(
convert_distance(10000, LENGTH_METERS, LENGTH_KILOMETERS))
assert state.attributes.get('friendly_name') == 'HomeWeather'
forecast = data.get(ATTR_FORECAST)
assert forecast[0].get(ATTR_FORECAST_CONDITION) == 'lightning-rainy'
assert forecast[0].get(ATTR_FORECAST_PRECIP_PROB) == 40
assert forecast[0].get(ATTR_FORECAST_TEMP) == round(
convert_temperature(41, TEMP_FAHRENHEIT, TEMP_CELSIUS))
assert forecast[0].get(ATTR_FORECAST_TIME) == \
'2018-12-21T15:00:00-05:00'
assert forecast[0].get(ATTR_FORECAST_WIND_BEARING) == 180
assert forecast[0].get(ATTR_FORECAST_WIND_SPEED) == round(
convert_distance(9, LENGTH_MILES, LENGTH_KILOMETERS))