* Add srp_energy

* Update message on TypeError. Add check for None state.

* Add check for none in history

* Add srpenergy to Test requirements.

* Add srpenergy to requirments.

* Change = to ==.

* Change import for srpenergy

* Fix Flak8 errors

* Add srp to gen requirements script

* Change config name.

* Add daily usage test

* Add test for daily usage.

* Fix Flake8 message.

* Remove blank after docstring.

* Add srpenergy to coverage

* Bump requires version to srpenergy

* Fix type in coverage. Import from Sensor. Use dict.

* Update to 1.0.5. Check credentials on setup. Standalone test.

* Fix formating.

* Remove period. Rename _ variables.

* Fix rebase merge

* Add rebase requirement

* Improve Mock Patching.
pull/18326/head
Brig Lamoreaux 2018-11-08 11:19:30 -07:00 committed by Martin Hjelmare
parent 8f107c46fe
commit 05eac915d1
6 changed files with 219 additions and 0 deletions

View File

@ -800,6 +800,7 @@ omit =
homeassistant/components/sensor/swiss_public_transport.py
homeassistant/components/sensor/syncthru.py
homeassistant/components/sensor/synologydsm.py
homeassistant/components/sensor/srp_energy.py
homeassistant/components/sensor/systemmonitor.py
homeassistant/components/sensor/sytadin.py
homeassistant/components/sensor/tank_utility.py

View File

@ -0,0 +1,149 @@
"""
Platform for retrieving energy data from SRP.
For more details about this platform, please refer to the documentation
https://home-assistant.io/components/sensor.srp_energy/
"""
from datetime import datetime, timedelta
import logging
from requests.exceptions import (
ConnectionError as ConnectError, HTTPError, Timeout)
import voluptuous as vol
from homeassistant.const import (
CONF_NAME, CONF_PASSWORD,
CONF_USERNAME, CONF_ID)
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.helpers.entity import Entity
REQUIREMENTS = ['srpenergy==1.0.5']
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Powered by SRP Energy"
DEFAULT_NAME = 'SRP Energy'
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=1440)
ENERGY_KWH = 'kWh'
ATTR_READING_COST = "reading_cost"
ATTR_READING_TIME = 'datetime'
ATTR_READING_USAGE = 'reading_usage'
ATTR_DAILY_USAGE = 'daily_usage'
ATTR_USAGE_HISTORY = 'usage_history'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_ID): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the SRP energy."""
name = config[CONF_NAME]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
account_id = config[CONF_ID]
from srpenergy.client import SrpEnergyClient
srp_client = SrpEnergyClient(account_id, username, password)
if not srp_client.validate():
_LOGGER.error("Couldn't connect to %s. Check credentials", name)
return
add_entities([SrpEnergy(name, srp_client)], True)
class SrpEnergy(Entity):
"""Representation of an srp usage."""
def __init__(self, name, client):
"""Initialize SRP Usage."""
self._state = None
self._name = name
self._client = client
self._history = None
self._usage = None
@property
def attribution(self):
"""Return the attribution."""
return ATTRIBUTION
@property
def state(self):
"""Return the current state."""
if self._state is None:
return None
return "{0:.2f}".format(self._state)
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return ENERGY_KWH
@property
def history(self):
"""Return the energy usage history of this entity, if any."""
if self._usage is None:
return None
history = [{
ATTR_READING_TIME: isodate,
ATTR_READING_USAGE: kwh,
ATTR_READING_COST: cost
} for _, _, isodate, kwh, cost in self._usage]
return history
@property
def device_state_attributes(self):
"""Return the state attributes."""
attributes = {
ATTR_USAGE_HISTORY: self.history
}
return attributes
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Get the latest usage from SRP Energy."""
start_date = datetime.now() + timedelta(days=-1)
end_date = datetime.now()
try:
usage = self._client.usage(start_date, end_date)
daily_usage = 0.0
for _, _, _, kwh, _ in usage:
daily_usage += float(kwh)
if usage:
self._state = daily_usage
self._usage = usage
else:
_LOGGER.error("Unable to fetch data from SRP. No data")
except (ConnectError, HTTPError, Timeout) as error:
_LOGGER.error("Unable to connect to SRP. %s", error)
except ValueError as error:
_LOGGER.error("Value error connecting to SRP. %s", error)
except TypeError as error:
_LOGGER.error("Type error connecting to SRP. "
"Check username and password. %s", error)

View File

@ -1447,6 +1447,9 @@ spotipy-homeassistant==2.4.4.dev1
# homeassistant.components.sensor.sql
sqlalchemy==1.2.13
# homeassistant.components.sensor.srp_energy
srpenergy==1.0.5
# homeassistant.components.sensor.starlingbank
starlingbank==1.2

View File

@ -238,6 +238,9 @@ somecomfort==0.5.2
# homeassistant.components.sensor.sql
sqlalchemy==1.2.13
# homeassistant.components.sensor.srp_energy
srpenergy==1.0.5
# homeassistant.components.statsd
statsd==3.2.1

View File

@ -104,6 +104,7 @@ TEST_REQUIREMENTS = (
'smhi-pkg',
'somecomfort',
'sqlalchemy',
'srpenergy',
'statsd',
'uvcclient',
'warrant',

View File

@ -0,0 +1,62 @@
"""The tests for the Srp Energy Platform."""
from unittest.mock import patch
import logging
from homeassistant.setup import async_setup_component
_LOGGER = logging.getLogger(__name__)
VALID_CONFIG_MINIMAL = {
'sensor': {
'platform': 'srp_energy',
'username': 'foo',
'password': 'bar',
'id': 1234
}
}
PATCH_INIT = 'srpenergy.client.SrpEnergyClient.__init__'
PATCH_VALIDATE = 'srpenergy.client.SrpEnergyClient.validate'
PATCH_USAGE = 'srpenergy.client.SrpEnergyClient.usage'
def mock_usage(self, startdate, enddate): # pylint: disable=invalid-name
"""Mock srpusage usage."""
_LOGGER.log(logging.INFO, "Calling mock usage")
usage = [
('9/19/2018', '12:00 AM', '2018-09-19T00:00:00-7:00', '1.2', '0.17'),
('9/19/2018', '1:00 AM', '2018-09-19T01:00:00-7:00', '2.1', '0.30'),
('9/19/2018', '2:00 AM', '2018-09-19T02:00:00-7:00', '1.5', '0.23'),
('9/19/2018', '9:00 PM', '2018-09-19T21:00:00-7:00', '1.2', '0.19'),
('9/19/2018', '10:00 PM', '2018-09-19T22:00:00-7:00', '1.1', '0.18'),
('9/19/2018', '11:00 PM', '2018-09-19T23:00:00-7:00', '0.4', '0.09')
]
return usage
async def test_setup_with_config(hass):
"""Test the platform setup with configuration."""
with patch(PATCH_INIT, return_value=None), \
patch(PATCH_VALIDATE, return_value=True), \
patch(PATCH_USAGE, new=mock_usage):
await async_setup_component(hass, 'sensor', VALID_CONFIG_MINIMAL)
state = hass.states.get('sensor.srp_energy')
assert state is not None
async def test_daily_usage(hass):
"""Test the platform daily usage."""
with patch(PATCH_INIT, return_value=None), \
patch(PATCH_VALIDATE, return_value=True), \
patch(PATCH_USAGE, new=mock_usage):
await async_setup_component(hass, 'sensor', VALID_CONFIG_MINIMAL)
state = hass.states.get('sensor.srp_energy')
assert state
assert state.state == '7.50'
assert state.attributes
assert state.attributes.get('unit_of_measurement')