Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GeoRSS events sensor refactored () 

* refactored geo_rss_events sensor to make use of new georss-client library that handles the communication with the rss feed

* fixed lint error
pull/17059/head^2
Malte Franken 2018-10-02 18:20:51 +10:00 committed by Paulus Schoutsen
parent b0b3620b2b
commit 13af61e103
6 changed files with 154 additions and 316 deletions

150
homeassistant/components/sensor/geo_rss_events.py
View File

@ -9,7 +9,6 @@ For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.geo_rss_events/
"""
import logging
from collections import namedtuple
from datetime import timedelta
import voluptuous as vol
@ -19,9 +18,8 @@ from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
STATE_UNKNOWN, CONF_UNIT_OF_MEASUREMENT, CONF_NAME, CONF_RADIUS, CONF_URL)
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
REQUIREMENTS = ['feedparser==5.2.1', 'haversine==0.4.5']
REQUIREMENTS = ['georss_client==0.1']
_LOGGER = logging.getLogger(__name__)
@ -38,9 +36,6 @@ DEFAULT_UNIT_OF_MEASUREMENT = 'Events'
DOMAIN = 'geo_rss_events'
# Minimum time between updates from the source.
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
SCAN_INTERVAL = timedelta(minutes=5)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
@ -67,18 +62,17 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
_LOGGER.debug("latitude=%s, longitude=%s, url=%s, radius=%s",
home_latitude, home_longitude, url, radius_in_km)
# Initialise update service.
data = GeoRssServiceData(home_latitude, home_longitude, url, radius_in_km)
data.update()
# Create all sensors based on categories.
devices = []
if not categories:
device = GeoRssServiceSensor(None, data, name, unit_of_measurement)
device = GeoRssServiceSensor((home_latitude, home_longitude), url,
radius_in_km, None, name,
unit_of_measurement)
devices.append(device)
else:
for category in categories:
device = GeoRssServiceSensor(category, data, name,
device = GeoRssServiceSensor((home_latitude, home_longitude), url,
radius_in_km, category, name,
unit_of_measurement)
devices.append(device)
add_entities(devices, True)
@ -87,14 +81,18 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
class GeoRssServiceSensor(Entity):
"""Representation of a Sensor."""
def __init__(self, category, data, service_name, unit_of_measurement):
def __init__(self, home_coordinates, url, radius, category, service_name,
unit_of_measurement):
"""Initialize the sensor."""
self._category = category
self._data = data
self._service_name = service_name
self._state = STATE_UNKNOWN
self._state_attributes = None
self._unit_of_measurement = unit_of_measurement
from georss_client.generic_feed import GenericFeed
self._feed = GenericFeed(home_coordinates, url, filter_radius=radius,
filter_categories=None if not category
else [category])
@property
def name(self):
@ -125,115 +123,25 @@ class GeoRssServiceSensor(Entity):
def update(self):
"""Update this sensor from the GeoRSS service."""
_LOGGER.debug("About to update sensor %s", self.entity_id)
self._data.update()
# If no events were found due to an error then just set state to zero.
if self._data.events is None:
self._state = 0
else:
if self._category is None:
# Add all events regardless of category.
my_events = self._data.events
else:
# Only keep events that belong to sensor's category.
my_events = [event for event in self._data.events if
event[ATTR_CATEGORY] == self._category]
import georss_client
status, feed_entries = self._feed.update()
if status == georss_client.UPDATE_OK:
_LOGGER.debug("Adding events to sensor %s: %s", self.entity_id,
my_events)
self._state = len(my_events)
feed_entries)
self._state = len(feed_entries)
# And now compute the attributes from the filtered events.
matrix = {}
for event in my_events:
matrix[event[ATTR_TITLE]] = '{:.0f}km'.format(
event[ATTR_DISTANCE])
for entry in feed_entries:
matrix[entry.title] = '{:.0f}km'.format(
entry.distance_to_home)
self._state_attributes = matrix
class GeoRssServiceData:
"""Provide access to GeoRSS feed and stores the latest data."""
def __init__(self, home_latitude, home_longitude, url, radius_in_km):
"""Initialize the update service."""
self._home_coordinates = [home_latitude, home_longitude]
self._url = url
self._radius_in_km = radius_in_km
self.events = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Retrieve data from GeoRSS feed and store events."""
import feedparser
feed_data = feedparser.parse(self._url)
if not feed_data:
_LOGGER.error("Error fetching feed data from %s", self._url)
elif status == georss_client.UPDATE_OK_NO_DATA:
_LOGGER.debug("Update successful, but no data received from %s",
self._feed)
# Don't change the state or state attributes.
else:
events = self.filter_entries(feed_data)
self.events = events
def filter_entries(self, feed_data):
"""Filter entries by distance from home coordinates."""
events = []
_LOGGER.debug("%s entri(es) available in feed %s",
len(feed_data.entries), self._url)
for entry in feed_data.entries:
geometry = None
if hasattr(entry, 'where'):
geometry = entry.where
elif hasattr(entry, 'geo_lat') and hasattr(entry, 'geo_long'):
coordinates = (float(entry.geo_long), float(entry.geo_lat))
point = namedtuple('Point', ['type', 'coordinates'])
geometry = point('Point', coordinates)
if geometry:
distance = self.calculate_distance_to_geometry(geometry)
if distance <= self._radius_in_km:
event = {
ATTR_CATEGORY: None if not hasattr(
entry, 'category') else entry.category,
ATTR_TITLE: None if not hasattr(
entry, 'title') else entry.title,
ATTR_DISTANCE: distance
}
events.append(event)
_LOGGER.debug("%s events found nearby", len(events))
return events
def calculate_distance_to_geometry(self, geometry):
"""Calculate the distance between HA and provided geometry."""
distance = float("inf")
if geometry.type == 'Point':
distance = self.calculate_distance_to_point(geometry)
elif geometry.type == 'Polygon':
distance = self.calculate_distance_to_polygon(
geometry.coordinates[0])
else:
_LOGGER.warning("Not yet implemented: %s", geometry.type)
return distance
def calculate_distance_to_point(self, point):
"""Calculate the distance between HA and the provided point."""
# Swap coordinates to match: (lat, lon).
coordinates = (point.coordinates[1], point.coordinates[0])
return self.calculate_distance_to_coords(coordinates)
def calculate_distance_to_coords(self, coordinates):
"""Calculate the distance between HA and the provided coordinates."""
# Expecting coordinates in format: (lat, lon).
from haversine import haversine
distance = haversine(coordinates, self._home_coordinates)
_LOGGER.debug("Distance from %s to %s: %s km", self._home_coordinates,
coordinates, distance)
return distance
def calculate_distance_to_polygon(self, polygon):
"""Calculate the distance between HA and the provided polygon."""
distance = float("inf")
# Calculate distance from polygon by calculating the distance
# to each point of the polygon but not to each edge of the
# polygon; should be good enough
for polygon_point in polygon:
coordinates = (polygon_point[1], polygon_point[0])
distance = min(distance,
self.calculate_distance_to_coords(coordinates))
_LOGGER.debug("Distance from %s to %s: %s km", self._home_coordinates,
polygon, distance)
return distance
_LOGGER.warning("Update not successful, no data received from %s",
self._feed)
# If no events were found due to an error then just set state to
# zero.
self._state = 0

7
requirements_all.txt
View File

@ -356,7 +356,6 @@ fastdotcom==0.0.3
fedexdeliverymanager==1.0.6
# homeassistant.components.feedreader
# homeassistant.components.sensor.geo_rss_events
feedparser==5.2.1
# homeassistant.components.sensor.fints
@ -397,6 +396,9 @@ geizhals==0.0.7
# homeassistant.components.geo_location.geo_json_events
geojson_client==0.1
# homeassistant.components.sensor.geo_rss_events
georss_client==0.1
# homeassistant.components.sensor.gitter
gitterpy==0.1.7
@ -433,9 +435,6 @@ habitipy==0.2.0
# homeassistant.components.hangouts
hangups==0.4.5
# homeassistant.components.sensor.geo_rss_events
haversine==0.4.5
# homeassistant.components.mqtt.server
hbmqtt==0.9.4

7
requirements_test_all.txt
View File

@ -57,7 +57,6 @@ ephem==3.7.6.0
evohomeclient==0.2.7
# homeassistant.components.feedreader
# homeassistant.components.sensor.geo_rss_events
feedparser==5.2.1
# homeassistant.components.sensor.foobot
@ -69,15 +68,15 @@ gTTS-token==1.1.2
# homeassistant.components.geo_location.geo_json_events
geojson_client==0.1
# homeassistant.components.sensor.geo_rss_events
georss_client==0.1
# homeassistant.components.ffmpeg
ha-ffmpeg==1.9
# homeassistant.components.hangouts
hangups==0.4.5
# homeassistant.components.sensor.geo_rss_events
haversine==0.4.5
# homeassistant.components.mqtt.server
hbmqtt==0.9.4

1
script/gen_requirements_all.py
View File

@ -51,6 +51,7 @@ TEST_REQUIREMENTS = (
'foobot_async',
'gTTS-token',
'geojson_client',
'georss_client',
'hangups',
'HAP-python',
'ha-ffmpeg',

229
tests/components/sensor/test_geo_rss_events.py
View File

@ -2,26 +2,38 @@
import unittest
from unittest import mock
import sys
from unittest.mock import MagicMock, patch
import feedparser
import pytest
from homeassistant.components import sensor
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, ATTR_FRIENDLY_NAME, \
EVENT_HOMEASSISTANT_START, ATTR_ICON
from homeassistant.setup import setup_component
from tests.common import load_fixture, get_test_home_assistant
from tests.common import get_test_home_assistant, \
assert_setup_component, fire_time_changed
import homeassistant.components.sensor.geo_rss_events as geo_rss_events
import homeassistant.util.dt as dt_util
URL = 'http://geo.rss.local/geo_rss_events.xml'
VALID_CONFIG_WITH_CATEGORIES = {
'platform': 'geo_rss_events',
geo_rss_events.CONF_URL: URL,
geo_rss_events.CONF_CATEGORIES: [
'Category 1',
'Category 2'
sensor.DOMAIN: [
{
'platform': 'geo_rss_events',
geo_rss_events.CONF_URL: URL,
geo_rss_events.CONF_CATEGORIES: [
'Category 1'
]
}
]
}
VALID_CONFIG_WITHOUT_CATEGORIES = {
'platform': 'geo_rss_events',
geo_rss_events.CONF_URL: URL
VALID_CONFIG = {
sensor.DOMAIN: [
{
'platform': 'geo_rss_events',
geo_rss_events.CONF_URL: URL
}
]
}
@ -34,119 +46,114 @@ class TestGeoRssServiceUpdater(unittest.TestCase):
def setUp(self):
"""Initialize values for this testcase class."""
self.hass = get_test_home_assistant()
self.config = VALID_CONFIG_WITHOUT_CATEGORIES
# self.config = VALID_CONFIG_WITHOUT_CATEGORIES
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
@mock.patch('feedparser.parse', return_value=feedparser.parse(""))
def test_setup_with_categories(self, mock_parse):
"""Test the general setup of this sensor."""
self.config = VALID_CONFIG_WITH_CATEGORIES
self.assertTrue(
setup_component(self.hass, 'sensor', {'sensor': self.config}))
self.assertIsNotNone(
self.hass.states.get('sensor.event_service_category_1'))
self.assertIsNotNone(
self.hass.states.get('sensor.event_service_category_2'))
@staticmethod
def _generate_mock_feed_entry(external_id, title, distance_to_home,
coordinates, category):
"""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
return feed_entry
@mock.patch('feedparser.parse', return_value=feedparser.parse(""))
def test_setup_without_categories(self, mock_parse):
"""Test the general setup of this sensor."""
self.assertTrue(
setup_component(self.hass, 'sensor', {'sensor': self.config}))
self.assertIsNotNone(self.hass.states.get('sensor.event_service_any'))
@mock.patch('georss_client.generic_feed.GenericFeed')
def test_setup(self, mock_feed):
"""Test the general setup of the platform."""
# Set up some mock feed entries for this test.
mock_entry_1 = self._generate_mock_feed_entry('1234', 'Title 1', 15.5,
(-31.0, 150.0),
'Category 1')
mock_entry_2 = self._generate_mock_feed_entry('2345', 'Title 2', 20.5,
(-31.1, 150.1),
'Category 1')
mock_feed.return_value.update.return_value = 'OK', [mock_entry_1,
mock_entry_2]
def setup_data(self, url='url'):
"""Set up data object for use by sensors."""
home_latitude = -33.865
home_longitude = 151.209444
radius_in_km = 500
data = geo_rss_events.GeoRssServiceData(home_latitude,
home_longitude, url,
radius_in_km)
return data
utcnow = dt_util.utcnow()
# Patching 'utcnow' to gain more control over the timed update.
with patch('homeassistant.util.dt.utcnow', return_value=utcnow):
with assert_setup_component(1, sensor.DOMAIN):
self.assertTrue(setup_component(self.hass, sensor.DOMAIN,
VALID_CONFIG))
# Artificially trigger update.
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
# Collect events.
self.hass.block_till_done()
def test_update_sensor_with_category(self):
"""Test updating sensor object."""
raw_data = load_fixture('geo_rss_events.xml')
# Loading raw data from fixture and plug in to data object as URL
# works since the third-party feedparser library accepts a URL
# as well as the actual data.
data = self.setup_data(raw_data)
category = "Category 1"
name = "Name 1"
unit_of_measurement = "Unit 1"
sensor = geo_rss_events.GeoRssServiceSensor(category,
data, name,
unit_of_measurement)
all_states = self.hass.states.all()
assert len(all_states) == 1
sensor.update()
assert sensor.name == "Name 1 Category 1"
assert sensor.unit_of_measurement == "Unit 1"
assert sensor.icon == "mdi:alert"
assert len(sensor._data.events) == 4
assert sensor.state == 1
assert sensor.device_state_attributes == {'Title 1': "117km"}
# Check entries of first hit
assert sensor._data.events[0][geo_rss_events.ATTR_TITLE] == "Title 1"
assert sensor._data.events[0][
geo_rss_events.ATTR_CATEGORY] == "Category 1"
self.assertAlmostEqual(sensor._data.events[0][
geo_rss_events.ATTR_DISTANCE], 116.586, 0)
state = self.hass.states.get("sensor.event_service_any")
self.assertIsNotNone(state)
assert state.name == "Event Service Any"
assert int(state.state) == 2
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Any",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
"Title 1": "16km", "Title 2": "20km"}
def test_update_sensor_without_category(self):
"""Test updating sensor object."""
raw_data = load_fixture('geo_rss_events.xml')
data = self.setup_data(raw_data)
category = None
name = "Name 2"
unit_of_measurement = "Unit 2"
sensor = geo_rss_events.GeoRssServiceSensor(category,
data, name,
unit_of_measurement)
# Simulate an update - empty data, but successful update,
# so no changes to entities.
mock_feed.return_value.update.return_value = 'OK_NO_DATA', None
fire_time_changed(self.hass, utcnow +
geo_rss_events.SCAN_INTERVAL)
self.hass.block_till_done()
sensor.update()
assert sensor.name == "Name 2 Any"
assert sensor.unit_of_measurement == "Unit 2"
assert sensor.icon == "mdi:alert"
assert len(sensor._data.events) == 4
assert sensor.state == 4
assert sensor.device_state_attributes == {'Title 1': "117km",
'Title 2': "302km",
'Title 3': "204km",
'Title 6': "48km"}
all_states = self.hass.states.all()
assert len(all_states) == 1
state = self.hass.states.get("sensor.event_service_any")
assert int(state.state) == 2
def test_update_sensor_without_data(self):
"""Test updating sensor object."""
data = self.setup_data()
category = None
name = "Name 3"
unit_of_measurement = "Unit 3"
sensor = geo_rss_events.GeoRssServiceSensor(category,
data, name,
unit_of_measurement)
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = 'ERROR', None
fire_time_changed(self.hass, utcnow +
2 * geo_rss_events.SCAN_INTERVAL)
self.hass.block_till_done()
sensor.update()
assert sensor.name == "Name 3 Any"
assert sensor.unit_of_measurement == "Unit 3"
assert sensor.icon == "mdi:alert"
assert len(sensor._data.events) == 0
assert sensor.state == 0
all_states = self.hass.states.all()
assert len(all_states) == 1
state = self.hass.states.get("sensor.event_service_any")
assert int(state.state) == 0
@mock.patch('feedparser.parse', return_value=None)
def test_update_sensor_with_none_result(self, parse_function):
"""Test updating sensor object."""
data = self.setup_data("http://invalid.url/")
category = None
name = "Name 4"
unit_of_measurement = "Unit 4"
sensor = geo_rss_events.GeoRssServiceSensor(category,
data, name,
unit_of_measurement)
@mock.patch('georss_client.generic_feed.GenericFeed')
def test_setup_with_categories(self, mock_feed):
"""Test the general setup of the platform."""
# Set up some mock feed entries for this test.
mock_entry_1 = self._generate_mock_feed_entry('1234', 'Title 1', 15.5,
(-31.0, 150.0),
'Category 1')
mock_entry_2 = self._generate_mock_feed_entry('2345', 'Title 2', 20.5,
(-31.1, 150.1),
'Category 1')
mock_feed.return_value.update.return_value = 'OK', [mock_entry_1,
mock_entry_2]
sensor.update()
assert sensor.name == "Name 4 Any"
assert sensor.unit_of_measurement == "Unit 4"
assert sensor.state == 0
with assert_setup_component(1, sensor.DOMAIN):
self.assertTrue(setup_component(self.hass, sensor.DOMAIN,
VALID_CONFIG_WITH_CATEGORIES))
# Artificially trigger update.
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
# Collect events.
self.hass.block_till_done()
all_states = self.hass.states.all()
assert len(all_states) == 1
state = self.hass.states.get("sensor.event_service_category_1")
self.assertIsNotNone(state)
assert state.name == "Event Service Category 1"
assert int(state.state) == 2
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Category 1",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
"Title 1": "16km", "Title 2": "20km"}

76
tests/fixtures/geo_rss_events.xml vendored
View File

@ -1,76 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<rss version="2.0" xmlns:georss="http://www.georss.org/georss"
xmlns:geo="http://www.w3.org/2003/01/geo/wgs84_pos#">
<channel>
<!-- Entry within vicinity of home coordinates - Point -->
<item>
<title>Title 1</title>
<description>Description 1</description>
<category>Category 1</category>
<pubDate>Sun, 30 Jul 2017 09:00:00 UTC</pubDate>
<guid>GUID 1</guid>
<georss:point>-32.916667 151.75</georss:point>
</item>
<!-- Entry within vicinity of home coordinates - Point -->
<item>
<title>Title 2</title>
<description>Description 2</description>
<category>Category 2</category>
<pubDate>Sun, 30 Jul 2017 09:05:00 GMT</pubDate>
<guid>GUID 2</guid>
<geo:long>148.601111</geo:long>
<geo:lat>-32.256944</geo:lat>
</item>
<!-- Entry within vicinity of home coordinates - Polygon -->
<item>
<title>Title 3</title>
<description>Description 3</description>
<category>Category 3</category>
<pubDate>Sun, 30 Jul 2017 09:05:00 GMT</pubDate>
<guid>GUID 3</guid>
<georss:polygon>
-33.283333 149.1
-33.2999997 149.1
-33.2999997 149.1166663888889
-33.283333 149.1166663888889
-33.283333 149.1
</georss:polygon>
</item>
<!-- Entry out of vicinity of home coordinates - Point -->
<item>
<title>Title 4</title>
<description>Description 4</description>
<category>Category 4</category>
<pubDate>Sun, 30 Jul 2017 09:15:00 GMT</pubDate>
<guid>GUID 4</guid>
<georss:point>52.518611 13.408333</georss:point>
</item>
<!-- Entry without coordinates -->
<item>
<title>Title 5</title>
<description>Description 5</description>
<category>Category 5</category>
<pubDate>Sun, 30 Jul 2017 09:20:00 GMT</pubDate>
<guid>GUID 5</guid>
</item>
<!-- Entry within vicinity of home coordinates -->
<!-- Link instead of GUID; updated instead of pubDate -->
<item>
<title>Title 6</title>
<description>Description 6</description>
<category>Category 6</category>
<updated>2017-07-30T09:25:00.000Z</updated>
<link>Link 6</link>
<georss:point>-33.75801 150.70544</georss:point>
</item>
<!-- Entry with unsupported geometry - Line -->
<item>
<title>Title 1</title>
<description>Description 1</description>
<category>Category 1</category>
<pubDate>Sun, 30 Jul 2017 09:00:00 UTC</pubDate>
<guid>GUID 1</guid>
<georss:line>45.256 -110.45 46.46 -109.48 43.84 -109.86</georss:line>
</item>
</channel>
</rss>