jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 12:47:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Octoprint custom path (#20302)

Browse Source 
* Added custom path config option to octoprint component

* Added some debug logging

* removed debug logging for base url

* Fixed single/double quotes style
This commit is contained in:
Fabien Piuzzi 2019-01-23 12:50:17 +01:00 committed by Martin Hjelmare
parent eaa9c4d437
commit 07f1e2ce75
1 changed files with 31 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
homeassistant/components/octoprint.py
View File

@ -13,8 +13,8 @@ from aiohttp.hdrs import CONTENT_TYPE
from homeassistant.components.discovery import SERVICE_OCTOPRINT from homeassistant.components.discovery import SERVICE_OCTOPRINT
from homeassistant.const import ( from homeassistant.const import (
CONF_API_KEY, CONF_HOST, CONTENT_TYPE_JSON, CONF_NAME, CONF_PORT, CONF_API_KEY, CONF_HOST, CONTENT_TYPE_JSON, CONF_NAME, CONF_PATH,
CONF_SSL, TEMP_CELSIUS, CONF_MONITORED_CONDITIONS, CONF_SENSORS, CONF_PORT, CONF_SSL, TEMP_CELSIUS, CONF_MONITORED_CONDITIONS, CONF_SENSORS,
CONF_BINARY_SENSORS) CONF_BINARY_SENSORS)
from homeassistant.helpers import discovery from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
@ -36,10 +36,20 @@ def has_all_unique_names(value):
return value return value
def ensure_valid_path(value):
"""Validate the path, ensuring it starts and ends with a /."""
vol.Schema(cv.string)(value)
if value[0] != '/':
value = '/' + value
if value[-1] != '/':
value += '/'
return value
BINARY_SENSOR_TYPES = { BINARY_SENSOR_TYPES = {
# API Endpoint, Group, Key, unit # API Endpoint, Group, Key, unit
'Printing': ['printer', 'state', 'printing', None], 'Printing': ['printer', 'state', 'printing', None],
'Printing Error': ['printer', 'state', 'error', None] "Printing Error": ['printer', 'state', 'error', None]
} }
BINARY_SENSOR_SCHEMA = vol.Schema({ BINARY_SENSOR_SCHEMA = vol.Schema({
@ -51,12 +61,12 @@ BINARY_SENSOR_SCHEMA = vol.Schema({
SENSOR_TYPES = { SENSOR_TYPES = {
# API Endpoint, Group, Key, unit, icon # API Endpoint, Group, Key, unit, icon
'Temperatures': ['printer', 'temperature', '*', TEMP_CELSIUS], 'Temperatures': ['printer', 'temperature', '*', TEMP_CELSIUS],
'Current State': ['printer', 'state', 'text', None, 'mdi:printer-3d'], "Current State": ['printer', 'state', 'text', None, 'mdi:printer-3d'],
'Job Percentage': ['job', 'progress', 'completion', '%', "Job Percentage": ['job', 'progress', 'completion', '%',
'mdi:file-percent'], 'mdi:file-percent'],
'Time Remaining': ['job', 'progress', 'printTimeLeft', 'seconds', "Time Remaining": ['job', 'progress', 'printTimeLeft', 'seconds',
'mdi:clock-end'], 'mdi:clock-end'],
'Time Elapsed': ['job', 'progress', 'printTime', 'seconds', "Time Elapsed": ['job', 'progress', 'printTime', 'seconds',
'mdi:clock-start'], 'mdi:clock-start'],
} }
@ -72,6 +82,7 @@ CONFIG_SCHEMA = vol.Schema({
vol.Required(CONF_HOST): cv.string, vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_SSL, default=False): cv.boolean, vol.Optional(CONF_SSL, default=False): cv.boolean,
vol.Optional(CONF_PORT, default=80): cv.port, vol.Optional(CONF_PORT, default=80): cv.port,
vol.Optional(CONF_PATH, default='/'): ensure_valid_path,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_NUMBER_OF_TOOLS, default=0): cv.positive_int, vol.Optional(CONF_NUMBER_OF_TOOLS, default=0): cv.positive_int,
vol.Optional(CONF_BED, default=False): cv.boolean, vol.Optional(CONF_BED, default=False): cv.boolean,
@ -88,7 +99,7 @@ def setup(hass, config):
def device_discovered(service, info): def device_discovered(service, info):
"""Get called when an Octoprint server has been discovered.""" """Get called when an Octoprint server has been discovered."""
_LOGGER.debug('Found an Octoprint server: %s', info) _LOGGER.debug("Found an Octoprint server: %s", info)
discovery.listen(hass, SERVICE_OCTOPRINT, device_discovered) discovery.listen(hass, SERVICE_OCTOPRINT, device_discovered)
@ -99,9 +110,10 @@ def setup(hass, config):
for printer in config[DOMAIN]: for printer in config[DOMAIN]:
name = printer[CONF_NAME] name = printer[CONF_NAME]
ssl = 's' if printer[CONF_SSL] else '' ssl = 's' if printer[CONF_SSL] else ''
base_url = 'http{}://{}:{}/api/'.format(ssl, base_url = 'http{}://{}:{}{}api/'.format(ssl,
printer[CONF_HOST], printer[CONF_HOST],
printer[CONF_PORT]) printer[CONF_PORT],
printer[CONF_PATH])
api_key = printer[CONF_API_KEY] api_key = printer[CONF_API_KEY]
number_of_tools = printer[CONF_NUMBER_OF_TOOLS] number_of_tools = printer[CONF_NUMBER_OF_TOOLS]
bed = printer[CONF_BED] bed = printer[CONF_BED]
@ -154,7 +166,7 @@ class OctoPrintAPI:
tools = [] tools = []
if self.number_of_tools > 0: if self.number_of_tools > 0:
for tool_number in range(0, self.number_of_tools): for tool_number in range(0, self.number_of_tools):
tools.append("tool" + str(tool_number)) tools.append('tool' + str(tool_number))
if self.bed: if self.bed:
tools.append('bed') tools.append('bed')
if not self.bed and self.number_of_tools == 0: if not self.bed and self.number_of_tools == 0:
@ -167,12 +179,12 @@ class OctoPrintAPI:
"""Send a get request, and return the response as a dict.""" """Send a get request, and return the response as a dict."""
# Only query the API at most every 30 seconds # Only query the API at most every 30 seconds
now = time.time() now = time.time()
if endpoint == "job": if endpoint == 'job':
last_time = self.job_last_reading[1] last_time = self.job_last_reading[1]
if last_time is not None: if last_time is not None:
if now - last_time < 30.0: if now - last_time < 30.0:
return self.job_last_reading[0] return self.job_last_reading[0]
elif endpoint == "printer": elif endpoint == 'printer':
last_time = self.printer_last_reading[1] last_time = self.printer_last_reading[1]
if last_time is not None: if last_time is not None:
if now - last_time < 30.0: if now - last_time < 30.0:
@ -183,11 +195,11 @@ class OctoPrintAPI:
response = requests.get( response = requests.get(
url, headers=self.headers, timeout=9) url, headers=self.headers, timeout=9)
response.raise_for_status() response.raise_for_status()
if endpoint == "job": if endpoint == 'job':
self.job_last_reading[0] = response.json() self.job_last_reading[0] = response.json()
self.job_last_reading[1] = time.time() self.job_last_reading[1] = time.time()
self.job_available = True self.job_available = True
elif endpoint == "printer": elif endpoint == 'printer':
self.printer_last_reading[0] = response.json() self.printer_last_reading[0] = response.json()
self.printer_last_reading[1] = time.time() self.printer_last_reading[1] = time.time()
self.printer_available = True self.printer_available = True
@ -200,13 +212,13 @@ class OctoPrintAPI:
log_string = "Failed to update OctoPrint status. " + \ log_string = "Failed to update OctoPrint status. " + \
" Error: %s" % (conn_exc) " Error: %s" % (conn_exc)
# Only log the first failure # Only log the first failure
if endpoint == "job": if endpoint == 'job':
log_string = "Endpoint: job " + log_string log_string = "Endpoint: job " + log_string
if not self.job_error_logged: if not self.job_error_logged:
_LOGGER.error(log_string) _LOGGER.error(log_string)
self.job_error_logged = True self.job_error_logged = True
self.job_available = False self.job_available = False
elif endpoint == "printer": elif endpoint == 'printer':
log_string = "Endpoint: printer " + log_string log_string = "Endpoint: printer " + log_string
if not self.printer_error_logged: if not self.printer_error_logged:
_LOGGER.error(log_string) _LOGGER.error(log_string)
@ -229,7 +241,7 @@ def get_value_from_json(json_dict, sensor_type, group, tool):
return None return None
if sensor_type in json_dict[group]: if sensor_type in json_dict[group]:
if sensor_type == "target" and json_dict[sensor_type] is None: if sensor_type == 'target' and json_dict[sensor_type] is None:
return 0 return 0
return json_dict[group][sensor_type] return json_dict[group][sensor_type]