jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 21:27:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify Rest sensors

Browse Source
This commit is contained in:
Paulus Schoutsen 2016-01-02 13:29:33 -08:00
parent 6d35bdafee
commit 4a421e25b0
2 changed files with 38 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
homeassistant/components/binary_sensor/rest.py
View File

@ -6,12 +6,11 @@ The rest binary sensor will consume responses sent by an exposed REST API.
For more details about this platform, please refer to the documentation at For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.rest/ https://home-assistant.io/components/binary_sensor.rest/
""" """
from datetime import timedelta
import logging import logging
import requests
from homeassistant.const import CONF_VALUE_TEMPLATE from homeassistant.const import CONF_VALUE_TEMPLATE
from homeassistant.util import template, Throttle from homeassistant.util import template
from homeassistant.components.sensor.rest import RestData
from homeassistant.components.binary_sensor import BinarySensorDevice from homeassistant.components.binary_sensor import BinarySensorDevice
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -19,60 +18,33 @@ _LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'REST Binary Sensor' DEFAULT_NAME = 'REST Binary Sensor'
DEFAULT_METHOD = 'GET' DEFAULT_METHOD = 'GET'
# Return cached results if last scan was less then this time ago
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
# pylint: disable=unused-variable # pylint: disable=unused-variable
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
""" Get the REST binary sensor. """ """Setup REST binary sensors."""
use_get = False
use_post = False
resource = config.get('resource', None) resource = config.get('resource', None)
method = config.get('method', DEFAULT_METHOD) method = config.get('method', DEFAULT_METHOD)
payload = config.get('payload', None) payload = config.get('payload', None)
verify_ssl = config.get('verify_ssl', True) verify_ssl = config.get('verify_ssl', True)
if method == 'GET': rest = RestData(method, resource, payload, verify_ssl)
use_get = True rest.update()
elif method == 'POST':
use_post = True
try: if rest.data is None:
if use_get: _LOGGER.error('Unable to fetch Rest data')
response = requests.get(resource, timeout=10, verify=verify_ssl)
elif use_post:
response = requests.post(resource, data=payload, timeout=10,
verify=verify_ssl)
if not response.ok:
_LOGGER.error("Response status is '%s'", response.status_code)
return False
except requests.exceptions.MissingSchema:
_LOGGER.error("Missing resource or schema in configuration. "
"Add http:// or https:// to your URL")
return False
except requests.exceptions.ConnectionError:
_LOGGER.error('No route to resource/endpoint: %s', resource)
return False return False
if use_get: add_devices([RestBinarySensor(
rest = RestDataGet(resource, verify_ssl) hass, rest, config.get('name', DEFAULT_NAME),
elif use_post: config.get(CONF_VALUE_TEMPLATE))])
rest = RestDataPost(resource, payload, verify_ssl)
add_devices([RestBinarySensor(hass,
rest,
config.get('name', DEFAULT_NAME),
config.get(CONF_VALUE_TEMPLATE))])
# pylint: disable=too-many-arguments # pylint: disable=too-many-arguments
class RestBinarySensor(BinarySensorDevice): class RestBinarySensor(BinarySensorDevice):
""" Implements a REST binary sensor. """ """REST binary sensor."""
def __init__(self, hass, rest, name, value_template): def __init__(self, hass, rest, name, value_template):
"""Initialize a REST binary sensor."""
self._hass = hass self._hass = hass
self.rest = rest self.rest = rest
self._name = name self._name = name
@ -82,63 +54,20 @@ class RestBinarySensor(BinarySensorDevice):
@property @property
def name(self): def name(self):
""" The name of the binary sensor. """ """Name of the binary sensor."""
return self._name return self._name
@property @property
def is_on(self): def is_on(self):
""" True if the binary sensor is on. """ """Return if the binary sensor is on."""
if self.rest.data is False: if self.rest.data is None:
return False return False
else:
if self._value_template is not None: if self._value_template is not None:
self.rest.data = template.render_with_possible_json_value( self.rest.data = template.render_with_possible_json_value(
self._hass, self._value_template, self.rest.data, False) self._hass, self._value_template, self.rest.data, False)
return bool(int(self.rest.data)) return bool(int(self.rest.data))
def update(self): def update(self):
""" Gets the latest data from REST API and updates the state. """ """Get the latest data from REST API and updates the state."""
self.rest.update() self.rest.update()
# pylint: disable=too-few-public-methods
class RestDataGet(object):
""" Class for handling the data retrieval with GET method. """
def __init__(self, resource, verify_ssl):
self._resource = resource
self._verify_ssl = verify_ssl
self.data = False
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
""" Gets the latest data from REST service with GET method. """
try:
response = requests.get(self._resource, timeout=10,
verify=self._verify_ssl)
self.data = response.text
except requests.exceptions.ConnectionError:
_LOGGER.error("No route to resource/endpoint: %s", self._resource)
self.data = False
# pylint: disable=too-few-public-methods
class RestDataPost(object):
""" Class for handling the data retrieval with POST method. """
def __init__(self, resource, payload, verify_ssl):
self._resource = resource
self._payload = payload
self._verify_ssl = verify_ssl
self.data = False
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
""" Gets the latest data from REST service with POST method. """
try:
response = requests.post(self._resource, data=self._payload,
timeout=10, verify=self._verify_ssl)
self.data = response.text
except requests.exceptions.ConnectionError:
_LOGGER.error("No route to resource/endpoint: %s", self._resource)
self.data = False

80
homeassistant/components/sensor/rest.py
View File

@ -26,47 +26,21 @@ MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
# pylint: disable=unused-variable # pylint: disable=unused-variable
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
""" Get the REST sensor. """ """ Get the REST sensor. """
use_get = False
use_post = False
resource = config.get('resource', None) resource = config.get('resource', None)
method = config.get('method', DEFAULT_METHOD) method = config.get('method', DEFAULT_METHOD)
payload = config.get('payload', None) payload = config.get('payload', None)
verify_ssl = config.get('verify_ssl', True) verify_ssl = config.get('verify_ssl', True)
if method == 'GET': rest = RestData(method, resource, payload, verify_ssl)
use_get = True rest.update()
elif method == 'POST':
use_post = True
try: if rest.data is None:
if use_get: _LOGGER.error('Unable to fetch Rest data')
response = requests.get(resource, timeout=10, verify=verify_ssl)
elif use_post:
response = requests.post(resource, data=payload, timeout=10,
verify=verify_ssl)
if not response.ok:
_LOGGER.error("Response status is '%s'", response.status_code)
return False
except requests.exceptions.MissingSchema:
_LOGGER.error("Missing resource or schema in configuration. "
"Add http:// or https:// to your URL")
return False
except requests.exceptions.ConnectionError:
_LOGGER.error("No route to resource/endpoint: %s", resource)
return False return False
if use_get: add_devices([RestSensor(
rest = RestDataGet(resource, verify_ssl) hass, rest, config.get('name', DEFAULT_NAME),
elif use_post: config.get('unit_of_measurement'), config.get(CONF_VALUE_TEMPLATE))])
rest = RestDataPost(resource, payload, verify_ssl)
add_devices([RestSensor(hass,
rest,
config.get('name', DEFAULT_NAME),
config.get('unit_of_measurement'),
config.get(CONF_VALUE_TEMPLATE))])
# pylint: disable=too-many-arguments # pylint: disable=too-many-arguments
@ -112,11 +86,11 @@ class RestSensor(Entity):
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods
class RestDataGet(object): class RestData(object):
""" Class for handling the data retrieval with GET method. """ """Class for handling the data retrieval."""
def __init__(self, resource, verify_ssl): def __init__(self, method, resource, data, verify_ssl):
self._resource = resource self._request = requests.Request(method, resource, data=data).prepare()
self._verify_ssl = verify_ssl self._verify_ssl = verify_ssl
self.data = None self.data = None
@ -124,31 +98,11 @@ class RestDataGet(object):
def update(self): def update(self):
""" Gets the latest data from REST service with GET method. """ """ Gets the latest data from REST service with GET method. """
try: try:
response = requests.get(self._resource, timeout=10, with requests.Session() as sess:
verify=self._verify_ssl) response = sess.send(self._request, timeout=10,
verify=self._verify_ssl)
self.data = response.text self.data = response.text
except requests.exceptions.ConnectionError: except requests.exceptions.RequestException:
_LOGGER.error("No route to resource/endpoint: %s", self._resource) _LOGGER.error("Error fetching data: %s", self._request)
self.data = None
# pylint: disable=too-few-public-methods
class RestDataPost(object):
""" Class for handling the data retrieval with POST method. """
def __init__(self, resource, payload, verify_ssl):
self._resource = resource
self._payload = payload
self._verify_ssl = verify_ssl
self.data = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
""" Gets the latest data from REST service with POST method. """
try:
response = requests.post(self._resource, data=self._payload,
timeout=10, verify=self._verify_ssl)
self.data = response.text
except requests.exceptions.ConnectionError:
_LOGGER.error("No route to resource/endpoint: %s", self._resource)
self.data = None self.data = None