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

Queensland bushfire alert feed platform (#24473)

Browse source 
* initial version of qfes bushfire geolocation platform

* removed all occurrences of legally protected names; using new georss library

* regenerated codeowners

* fixed pylint

* added one more valid category

* moved library import to top and ran isort
This commit is contained in:
Malte Franken 2019-06-19 21:59:29 +10:00 • committed by Charles Garwood
parent 9413b5a415
commit 03bb3d9ddc
9 changed files with 439 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
CODEOWNERS
View file

@ -199,6 +199,7 @@ homeassistant/components/ps4/* @ktnrg45
homeassistant/components/ptvsd/* @swamp-ig
homeassistant/components/push/* @dgomes
homeassistant/components/pvoutput/* @fabaff
homeassistant/components/qld_bushfire/* @exxamalte
homeassistant/components/qnap/* @colinodell
homeassistant/components/quantum_gateway/* @cisasteelersfan
homeassistant/components/qwikswitch/* @kellerza

1
homeassistant/components/qld_bushfire/__init__.py Normal file
View file

@ -0,0 +1 @@
"""The qld_bushfire component."""

228
homeassistant/components/qld_bushfire/geo_location.py Normal file
View file

@ -0,0 +1,228 @@
"""Support for Queensland Bushfire Alert Feeds."""
from datetime import timedelta
import logging
from typing import Optional
from georss_qld_bushfire_alert_client import QldBushfireAlertFeedManager
import voluptuous as vol
from homeassistant.components.geo_location import (
PLATFORM_SCHEMA, GeolocationEvent)
from homeassistant.const import (
ATTR_ATTRIBUTION, CONF_LATITUDE, CONF_LONGITUDE, CONF_RADIUS,
CONF_SCAN_INTERVAL, EVENT_HOMEASSISTANT_START)
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
_LOGGER = logging.getLogger(__name__)
ATTR_CATEGORY = 'category'
ATTR_EXTERNAL_ID = 'external_id'
ATTR_PUBLICATION_DATE = 'publication_date'
ATTR_STATUS = 'status'
ATTR_UPDATED_DATE = 'updated_date'
CONF_CATEGORIES = 'categories'
DEFAULT_RADIUS_IN_KM = 20.0
DEFAULT_UNIT_OF_MEASUREMENT = 'km'
SCAN_INTERVAL = timedelta(minutes=5)
SIGNAL_DELETE_ENTITY = 'qld_bushfire_delete_{}'
SIGNAL_UPDATE_ENTITY = 'qld_bushfire_update_{}'
SOURCE = 'qld_bushfire'
VALID_CATEGORIES = [
'Emergency Warning',
'Watch and Act',
'Advice',
'Notification',
'Information',
]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
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_CATEGORIES, default=[]):
vol.All(cv.ensure_list, [vol.In(VALID_CATEGORIES)]),
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Queensland Bushfire Alert Feed platform."""
scan_interval = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
coordinates = (config.get(CONF_LATITUDE, hass.config.latitude),
config.get(CONF_LONGITUDE, hass.config.longitude))
radius_in_km = config[CONF_RADIUS]
categories = config[CONF_CATEGORIES]
# Initialize the entity manager.
feed = QldBushfireFeedEntityManager(
hass, add_entities, scan_interval, coordinates, radius_in_km,
categories)
def start_feed_manager(event):
"""Start feed manager."""
feed.startup()
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_feed_manager)
class QldBushfireFeedEntityManager:
"""Feed Entity Manager for Qld Bushfire Alert GeoRSS feed."""
def __init__(self, hass, add_entities, scan_interval, coordinates,
radius_in_km, categories):
"""Initialize the Feed Entity Manager."""
self._hass = hass
self._feed_manager = QldBushfireAlertFeedManager(
self._generate_entity, self._update_entity, self._remove_entity,
coordinates, filter_radius=radius_in_km,
filter_categories=categories)
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 = QldBushfireLocationEvent(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 QldBushfireLocationEvent(GeolocationEvent):
"""This represents an external event with Qld Bushfire feed 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._category = None
self._publication_date = None
self._updated_date = None
self._status = 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 Qld Bushfire Alert feed location 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._category = feed_entry.category
self._publication_date = feed_entry.published
self._updated_date = feed_entry.updated
self._status = feed_entry.status
@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_CATEGORY, self._category),
(ATTR_ATTRIBUTION, self._attribution),
(ATTR_PUBLICATION_DATE, self._publication_date),
(ATTR_UPDATED_DATE, self._updated_date),
(ATTR_STATUS, self._status)
):
if value or isinstance(value, bool):
attributes[key] = value
return attributes

12
homeassistant/components/qld_bushfire/manifest.json Normal file
View file

@ -0,0 +1,12 @@
{
"domain": "qld_bushfire",
"name": "Queensland Bushfire Alert",
"documentation": "https://www.home-assistant.io/components/qld_bushfire",
"requirements": [
"georss_qld_bushfire_alert_client==0.3"
],
"dependencies": [],
"codeowners": [
"@exxamalte"
]
}

3
requirements_all.txt
View file

@ -510,6 +510,9 @@ georss_generic_client==0.2
# homeassistant.components.ign_sismologia
georss_ign_sismologia_client==0.2
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.3
# homeassistant.components.gitter
gitterpy==0.1.7

3
requirements_test_all.txt
View file

@ -135,6 +135,9 @@ georss_generic_client==0.2
# homeassistant.components.ign_sismologia
georss_ign_sismologia_client==0.2
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.3
# homeassistant.components.google
google-api-python-client==1.6.4

1
script/gen_requirements_all.py
View file

@ -70,6 +70,7 @@ TEST_REQUIREMENTS = (
'geopy',
'georss_generic_client',
'georss_ign_sismologia_client',
'georss_qld_bushfire_alert_client',
'google-api-python-client',
'gTTS-token',
'ha-ffmpeg',

1
tests/components/qld_bushfire/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the qld_bushfire component."""

189
tests/components/qld_bushfire/test_geo_location.py Normal file
View file

@ -0,0 +1,189 @@
"""The tests for the Queensland Bushfire Alert 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.qld_bushfire.geo_location import (
ATTR_EXTERNAL_ID, SCAN_INTERVAL, ATTR_CATEGORY,
ATTR_STATUS, ATTR_PUBLICATION_DATE, ATTR_UPDATED_DATE)
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': 'qld_bushfire',
CONF_RADIUS: 200
}
]
}
CONFIG_WITH_CUSTOM_LOCATION = {
geo_location.DOMAIN: [
{
'platform': 'qld_bushfire',
CONF_RADIUS: 200,
CONF_LATITUDE: 40.4,
CONF_LONGITUDE: -3.7
}
]
}
def _generate_mock_feed_entry(external_id, title, distance_to_home,
coordinates, category=None, attribution=None,
published=None, updated=None, status=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.category = category
feed_entry.attribution = attribution
feed_entry.published = published
feed_entry.updated = updated
feed_entry.status = status
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, (38.0, -3.0),
category='Category 1',
attribution='Attribution 1',
published=datetime.datetime(2018, 9, 22, 8, 0,
tzinfo=datetime.timezone.utc),
updated=datetime.datetime(2018, 9, 22, 8, 10,
tzinfo=datetime.timezone.utc),
status='Status 1')
mock_entry_2 = _generate_mock_feed_entry(
'2345', 'Title 2', 20.5, (38.1, -3.1))
mock_entry_3 = _generate_mock_feed_entry(
'3456', 'Title 3', 25.5, (38.2, -3.2))
mock_entry_4 = _generate_mock_feed_entry(
'4567', 'Title 4', 12.5, (38.3, -3.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('georss_qld_bushfire_alert_client.'
'QldBushfireAlertFeed') 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: 38.0,
ATTR_LONGITUDE: -3.0,
ATTR_FRIENDLY_NAME: "Title 1",
ATTR_CATEGORY: "Category 1",
ATTR_ATTRIBUTION: "Attribution 1",
ATTR_PUBLICATION_DATE:
datetime.datetime(
2018, 9, 22, 8, 0, tzinfo=datetime.timezone.utc),
ATTR_UPDATED_DATE:
datetime.datetime(
2018, 9, 22, 8, 10, tzinfo=datetime.timezone.utc),
ATTR_STATUS: 'Status 1',
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'qld_bushfire'}
assert float(state.state) == 15.5
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: 38.1,
ATTR_LONGITUDE: -3.1,
ATTR_FRIENDLY_NAME: "Title 2",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'qld_bushfire'}
assert float(state.state) == 20.5
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: 38.2,
ATTR_LONGITUDE: -3.2,
ATTR_FRIENDLY_NAME: "Title 3",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: 'qld_bushfire'}
assert float(state.state) == 25.5
# 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, (38.1, -3.1), category="Category 1")
with patch('georss_qld_bushfire_alert_client.'
'QldBushfireAlertFeed') 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(
(40.4, -3.7), filter_categories=[], filter_radius=200.0)