6c6a5c50a5
* Fix digest auth rest sensors * Refactor to use request() * Fix black complaints * Don't rename data variable Don't rename the data variable, appears several other sensors have been coded to rely on it * Fix tests test_setup_missing_schema: Change the exception to check for to PlatformNotReady. With the change away from prepared statements, we no longer get the MissingSchema error during setup - we get it when we attempt to call the endpoint. Since the result is PlatformNotReady, which is what we want, and the error log clearly contains a note that the schema is incorrect I think this is fine. test_setup_failed_connect & test_setup_timeout: These aren't checking for minimum config, and they're not invoking the correct method to have the default config filled in. Therefore I've just added the correct minimum config so these can continue to test what they're there to test. test_update_request_exception: Testing a request exception with the assert on line 404! Excellent. Anyway, we've moved from using the requests Session object to the requests.request function - so the patch needed to be altered to ensure the RequestException was raised. * Remove no longer needed import * Fix binary sensor test in same way
259 lines
8.1 KiB
Python
259 lines
8.1 KiB
Python
"""Support for RESTful API sensors."""
|
|
import logging
|
|
import json
|
|
|
|
import voluptuous as vol
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
|
|
|
|
from homeassistant.components.sensor import PLATFORM_SCHEMA, DEVICE_CLASSES_SCHEMA
|
|
from homeassistant.const import (
|
|
CONF_AUTHENTICATION,
|
|
CONF_FORCE_UPDATE,
|
|
CONF_HEADERS,
|
|
CONF_NAME,
|
|
CONF_METHOD,
|
|
CONF_PASSWORD,
|
|
CONF_PAYLOAD,
|
|
CONF_RESOURCE,
|
|
CONF_RESOURCE_TEMPLATE,
|
|
CONF_UNIT_OF_MEASUREMENT,
|
|
CONF_USERNAME,
|
|
CONF_TIMEOUT,
|
|
CONF_VALUE_TEMPLATE,
|
|
CONF_VERIFY_SSL,
|
|
CONF_DEVICE_CLASS,
|
|
HTTP_BASIC_AUTHENTICATION,
|
|
HTTP_DIGEST_AUTHENTICATION,
|
|
)
|
|
from homeassistant.exceptions import PlatformNotReady
|
|
from homeassistant.helpers.entity import Entity
|
|
import homeassistant.helpers.config_validation as cv
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEFAULT_METHOD = "GET"
|
|
DEFAULT_NAME = "REST Sensor"
|
|
DEFAULT_VERIFY_SSL = True
|
|
DEFAULT_FORCE_UPDATE = False
|
|
DEFAULT_TIMEOUT = 10
|
|
|
|
CONF_JSON_ATTRS = "json_attributes"
|
|
METHODS = ["POST", "GET"]
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|
{
|
|
vol.Exclusive(CONF_RESOURCE, CONF_RESOURCE): cv.url,
|
|
vol.Exclusive(CONF_RESOURCE_TEMPLATE, CONF_RESOURCE): cv.template,
|
|
vol.Optional(CONF_AUTHENTICATION): vol.In(
|
|
[HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]
|
|
),
|
|
vol.Optional(CONF_HEADERS): vol.Schema({cv.string: cv.string}),
|
|
vol.Optional(CONF_JSON_ATTRS, default=[]): cv.ensure_list_csv,
|
|
vol.Optional(CONF_METHOD, default=DEFAULT_METHOD): vol.In(METHODS),
|
|
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
|
vol.Optional(CONF_PASSWORD): cv.string,
|
|
vol.Optional(CONF_PAYLOAD): cv.string,
|
|
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
|
|
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
|
|
vol.Optional(CONF_USERNAME): cv.string,
|
|
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): cv.boolean,
|
|
vol.Optional(CONF_FORCE_UPDATE, default=DEFAULT_FORCE_UPDATE): cv.boolean,
|
|
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
|
|
}
|
|
)
|
|
|
|
PLATFORM_SCHEMA = vol.All(
|
|
cv.has_at_least_one_key(CONF_RESOURCE, CONF_RESOURCE_TEMPLATE), PLATFORM_SCHEMA
|
|
)
|
|
|
|
|
|
def setup_platform(hass, config, add_entities, discovery_info=None):
|
|
"""Set up the RESTful sensor."""
|
|
name = config.get(CONF_NAME)
|
|
resource = config.get(CONF_RESOURCE)
|
|
resource_template = config.get(CONF_RESOURCE_TEMPLATE)
|
|
method = config.get(CONF_METHOD)
|
|
payload = config.get(CONF_PAYLOAD)
|
|
verify_ssl = config.get(CONF_VERIFY_SSL)
|
|
username = config.get(CONF_USERNAME)
|
|
password = config.get(CONF_PASSWORD)
|
|
headers = config.get(CONF_HEADERS)
|
|
unit = config.get(CONF_UNIT_OF_MEASUREMENT)
|
|
device_class = config.get(CONF_DEVICE_CLASS)
|
|
value_template = config.get(CONF_VALUE_TEMPLATE)
|
|
json_attrs = config.get(CONF_JSON_ATTRS)
|
|
force_update = config.get(CONF_FORCE_UPDATE)
|
|
timeout = config.get(CONF_TIMEOUT)
|
|
|
|
if value_template is not None:
|
|
value_template.hass = hass
|
|
|
|
if resource_template is not None:
|
|
resource_template.hass = hass
|
|
resource = resource_template.render()
|
|
|
|
if username and password:
|
|
if config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION:
|
|
auth = HTTPDigestAuth(username, password)
|
|
else:
|
|
auth = HTTPBasicAuth(username, password)
|
|
else:
|
|
auth = None
|
|
rest = RestData(method, resource, auth, headers, payload, verify_ssl, timeout)
|
|
rest.update()
|
|
if rest.data is None:
|
|
raise PlatformNotReady
|
|
|
|
# Must update the sensor now (including fetching the rest resource) to
|
|
# ensure it's updating its state.
|
|
add_entities(
|
|
[
|
|
RestSensor(
|
|
hass,
|
|
rest,
|
|
name,
|
|
unit,
|
|
device_class,
|
|
value_template,
|
|
json_attrs,
|
|
force_update,
|
|
resource_template,
|
|
)
|
|
],
|
|
True,
|
|
)
|
|
|
|
|
|
class RestSensor(Entity):
|
|
"""Implementation of a REST sensor."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass,
|
|
rest,
|
|
name,
|
|
unit_of_measurement,
|
|
device_class,
|
|
value_template,
|
|
json_attrs,
|
|
force_update,
|
|
resource_template,
|
|
):
|
|
"""Initialize the REST sensor."""
|
|
self._hass = hass
|
|
self.rest = rest
|
|
self._name = name
|
|
self._state = None
|
|
self._unit_of_measurement = unit_of_measurement
|
|
self._device_class = device_class
|
|
self._value_template = value_template
|
|
self._json_attrs = json_attrs
|
|
self._attributes = None
|
|
self._force_update = force_update
|
|
self._resource_template = resource_template
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the sensor."""
|
|
return self._name
|
|
|
|
@property
|
|
def unit_of_measurement(self):
|
|
"""Return the unit the value is expressed in."""
|
|
return self._unit_of_measurement
|
|
|
|
@property
|
|
def device_class(self):
|
|
"""Return the class of this sensor."""
|
|
return self._device_class
|
|
|
|
@property
|
|
def available(self):
|
|
"""Return if the sensor data are available."""
|
|
return self.rest.data is not None
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the device."""
|
|
return self._state
|
|
|
|
@property
|
|
def force_update(self):
|
|
"""Force update."""
|
|
return self._force_update
|
|
|
|
def update(self):
|
|
"""Get the latest data from REST API and update the state."""
|
|
if self._resource_template is not None:
|
|
self.rest.set_url(self._resource_template.render())
|
|
|
|
self.rest.update()
|
|
value = self.rest.data
|
|
|
|
if self._json_attrs:
|
|
self._attributes = {}
|
|
if value:
|
|
try:
|
|
json_dict = json.loads(value)
|
|
if isinstance(json_dict, dict):
|
|
attrs = {
|
|
k: json_dict[k] for k in self._json_attrs if k in json_dict
|
|
}
|
|
self._attributes = attrs
|
|
else:
|
|
_LOGGER.warning("JSON result was not a dictionary")
|
|
except ValueError:
|
|
_LOGGER.warning("REST result could not be parsed as JSON")
|
|
_LOGGER.debug("Erroneous JSON: %s", value)
|
|
else:
|
|
_LOGGER.warning("Empty reply found when expecting JSON data")
|
|
if value is not None and self._value_template is not None:
|
|
value = self._value_template.render_with_possible_json_value(value, None)
|
|
|
|
self._state = value
|
|
|
|
@property
|
|
def device_state_attributes(self):
|
|
"""Return the state attributes."""
|
|
return self._attributes
|
|
|
|
|
|
class RestData:
|
|
"""Class for handling the data retrieval."""
|
|
|
|
def __init__(
|
|
self, method, resource, auth, headers, data, verify_ssl, timeout=DEFAULT_TIMEOUT
|
|
):
|
|
"""Initialize the data object."""
|
|
self._method = method
|
|
self._resource = resource
|
|
self._auth = auth
|
|
self._headers = headers
|
|
self._request_data = data
|
|
self._verify_ssl = verify_ssl
|
|
self._timeout = timeout
|
|
self.data = None
|
|
|
|
def set_url(self, url):
|
|
"""Set url."""
|
|
self._resource = url
|
|
|
|
def update(self):
|
|
"""Get the latest data from REST service with provided method."""
|
|
_LOGGER.debug("Updating from %s", self._resource)
|
|
try:
|
|
response = requests.request(
|
|
self._method,
|
|
self._resource,
|
|
headers=self._headers,
|
|
auth=self._auth,
|
|
data=self._request_data,
|
|
timeout=self._timeout,
|
|
verify=self._verify_ssl,
|
|
)
|
|
self.data = response.text
|
|
except requests.exceptions.RequestException as ex:
|
|
_LOGGER.error("Error fetching data: %s failed with %s", self._resource, ex)
|
|
self.data = None
|