From 6d9254ce25521dd8dc9fcc561484289a1c0a7fdd Mon Sep 17 00:00:00 2001 From: William Scanlon Date: Tue, 3 May 2016 21:35:11 -0400 Subject: [PATCH] Support for OctoPrint sensors (#1924) --- .coveragerc | 3 + .../components/binary_sensor/octoprint.py | 111 ++++++++++++++++ homeassistant/components/octoprint.py | 121 ++++++++++++++++++ homeassistant/components/sensor/octoprint.py | 118 +++++++++++++++++ 4 files changed, 353 insertions(+) create mode 100644 homeassistant/components/binary_sensor/octoprint.py create mode 100644 homeassistant/components/octoprint.py create mode 100644 homeassistant/components/sensor/octoprint.py diff --git a/.coveragerc b/.coveragerc index 153d0182f62..32cbdc38047 100644 --- a/.coveragerc +++ b/.coveragerc @@ -32,6 +32,9 @@ omit = homeassistant/components/nest.py homeassistant/components/*/nest.py + homeassistant/components/octoprint.py + homeassistant/components/*/octoprint.py + homeassistant/components/rpi_gpio.py homeassistant/components/*/rpi_gpio.py diff --git a/homeassistant/components/binary_sensor/octoprint.py b/homeassistant/components/binary_sensor/octoprint.py new file mode 100644 index 00000000000..803416ce66c --- /dev/null +++ b/homeassistant/components/binary_sensor/octoprint.py @@ -0,0 +1,111 @@ +""" +Support for monitoring OctoPrint binary sensors. + +Uses OctoPrint REST JSON API to query for monitored variables. +For more details about this component, please refer to the documentation at +http://docs.octoprint.org/en/master/api/ +""" + +import logging +import requests + +from homeassistant.const import CONF_NAME, STATE_ON, STATE_OFF +from homeassistant.components.binary_sensor import BinarySensorDevice +from homeassistant.loader import get_component + +DEPENDENCIES = ["octoprint"] + +SENSOR_TYPES = { + # API Endpoint, Group, Key, unit + "Printing": ["printer", "state", "printing", None], + "Printing Error": ["printer", "state", "error", None] +} + +_LOGGER = logging.getLogger(__name__) + + +# pylint: disable=unused-argument +def setup_platform(hass, config, add_devices, discovery_info=None): + """Setup the available OctoPrint binary sensors.""" + octoprint = get_component('octoprint') + name = config.get(CONF_NAME, "OctoPrint") + monitored_conditions = config.get("monitored_conditions", + SENSOR_TYPES.keys()) + + devices = [] + for octo_type in monitored_conditions: + if octo_type in SENSOR_TYPES: + new_sensor = OctoPrintBinarySensor(octoprint.OCTOPRINT, + octo_type, + SENSOR_TYPES[octo_type][2], + name, + SENSOR_TYPES[octo_type][3], + SENSOR_TYPES[octo_type][0], + SENSOR_TYPES[octo_type][1], + "flags") + devices.append(new_sensor) + else: + _LOGGER.error("Unknown OctoPrint sensor type: %s", octo_type) + add_devices(devices) + + +# pylint: disable=too-many-instance-attributes +class OctoPrintBinarySensor(BinarySensorDevice): + """Represents an OctoPrint binary sensor.""" + + # pylint: disable=too-many-arguments + def __init__(self, api, condition, sensor_type, sensor_name, + unit, endpoint, group, tool=None): + """Initialize a new OctoPrint sensor.""" + self.sensor_name = sensor_name + if tool is None: + self._name = sensor_name + ' ' + condition + else: + self._name = sensor_name + ' ' + condition + self.sensor_type = sensor_type + self.api = api + self._state = False + self._unit_of_measurement = unit + self.api_endpoint = endpoint + self.api_group = group + self.api_tool = tool + # Set initial state + self.update() + _LOGGER.debug("created OctoPrint binary sensor %r", self) + + @property + def name(self): + """Return the name of the sensor.""" + return self._name + + @property + def state(self): + """Return the state of the sensor.""" + return self.is_on + + @property + def is_on(self): + """Return true if binary sensor is on.""" + if self._state: + return STATE_ON + else: + return STATE_OFF + + @property + def sensor_class(self): + """Return the class of this sensor, from SENSOR_CLASSES.""" + return None + + def update(self): + """Update state of sensor.""" + try: + self._state = self.api.update(self.sensor_type, + self.api_endpoint, + self.api_group, + self.api_tool) + except requests.exceptions.ConnectionError: + # Error calling the api, already logged in api.update() + return + + if self._state is None: + _LOGGER.warning("unable to locate value for %s", self.sensor_type) diff --git a/homeassistant/components/octoprint.py b/homeassistant/components/octoprint.py new file mode 100644 index 00000000000..9d7f015a2f4 --- /dev/null +++ b/homeassistant/components/octoprint.py @@ -0,0 +1,121 @@ +""" +Support for monitoring OctoPrint 3D printers. + +Uses OctoPrint REST JSON API to query for monitored variables. +http://docs.octoprint.org/en/master/api/ +""" + +import logging + +import time +import requests + +from homeassistant.components import discovery +from homeassistant.const import CONF_API_KEY, CONF_HOST +from homeassistant.helpers import validate_config + +DOMAIN = "octoprint" +OCTOPRINT = None + +_LOGGER = logging.getLogger(__name__) + +DISCOVER_SENSORS = 'octoprint.sensors' +DISCOVER_BINARY_SENSORS = 'octoprint.binary_sensor' + + +def setup(hass, config): + """Set up OctoPrint API.""" + if not validate_config(config, {DOMAIN: [CONF_API_KEY], + DOMAIN: [CONF_HOST]}, + _LOGGER): + return False + + base_url = config[DOMAIN][CONF_HOST] + "/api/" + api_key = config[DOMAIN][CONF_API_KEY] + + global OCTOPRINT + try: + OCTOPRINT = OctoPrintAPI(base_url, api_key) + OCTOPRINT.get("printer") + OCTOPRINT.get("job") + except requests.exceptions.RequestException as conn_err: + _LOGGER.error("Error setting up OctoPrint API: %r", conn_err) + return False + + for component, discovery_service in ( + ('sensor', DISCOVER_SENSORS), + ('binary_sensor', DISCOVER_BINARY_SENSORS)): + discovery.discover(hass, discovery_service, component=component, + hass_config=config) + + return True + + +class OctoPrintAPI(object): + """Simple json wrapper for OctoPrint's api.""" + + def __init__(self, api_url, key): + """Initialize OctoPrint API and set headers needed later.""" + self.api_url = api_url + self.headers = {'content-type': 'application/json', + 'X-Api-Key': key} + self.printer_last_reading = [{}, None] + self.job_last_reading = [{}, None] + + def get_tools(self): + """Get the dynamic list of tools that temperature is monitored on.""" + tools = self.printer_last_reading[0]['temperature'] + return tools.keys() + + def get(self, endpoint): + """Send a get request, and return the response as a dict.""" + now = time.time() + if endpoint == "job": + last_time = self.job_last_reading[1] + if last_time is not None: + if now - last_time < 30.0: + return self.job_last_reading[0] + elif endpoint == "printer": + last_time = self.printer_last_reading[1] + if last_time is not None: + if now - last_time < 30.0: + return self.printer_last_reading[0] + url = self.api_url + endpoint + try: + response = requests.get(url, + headers=self.headers, + timeout=30) + response.raise_for_status() + if endpoint == "job": + self.job_last_reading[0] = response.json() + self.job_last_reading[1] = time.time() + elif endpoint == "printer": + self.printer_last_reading[0] = response.json() + self.printer_last_reading[1] = time.time() + return response.json() + except requests.exceptions.ConnectionError as conn_exc: + _LOGGER.error("Failed to update OctoPrint status. Error: %s", + conn_exc) + raise + + def update(self, sensor_type, end_point, group, tool=None): + """Return the value for sensor_type from the provided endpoint.""" + try: + return get_value_from_json(self.get(end_point), sensor_type, + group, tool) + except requests.exceptions.ConnectionError: + raise + + +# pylint: disable=unused-variable +def get_value_from_json(json_dict, sensor_type, group, tool): + """Return the value for sensor_type from the provided json.""" + if group in json_dict: + if sensor_type in json_dict[group]: + if sensor_type == "target" and json_dict[sensor_type] is None: + return 0 + else: + return json_dict[group][sensor_type] + elif tool is not None: + if sensor_type in json_dict[group][tool]: + return json_dict[group][tool][sensor_type] diff --git a/homeassistant/components/sensor/octoprint.py b/homeassistant/components/sensor/octoprint.py new file mode 100644 index 00000000000..3e47c4d51fe --- /dev/null +++ b/homeassistant/components/sensor/octoprint.py @@ -0,0 +1,118 @@ +""" +Support for monitoring OctoPrint sensors. + +Uses OctoPrint REST JSON API to query for monitored variables. +For more details about this component, please refer to the documentation at +http://docs.octoprint.org/en/master/api/ +""" + +import logging +import requests + +from homeassistant.const import TEMP_CELSIUS, CONF_NAME +from homeassistant.helpers.entity import Entity +from homeassistant.loader import get_component + +DEPENDENCIES = ["octoprint"] + +SENSOR_TYPES = { + # API Endpoint, Group, Key, unit + "Temperatures": ["printer", "temperature", "*", TEMP_CELSIUS], + "Current State": ["printer", "state", "text", None], + "Job Percentage": ["job", "progress", "completion", "%"], +} + +_LOGGER = logging.getLogger(__name__) + + +# pylint: disable=unused-argument +def setup_platform(hass, config, add_devices, discovery_info=None): + """Setup the available OctoPrint sensors.""" + octoprint = get_component('octoprint') + name = config.get(CONF_NAME, "OctoPrint") + monitored_conditions = config.get("monitored_conditions", + SENSOR_TYPES.keys()) + + devices = [] + types = ["actual", "target"] + for octo_type in monitored_conditions: + if octo_type == "Temperatures": + for tool in octoprint.OCTOPRINT.get_tools(): + for temp_type in types: + new_sensor = OctoPrintSensor(octoprint.OCTOPRINT, + temp_type, + temp_type, + name, + SENSOR_TYPES[octo_type][3], + SENSOR_TYPES[octo_type][0], + SENSOR_TYPES[octo_type][1], + tool) + devices.append(new_sensor) + elif octo_type in SENSOR_TYPES: + new_sensor = OctoPrintSensor(octoprint.OCTOPRINT, + octo_type, + SENSOR_TYPES[octo_type][2], + name, + SENSOR_TYPES[octo_type][3], + SENSOR_TYPES[octo_type][0], + SENSOR_TYPES[octo_type][1]) + devices.append(new_sensor) + else: + _LOGGER.error("Unknown OctoPrint sensor type: %s", octo_type) + + add_devices(devices) + + +# pylint: disable=too-many-instance-attributes +class OctoPrintSensor(Entity): + """Represents an OctoPrint sensor.""" + + # pylint: disable=too-many-arguments + def __init__(self, api, condition, sensor_type, sensor_name, + unit, endpoint, group, tool=None): + """Initialize a new OctoPrint sensor.""" + self.sensor_name = sensor_name + if tool is None: + self._name = sensor_name + ' ' + condition + else: + self._name = sensor_name + ' ' + condition + ' ' + tool + ' temp' + self.sensor_type = sensor_type + self.api = api + self._state = None + self._unit_of_measurement = unit + self.api_endpoint = endpoint + self.api_group = group + self.api_tool = tool + # Set initial state + self.update() + _LOGGER.debug("created OctoPrint sensor %r", self) + + @property + def name(self): + """Return the name of the sensor.""" + return self._name + + @property + def state(self): + """Return the state of the sensor.""" + return self._state + + @property + def unit_of_measurement(self): + """Unit of measurement of this entity, if any.""" + return self._unit_of_measurement + + def update(self): + """Update state of sensor.""" + try: + self._state = self.api.update(self.sensor_type, + self.api_endpoint, + self.api_group, + self.api_tool) + except requests.exceptions.ConnectionError: + # Error calling the api, already logged in api.update() + return + + if self._state is None: + _LOGGER.warning("unable to locate value for %s", self.sensor_type) + return