mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 13:47:35 +00:00
Support for OctoPrint sensors (#1924)
This commit is contained in:
William Scanlon
Paulus Schoutsen
parent
1a59ba735f
commit
6d9254ce25
@ -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
|
||||
|
||||
|
||||
111
homeassistant/components/binary_sensor/octoprint.py
Normal file
111
homeassistant/components/binary_sensor/octoprint.py
Normal file
@ -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)
|
||||
121
homeassistant/components/octoprint.py
Normal file
121
homeassistant/components/octoprint.py
Normal file
@ -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]
|
||||
118
homeassistant/components/sensor/octoprint.py
Normal file
118
homeassistant/components/sensor/octoprint.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user