Weather Platform - IPMA (#14716)
* initial commit * lint * update with pyipma * Added test * Added test * lint * missing dep * address comments * lint * make sure list is iterable * don't bother with list * mock dependency * no need to add test requirements * last correction
This commit is contained in:
Diogo Gomes
Martin Hjelmare
parent
855ed2b4e4
commit
aec425d1f6
3 changed files with 260 additions and 0 deletions
172
homeassistant/components/weather/ipma.py
Normal file
172
homeassistant/components/weather/ipma.py
Normal file
|
|
@ -0,0 +1,172 @@
|
|||
"""
|
||||
Support for IPMA weather service.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/weather.ipma/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
import async_timeout
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.weather import (
|
||||
WeatherEntity, PLATFORM_SCHEMA, ATTR_FORECAST_CONDITION,
|
||||
ATTR_FORECAST_PRECIPITATION, ATTR_FORECAST_TEMP,
|
||||
ATTR_FORECAST_TEMP_LOW, ATTR_FORECAST_TIME)
|
||||
from homeassistant.const import \
|
||||
CONF_NAME, TEMP_CELSIUS, CONF_LATITUDE, CONF_LONGITUDE
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
REQUIREMENTS = ['pyipma==1.1.3']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTRIBUTION = 'Instituto Português do Mar e Atmosfera'
|
||||
|
||||
ATTR_WEATHER_DESCRIPTION = "description"
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
|
||||
|
||||
CONDITION_CLASSES = {
|
||||
'cloudy': [4, 5, 24, 25, 27],
|
||||
'fog': [16, 17, 26],
|
||||
'hail': [21, 22],
|
||||
'lightning': [19],
|
||||
'lightning-rainy': [20, 23],
|
||||
'partlycloudy': [2, 3],
|
||||
'pouring': [8, 11],
|
||||
'rainy': [6, 7, 9, 10, 12, 13, 14, 15],
|
||||
'snowy': [18],
|
||||
'snowy-rainy': [],
|
||||
'sunny': [1],
|
||||
'windy': [],
|
||||
'windy-variant': [],
|
||||
'exceptional': [],
|
||||
}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
vol.Optional(CONF_LATITUDE): cv.latitude,
|
||||
vol.Optional(CONF_LONGITUDE): cv.longitude,
|
||||
})
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_devices,
|
||||
discovery_info=None):
|
||||
"""Set up the ipma platform."""
|
||||
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
|
||||
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
|
||||
|
||||
if None in (latitude, longitude):
|
||||
_LOGGER.error("Latitude or longitude not set in Home Assistant config")
|
||||
return
|
||||
|
||||
from pyipma import Station
|
||||
|
||||
websession = async_get_clientsession(hass)
|
||||
with async_timeout.timeout(10, loop=hass.loop):
|
||||
station = await Station.get(websession, float(latitude),
|
||||
float(longitude))
|
||||
|
||||
_LOGGER.debug("Initializing ipma weather: coordinates %s, %s",
|
||||
latitude, longitude)
|
||||
|
||||
async_add_devices([IPMAWeather(station, config)], True)
|
||||
|
||||
|
||||
class IPMAWeather(WeatherEntity):
|
||||
"""Representation of a weather condition."""
|
||||
|
||||
def __init__(self, station, config):
|
||||
"""Initialise the platform with a data instance and station name."""
|
||||
self._station_name = config.get(CONF_NAME, station.local)
|
||||
self._station = station
|
||||
self._condition = None
|
||||
self._forecast = None
|
||||
self._description = None
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
async def async_update(self):
|
||||
"""Update Condition and Forecast."""
|
||||
with async_timeout.timeout(10, loop=self.hass.loop):
|
||||
self._condition = await self._station.observation()
|
||||
self._forecast = await self._station.forecast()
|
||||
self._description = self._forecast[0].description
|
||||
|
||||
@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 condition(self):
|
||||
"""Return the current condition."""
|
||||
return next((k for k, v in CONDITION_CLASSES.items()
|
||||
if self._forecast[0].idWeatherType in v), None)
|
||||
|
||||
@property
|
||||
def temperature(self):
|
||||
"""Return the current temperature."""
|
||||
return self._condition.temperature
|
||||
|
||||
@property
|
||||
def pressure(self):
|
||||
"""Return the current pressure."""
|
||||
return self._condition.pressure
|
||||
|
||||
@property
|
||||
def humidity(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._condition.humidity
|
||||
|
||||
@property
|
||||
def wind_speed(self):
|
||||
"""Return the current windspeed."""
|
||||
return self._condition.windspeed
|
||||
|
||||
@property
|
||||
def wind_bearing(self):
|
||||
"""Return the current wind bearing (degrees)."""
|
||||
return self._condition.winddirection
|
||||
|
||||
@property
|
||||
def temperature_unit(self):
|
||||
"""Return the unit of measurement."""
|
||||
return TEMP_CELSIUS
|
||||
|
||||
@property
|
||||
def forecast(self):
|
||||
"""Return the forecast array."""
|
||||
if self._forecast:
|
||||
fcdata_out = []
|
||||
for data_in in self._forecast:
|
||||
data_out = {}
|
||||
data_out[ATTR_FORECAST_TIME] = data_in.forecastDate
|
||||
data_out[ATTR_FORECAST_CONDITION] =\
|
||||
next((k for k, v in CONDITION_CLASSES.items()
|
||||
if int(data_in.idWeatherType) in v), None)
|
||||
data_out[ATTR_FORECAST_TEMP_LOW] = data_in.tMin
|
||||
data_out[ATTR_FORECAST_TEMP] = data_in.tMax
|
||||
data_out[ATTR_FORECAST_PRECIPITATION] = data_in.precipitaProb
|
||||
|
||||
fcdata_out.append(data_out)
|
||||
|
||||
return fcdata_out
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
data = dict()
|
||||
|
||||
if self._description:
|
||||
data[ATTR_WEATHER_DESCRIPTION] = self._description
|
||||
|
||||
return data
|
||||
|
|
@ -834,6 +834,9 @@ pyialarm==0.2
|
|||
# homeassistant.components.device_tracker.icloud
|
||||
pyicloud==0.9.1
|
||||
|
||||
# homeassistant.components.weather.ipma
|
||||
pyipma==1.1.3
|
||||
|
||||
# homeassistant.components.sensor.irish_rail_transport
|
||||
pyirishrail==0.0.2
|
||||
|
||||
|
|
|
|||
85
tests/components/weather/test_ipma.py
Normal file
85
tests/components/weather/test_ipma.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
"""The tests for the IPMA weather component."""
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from collections import namedtuple
|
||||
|
||||
from homeassistant.components import weather
|
||||
from homeassistant.components.weather import (
|
||||
ATTR_WEATHER_HUMIDITY, ATTR_WEATHER_PRESSURE, ATTR_WEATHER_TEMPERATURE,
|
||||
ATTR_WEATHER_WIND_BEARING, ATTR_WEATHER_WIND_SPEED)
|
||||
from homeassistant.util.unit_system import METRIC_SYSTEM
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
from tests.common import get_test_home_assistant, MockDependency
|
||||
|
||||
|
||||
class MockStation():
|
||||
"""Mock Station from pyipma."""
|
||||
|
||||
@classmethod
|
||||
async def get(cls, websession, lat, lon):
|
||||
"""Mock Factory."""
|
||||
return MockStation()
|
||||
|
||||
async def observation(self):
|
||||
"""Mock Observation."""
|
||||
Observation = namedtuple('Observation', ['temperature', 'humidity',
|
||||
'windspeed', 'winddirection',
|
||||
'precipitation', 'pressure',
|
||||
'description'])
|
||||
|
||||
return Observation(18, 71.0, 3.94, 'NW', 0, 1000.0, '---')
|
||||
|
||||
async def forecast(self):
|
||||
"""Mock Forecast."""
|
||||
Forecast = namedtuple('Forecast', ['precipitaProb', 'tMin', 'tMax',
|
||||
'predWindDir', 'idWeatherType',
|
||||
'classWindSpeed', 'longitude',
|
||||
'forecastDate', 'classPrecInt',
|
||||
'latitude', 'description'])
|
||||
|
||||
return [Forecast(73.0, 13.7, 18.7, 'NW', 6, 2, -8.64,
|
||||
'2018-05-31', 2, 40.61,
|
||||
'Aguaceiros, com vento Moderado de Noroeste')]
|
||||
|
||||
@property
|
||||
def local(self):
|
||||
"""Mock location."""
|
||||
return "HomeTown"
|
||||
|
||||
|
||||
class TestIPMA(unittest.TestCase):
|
||||
"""Test the IPMA weather component."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup 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("pyipma")
|
||||
@patch("pyipma.Station", new=MockStation)
|
||||
def test_setup(self, mock_pyipma):
|
||||
"""Test for successfully setting up the IPMA platform."""
|
||||
self.assertTrue(setup_component(self.hass, weather.DOMAIN, {
|
||||
'weather': {
|
||||
'name': 'HomeTown',
|
||||
'platform': 'ipma',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('weather.hometown')
|
||||
self.assertEqual(state.state, 'rainy')
|
||||
|
||||
data = state.attributes
|
||||
self.assertEqual(data.get(ATTR_WEATHER_TEMPERATURE), 18.0)
|
||||
self.assertEqual(data.get(ATTR_WEATHER_HUMIDITY), 71)
|
||||
self.assertEqual(data.get(ATTR_WEATHER_PRESSURE), 1000.0)
|
||||
self.assertEqual(data.get(ATTR_WEATHER_WIND_SPEED), 3.94)
|
||||
self.assertEqual(data.get(ATTR_WEATHER_WIND_BEARING), 'NW')
|
||||
self.assertEqual(state.attributes.get('friendly_name'), 'HomeTown')
|
||||
Loading…
Add table
Reference in a new issue