evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

U.S. Geological Survey Earthquake Hazards Program Feed platform (#18207)

Browse source 
* new platform for usgs earthquake hazards program feed

* lint and pylint issues

* fixed config access

* shortened names of platform, classes, etc.

* refactored tests

* fixed hound

* regenerated requirements

* refactored tests

* fixed hound
This commit is contained in:
Malte Franken 2018-11-27 22:55:15 +11:00 • committed by Martin Hjelmare
parent 9a25054a0d
commit 013e181497
4 changed files with 464 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

268
homeassistant/components/geo_location/usgs_earthquakes_feed.py Normal file
View file

@ -0,0 +1,268 @@
"""
U.S. Geological Survey Earthquake Hazards Program Feed platform.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/geo_location/usgs_earthquakes_feed/
"""
from datetime import timedelta
import logging
from typing import Optional
import voluptuous as vol
from homeassistant.components.geo_location import (
PLATFORM_SCHEMA, GeoLocationEvent)
from homeassistant.const import (
ATTR_ATTRIBUTION, CONF_RADIUS, CONF_SCAN_INTERVAL,
EVENT_HOMEASSISTANT_START, CONF_LATITUDE, CONF_LONGITUDE)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect, dispatcher_send)
from homeassistant.helpers.event import track_time_interval
REQUIREMENTS = ['geojson_client==0.3']
_LOGGER = logging.getLogger(__name__)
ATTR_ALERT = 'alert'
ATTR_EXTERNAL_ID = 'external_id'
ATTR_MAGNITUDE = 'magnitude'
ATTR_PLACE = 'place'
ATTR_STATUS = 'status'
ATTR_TIME = 'time'
ATTR_TYPE = 'type'
ATTR_UPDATED = 'updated'
CONF_FEED_TYPE = 'feed_type'
CONF_MINIMUM_MAGNITUDE = 'minimum_magnitude'
DEFAULT_MINIMUM_MAGNITUDE = 0.0
DEFAULT_RADIUS_IN_KM = 50.0
DEFAULT_UNIT_OF_MEASUREMENT = 'km'
SCAN_INTERVAL = timedelta(minutes=5)
SIGNAL_DELETE_ENTITY = 'usgs_earthquakes_feed_delete_{}'
SIGNAL_UPDATE_ENTITY = 'usgs_earthquakes_feed_update_{}'
SOURCE = 'usgs_earthquakes_feed'
VALID_FEED_TYPES = [
'past_hour_significant_earthquakes',
'past_hour_m45_earthquakes',
'past_hour_m25_earthquakes',
'past_hour_m10_earthquakes',
'past_hour_all_earthquakes',
'past_day_significant_earthquakes',
'past_day_m45_earthquakes',
'past_day_m25_earthquakes',
'past_day_m10_earthquakes',
'past_day_all_earthquakes',
'past_week_significant_earthquakes',
'past_week_m45_earthquakes',
'past_week_m25_earthquakes',
'past_week_m10_earthquakes',
'past_week_all_earthquakes',
'past_month_significant_earthquakes',
'past_month_m45_earthquakes',
'past_month_m25_earthquakes',
'past_month_m10_earthquakes',
'past_month_all_earthquakes',
]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_FEED_TYPE): vol.In(VALID_FEED_TYPES),
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS_IN_KM): vol.Coerce(float),
vol.Optional(CONF_MINIMUM_MAGNITUDE, default=DEFAULT_MINIMUM_MAGNITUDE):
vol.All(vol.Coerce(float), vol.Range(min=0))
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the USGS Earthquake Hazards Program Feed platform."""
scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
feed_type = config[CONF_FEED_TYPE]
coordinates = (config.get(CONF_LATITUDE, hass.config.latitude),
config.get(CONF_LONGITUDE, hass.config.longitude))
radius_in_km = config[CONF_RADIUS]
minimum_magnitude = config[CONF_MINIMUM_MAGNITUDE]
# Initialize the entity manager.
feed = UsgsEarthquakesFeedEntityManager(
hass, add_entities, scan_interval, coordinates, feed_type,
radius_in_km, minimum_magnitude)
def start_feed_manager(event):
"""Start feed manager."""
feed.startup()
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_feed_manager)
class UsgsEarthquakesFeedEntityManager:
"""Feed Entity Manager for USGS Earthquake Hazards Program feed."""
def __init__(self, hass, add_entities, scan_interval, coordinates,
feed_type, radius_in_km, minimum_magnitude):
"""Initialize the Feed Entity Manager."""
from geojson_client.usgs_earthquake_hazards_program_feed \
import UsgsEarthquakeHazardsProgramFeedManager
self._hass = hass
self._feed_manager = UsgsEarthquakeHazardsProgramFeedManager(
self._generate_entity, self._update_entity, self._remove_entity,
coordinates, feed_type, filter_radius=radius_in_km,
filter_minimum_magnitude=minimum_magnitude)
self._add_entities = add_entities
self._scan_interval = scan_interval
def startup(self):
"""Start up this manager."""
self._feed_manager.update()
self._init_regular_updates()
def _init_regular_updates(self):
"""Schedule regular updates at the specified interval."""
track_time_interval(
self._hass, lambda now: self._feed_manager.update(),
self._scan_interval)
def get_entry(self, external_id):
"""Get feed entry by external id."""
return self._feed_manager.feed_entries.get(external_id)
def _generate_entity(self, external_id):
"""Generate new entity."""
new_entity = UsgsEarthquakesEvent(self, external_id)
# Add new entities to HA.
self._add_entities([new_entity], True)
def _update_entity(self, external_id):
"""Update entity."""
dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY.format(external_id))
def _remove_entity(self, external_id):
"""Remove entity."""
dispatcher_send(self._hass, SIGNAL_DELETE_ENTITY.format(external_id))
class UsgsEarthquakesEvent(GeoLocationEvent):
"""This represents an external event with USGS Earthquake data."""
def __init__(self, feed_manager, external_id):
"""Initialize entity with data from feed entry."""
self._feed_manager = feed_manager
self._external_id = external_id
self._name = None
self._distance = None
self._latitude = None
self._longitude = None
self._attribution = None
self._place = None
self._magnitude = None
self._time = None
self._updated = None
self._status = None
self._type = None
self._alert = None
self._remove_signal_delete = None
self._remove_signal_update = None
async def async_added_to_hass(self):
"""Call when entity is added to hass."""
self._remove_signal_delete = async_dispatcher_connect(
self.hass, SIGNAL_DELETE_ENTITY.format(self._external_id),
self._delete_callback)
self._remove_signal_update = async_dispatcher_connect(
self.hass, SIGNAL_UPDATE_ENTITY.format(self._external_id),
self._update_callback)
@callback
def _delete_callback(self):
"""Remove this entity."""
self._remove_signal_delete()
self._remove_signal_update()
self.hass.async_create_task(self.async_remove())
@callback
def _update_callback(self):
"""Call update method."""
self.async_schedule_update_ha_state(True)
@property
def should_poll(self):
"""No polling needed for USGS Earthquake events."""
return False
async def async_update(self):
"""Update this entity from the data held in the feed manager."""
_LOGGER.debug("Updating %s", self._external_id)
feed_entry = self._feed_manager.get_entry(self._external_id)
if feed_entry:
self._update_from_feed(feed_entry)
def _update_from_feed(self, feed_entry):
"""Update the internal state from the provided feed entry."""
self._name = feed_entry.title
self._distance = feed_entry.distance_to_home
self._latitude = feed_entry.coordinates[0]
self._longitude = feed_entry.coordinates[1]
self._attribution = feed_entry.attribution
self._place = feed_entry.place
self._magnitude = feed_entry.magnitude
self._time = feed_entry.time
self._updated = feed_entry.updated
self._status = feed_entry.status
self._type = feed_entry.type
self._alert = feed_entry.alert
@property
def source(self) -> str:
"""Return source value of this external event."""
return SOURCE
@property
def name(self) -> Optional[str]:
"""Return the name of the entity."""
return self._name
@property
def distance(self) -> Optional[float]:
"""Return distance value of this external event."""
return self._distance
@property
def latitude(self) -> Optional[float]:
"""Return latitude value of this external event."""
return self._latitude
@property
def longitude(self) -> Optional[float]:
"""Return longitude value of this external event."""
return self._longitude
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return DEFAULT_UNIT_OF_MEASUREMENT
@property
def device_state_attributes(self):
"""Return the device state attributes."""
attributes = {}
for key, value in (
(ATTR_EXTERNAL_ID, self._external_id),
(ATTR_PLACE, self._place),
(ATTR_MAGNITUDE, self._magnitude),
(ATTR_TIME, self._time),
(ATTR_UPDATED, self._updated),
(ATTR_STATUS, self._status),
(ATTR_TYPE, self._type),
(ATTR_ALERT, self._alert),
(ATTR_ATTRIBUTION, self._attribution),
):
if value or isinstance(value, bool):
attributes[key] = value
return attributes

1
requirements_all.txt
View file

@ -416,6 +416,7 @@ geizhals==0.0.7
# homeassistant.components.geo_location.geo_json_events
# homeassistant.components.geo_location.nsw_rural_fire_service_feed
# homeassistant.components.geo_location.usgs_earthquakes_feed
geojson_client==0.3
# homeassistant.components.sensor.geo_rss_events

1
requirements_test_all.txt
View file

@ -76,6 +76,7 @@ gTTS-token==1.1.2
# homeassistant.components.geo_location.geo_json_events
# homeassistant.components.geo_location.nsw_rural_fire_service_feed
# homeassistant.components.geo_location.usgs_earthquakes_feed
geojson_client==0.3
# homeassistant.components.sensor.geo_rss_events

194
tests/components/geo_location/test_usgs_earthquakes_feed.py Normal file
View file

@ -0,0 +1,194 @@
"""The tests for the USGS Earthquake Hazards Program Feed platform."""
import datetime
from unittest.mock import patch, MagicMock, call
from homeassistant.components import geo_location
from homeassistant.components.geo_location import ATTR_SOURCE
from homeassistant.components.geo_location\
.usgs_earthquakes_feed import \
ATTR_ALERT, ATTR_EXTERNAL_ID, SCAN_INTERVAL, ATTR_PLACE, \
ATTR_MAGNITUDE, ATTR_STATUS, ATTR_TYPE, \
ATTR_TIME, ATTR_UPDATED, CONF_FEED_TYPE
from homeassistant.const import EVENT_HOMEASSISTANT_START, \
CONF_RADIUS, ATTR_LATITUDE, ATTR_LONGITUDE, ATTR_FRIENDLY_NAME, \
ATTR_UNIT_OF_MEASUREMENT, ATTR_ATTRIBUTION, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, async_fire_time_changed
import homeassistant.util.dt as dt_util
CONFIG = {
geo_location.DOMAIN: [
{
'platform': 'usgs_earthquakes_feed',
CONF_FEED_TYPE: 'past_hour_m25_earthquakes',
CONF_RADIUS: 200
}
]
}
CONFIG_WITH_CUSTOM_LOCATION = {
geo_location.DOMAIN: [
{
'platform': 'usgs_earthquakes_feed',
CONF_FEED_TYPE: 'past_hour_m25_earthquakes',
CONF_RADIUS: 200,
CONF_LATITUDE: 15.1,
CONF_LONGITUDE: 25.2
}
]
}
def _generate_mock_feed_entry(external_id, title, distance_to_home,
coordinates, place=None,
attribution=None, time=None, updated=None,
magnitude=None, status=None,
entry_type=None, alert=None):
"""Construct a mock feed entry for testing purposes."""
feed_entry = MagicMock()
feed_entry.external_id = external_id
feed_entry.title = title
feed_entry.distance_to_home = distance_to_home
feed_entry.coordinates = coordinates
feed_entry.place = place
feed_entry.attribution = attribution
feed_entry.time = time
feed_entry.updated = updated
feed_entry.magnitude = magnitude
feed_entry.status = status
feed_entry.type = entry_type
feed_entry.alert = alert
return feed_entry
async def test_setup(hass):
"""Test the general setup of the platform."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
'1234', 'Title 1', 15.5, (-31.0, 150.0),
place='Location 1', attribution='Attribution 1',
time=datetime.datetime(2018, 9, 22, 8, 0,
tzinfo=datetime.timezone.utc),
updated=datetime.datetime(2018, 9, 22, 9, 0,
tzinfo=datetime.timezone.utc),
magnitude=5.7, status='Status 1', entry_type='Type 1',
alert='Alert 1')
mock_entry_2 = _generate_mock_feed_entry(
'2345', 'Title 2', 20.5, (-31.1, 150.1))
mock_entry_3 = _generate_mock_feed_entry(
'3456', 'Title 3', 25.5, (-31.2, 150.2))
mock_entry_4 = _generate_mock_feed_entry(
'4567', 'Title 4', 12.5, (-31.3, 150.3))
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch('homeassistant.util.dt.utcnow', return_value=utcnow), \
patch('geojson_client.usgs_earthquake_hazards_program_feed.'
'UsgsEarthquakeHazardsProgramFeed') as mock_feed:
mock_feed.return_value.update.return_value = 'OK', [mock_entry_1,
mock_entry_2,
mock_entry_3]
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(
hass, geo_location.DOMAIN, CONFIG)
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 3
state = hass.states.get("geo_location.title_1")
assert state is not None
assert state.name == "Title 1"
assert state.attributes == {
ATTR_EXTERNAL_ID: "1234", ATTR_LATITUDE: -31.0,
ATTR_LONGITUDE: 150.0, ATTR_FRIENDLY_NAME: "Title 1",
ATTR_PLACE: "Location 1",
ATTR_ATTRIBUTION: "Attribution 1",
ATTR_TIME:
datetime.datetime(
2018, 9, 22, 8, 0, tzinfo=datetime.timezone.utc),
ATTR_UPDATED:
datetime.datetime(
2018, 9, 22, 9, 0, tzinfo=datetime.timezone.utc),
ATTR_STATUS: 'Status 1', ATTR_TYPE: 'Type 1',
ATTR_ALERT: 'Alert 1', ATTR_MAGNITUDE: 5.7,
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'usgs_earthquakes_feed'}
assert round(abs(float(state.state)-15.5), 7) == 0
state = hass.states.get("geo_location.title_2")
assert state is not None
assert state.name == "Title 2"
assert state.attributes == {
ATTR_EXTERNAL_ID: "2345", ATTR_LATITUDE: -31.1,
ATTR_LONGITUDE: 150.1, ATTR_FRIENDLY_NAME: "Title 2",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'usgs_earthquakes_feed'}
assert round(abs(float(state.state)-20.5), 7) == 0
state = hass.states.get("geo_location.title_3")
assert state is not None
assert state.name == "Title 3"
assert state.attributes == {
ATTR_EXTERNAL_ID: "3456", ATTR_LATITUDE: -31.2,
ATTR_LONGITUDE: 150.2, ATTR_FRIENDLY_NAME: "Title 3",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'usgs_earthquakes_feed'}
assert round(abs(float(state.state)-25.5), 7) == 0
# Simulate an update - one existing, one new entry,
# one outdated entry
mock_feed.return_value.update.return_value = 'OK', [
mock_entry_1, mock_entry_4, mock_entry_3]
async_fire_time_changed(hass, utcnow + SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 3
# Simulate an update - empty data, but successful update,
# so no changes to entities.
mock_feed.return_value.update.return_value = 'OK_NO_DATA', None
async_fire_time_changed(hass, utcnow + 2 * SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 3
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = 'ERROR', None
async_fire_time_changed(hass, utcnow + 3 * SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 0
async def test_setup_with_custom_location(hass):
"""Test the setup with a custom location."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
'1234', 'Title 1', 20.5, (-31.1, 150.1))
with patch('geojson_client.usgs_earthquake_hazards_program_feed.'
'UsgsEarthquakeHazardsProgramFeed') as mock_feed:
mock_feed.return_value.update.return_value = 'OK', [mock_entry_1]
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(
hass, geo_location.DOMAIN, CONFIG_WITH_CUSTOM_LOCATION)
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
assert mock_feed.call_args == call(
(15.1, 25.2), 'past_hour_m25_earthquakes',
filter_minimum_magnitude=0.0, filter_radius=200.0)