mirror of
https://github.com/skalavala/mysmarthome.git
synced 2025-08-21 12:42:47 +00:00
minor changes.
This commit is contained in:
Mahasri Kalavala
292
custom_components/octoprint/__init__.py
Normal file
292
custom_components/octoprint/__init__.py
Normal file
@@ -0,0 +1,292 @@
|
||||
"""Support for monitoring OctoPrint 3D printers."""
|
||||
import logging
|
||||
import time
|
||||
|
||||
from aiohttp.hdrs import CONTENT_TYPE
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.discovery import SERVICE_OCTOPRINT
|
||||
from homeassistant.const import (
|
||||
CONF_API_KEY,
|
||||
CONF_BINARY_SENSORS,
|
||||
CONF_HOST,
|
||||
CONF_MONITORED_CONDITIONS,
|
||||
CONF_NAME,
|
||||
CONF_PATH,
|
||||
CONF_PORT,
|
||||
CONF_SENSORS,
|
||||
CONF_SSL,
|
||||
CONTENT_TYPE_JSON,
|
||||
TEMP_CELSIUS,
|
||||
TIME_SECONDS,
|
||||
UNIT_PERCENTAGE,
|
||||
)
|
||||
from homeassistant.helpers import discovery
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.discovery import load_platform
|
||||
from homeassistant.util import slugify as util_slugify
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_BED = "bed"
|
||||
CONF_NUMBER_OF_TOOLS = "number_of_tools"
|
||||
|
||||
DEFAULT_NAME = "OctoPrint"
|
||||
DOMAIN = "octoprint"
|
||||
|
||||
|
||||
def has_all_unique_names(value):
|
||||
"""Validate that printers have an unique name."""
|
||||
names = [util_slugify(printer["name"]) for printer in value]
|
||||
vol.Schema(vol.Unique())(names)
|
||||
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 = f"/{value}"
|
||||
if value[-1] != "/":
|
||||
value += "/"
|
||||
return value
|
||||
|
||||
|
||||
BINARY_SENSOR_TYPES = {
|
||||
# API Endpoint, Group, Key, unit
|
||||
"Printing": ["printer", "state", "printing", None],
|
||||
"Printing Error": ["printer", "state", "error", None],
|
||||
}
|
||||
|
||||
BINARY_SENSOR_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_MONITORED_CONDITIONS, default=list(BINARY_SENSOR_TYPES)
|
||||
): vol.All(cv.ensure_list, [vol.In(BINARY_SENSOR_TYPES)]),
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
SENSOR_TYPES = {
|
||||
# API Endpoint, Group, Key, unit, icon
|
||||
"Temperatures": ["printer", "temperature", "*", TEMP_CELSIUS],
|
||||
"Current State": ["printer", "state", "text", None, "mdi:printer-3d"],
|
||||
"Job Percentage": [
|
||||
"job",
|
||||
"progress",
|
||||
"completion",
|
||||
UNIT_PERCENTAGE,
|
||||
"mdi:file-percent",
|
||||
],
|
||||
"Time Remaining": [
|
||||
"job",
|
||||
"progress",
|
||||
"printTimeLeft",
|
||||
TIME_SECONDS,
|
||||
"mdi:clock-end",
|
||||
],
|
||||
"Time Elapsed": ["job", "progress", "printTime", TIME_SECONDS, "mdi:clock-start"],
|
||||
"User":["job", "job", "user", None, "mdi:printer-3d"],
|
||||
"File":["job", "job", "file", None, "mdi:printer-3d"],
|
||||
}
|
||||
|
||||
SENSOR_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_MONITORED_CONDITIONS, default=list(SENSOR_TYPES)): vol.All(
|
||||
cv.ensure_list, [vol.In(SENSOR_TYPES)]
|
||||
),
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.All(
|
||||
cv.ensure_list,
|
||||
[
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_API_KEY): cv.string,
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_SSL, default=False): cv.boolean,
|
||||
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_NUMBER_OF_TOOLS, default=0): cv.positive_int,
|
||||
vol.Optional(CONF_BED, default=False): cv.boolean,
|
||||
vol.Optional(CONF_SENSORS, default={}): SENSOR_SCHEMA,
|
||||
vol.Optional(
|
||||
CONF_BINARY_SENSORS, default={}
|
||||
): BINARY_SENSOR_SCHEMA,
|
||||
}
|
||||
)
|
||||
],
|
||||
has_all_unique_names,
|
||||
)
|
||||
},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the OctoPrint component."""
|
||||
printers = hass.data[DOMAIN] = {}
|
||||
success = False
|
||||
|
||||
def device_discovered(service, info):
|
||||
"""Get called when an Octoprint server has been discovered."""
|
||||
_LOGGER.debug("Found an Octoprint server: %s", info)
|
||||
|
||||
discovery.listen(hass, SERVICE_OCTOPRINT, device_discovered)
|
||||
|
||||
if DOMAIN not in config:
|
||||
# Skip the setup if there is no configuration present
|
||||
return True
|
||||
|
||||
for printer in config[DOMAIN]:
|
||||
name = printer[CONF_NAME]
|
||||
ssl = "s" if printer[CONF_SSL] else ""
|
||||
base_url = "http{}://{}:{}{}api/".format(
|
||||
ssl, printer[CONF_HOST], printer[CONF_PORT], printer[CONF_PATH]
|
||||
)
|
||||
api_key = printer[CONF_API_KEY]
|
||||
number_of_tools = printer[CONF_NUMBER_OF_TOOLS]
|
||||
bed = printer[CONF_BED]
|
||||
try:
|
||||
octoprint_api = OctoPrintAPI(base_url, api_key, bed, number_of_tools)
|
||||
printers[base_url] = octoprint_api
|
||||
octoprint_api.get("printer")
|
||||
octoprint_api.get("job")
|
||||
except requests.exceptions.RequestException as conn_err:
|
||||
_LOGGER.error("Error setting up OctoPrint API: %r", conn_err)
|
||||
continue
|
||||
|
||||
sensors = printer[CONF_SENSORS][CONF_MONITORED_CONDITIONS]
|
||||
load_platform(
|
||||
hass,
|
||||
"sensor",
|
||||
DOMAIN,
|
||||
{"name": name, "base_url": base_url, "sensors": sensors},
|
||||
config,
|
||||
)
|
||||
b_sensors = printer[CONF_BINARY_SENSORS][CONF_MONITORED_CONDITIONS]
|
||||
load_platform(
|
||||
hass,
|
||||
"binary_sensor",
|
||||
DOMAIN,
|
||||
{"name": name, "base_url": base_url, "sensors": b_sensors},
|
||||
config,
|
||||
)
|
||||
success = True
|
||||
|
||||
return success
|
||||
|
||||
|
||||
class OctoPrintAPI:
|
||||
"""Simple JSON wrapper for OctoPrint's API."""
|
||||
|
||||
def __init__(self, api_url, key, bed, number_of_tools):
|
||||
"""Initialize OctoPrint API and set headers needed later."""
|
||||
self.api_url = api_url
|
||||
self.headers = {CONTENT_TYPE: CONTENT_TYPE_JSON, "X-Api-Key": key}
|
||||
self.printer_last_reading = [{}, None]
|
||||
self.job_last_reading = [{}, None]
|
||||
self.job_available = False
|
||||
self.printer_available = False
|
||||
self.available = False
|
||||
self.printer_error_logged = False
|
||||
self.job_error_logged = False
|
||||
self.bed = bed
|
||||
self.number_of_tools = number_of_tools
|
||||
|
||||
def get_tools(self):
|
||||
"""Get the list of tools that temperature is monitored on."""
|
||||
tools = []
|
||||
if self.number_of_tools > 0:
|
||||
for tool_number in range(0, self.number_of_tools):
|
||||
tools.append(f"tool{tool_number!s}")
|
||||
if self.bed:
|
||||
tools.append("bed")
|
||||
if not self.bed and self.number_of_tools == 0:
|
||||
temps = self.printer_last_reading[0].get("temperature")
|
||||
if temps is not None:
|
||||
tools = temps.keys()
|
||||
return tools
|
||||
|
||||
def get(self, endpoint):
|
||||
"""Send a get request, and return the response as a dict."""
|
||||
# Only query the API at most every 30 seconds
|
||||
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=9)
|
||||
response.raise_for_status()
|
||||
if endpoint == "job":
|
||||
self.job_last_reading[0] = response.json()
|
||||
self.job_last_reading[1] = time.time()
|
||||
self.job_available = True
|
||||
elif endpoint == "printer":
|
||||
self.printer_last_reading[0] = response.json()
|
||||
self.printer_last_reading[1] = time.time()
|
||||
self.printer_available = True
|
||||
self.available = self.printer_available and self.job_available
|
||||
if self.available:
|
||||
self.job_error_logged = False
|
||||
self.printer_error_logged = False
|
||||
return response.json()
|
||||
except Exception as conn_exc: # pylint: disable=broad-except
|
||||
log_string = "Failed to update OctoPrint status. Error: %s" % conn_exc
|
||||
# Only log the first failure
|
||||
if endpoint == "job":
|
||||
log_string = f"Endpoint: job {log_string}"
|
||||
if not self.job_error_logged:
|
||||
_LOGGER.error(log_string)
|
||||
self.job_error_logged = True
|
||||
self.job_available = False
|
||||
elif endpoint == "printer":
|
||||
log_string = f"Endpoint: printer {log_string}"
|
||||
if not self.printer_error_logged:
|
||||
_LOGGER.error(log_string)
|
||||
self.printer_error_logged = True
|
||||
self.printer_available = False
|
||||
self.available = False
|
||||
return None
|
||||
|
||||
def update(self, sensor_type, end_point, group, tool=None):
|
||||
"""Return the value for sensor_type from the provided endpoint."""
|
||||
response = self.get(end_point)
|
||||
if response is not None:
|
||||
return get_value_from_json(response, sensor_type, group, tool)
|
||||
return response
|
||||
|
||||
|
||||
def get_value_from_json(json_dict, sensor_type, group, tool):
|
||||
"""Return the value for sensor_type from the JSON."""
|
||||
if group not in json_dict:
|
||||
return None
|
||||
|
||||
if sensor_type in json_dict[group]:
|
||||
if sensor_type == "target" and json_dict[sensor_type] is None:
|
||||
return 0
|
||||
if sensor_type == "file":
|
||||
return json_dict[group][sensor_type]["name"]
|
||||
return json_dict[group][sensor_type]
|
||||
|
||||
if tool is not None:
|
||||
if sensor_type in json_dict[group][tool]:
|
||||
return json_dict[group][tool][sensor_type]
|
||||
|
||||
return None
|
||||
83
custom_components/octoprint/binary_sensor.py
Normal file
83
custom_components/octoprint/binary_sensor.py
Normal file
@@ -0,0 +1,83 @@
|
||||
"""Support for monitoring OctoPrint binary sensors."""
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorDevice
|
||||
|
||||
from . import BINARY_SENSOR_TYPES, DOMAIN as COMPONENT_DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the available OctoPrint binary sensors."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
name = discovery_info["name"]
|
||||
base_url = discovery_info["base_url"]
|
||||
monitored_conditions = discovery_info["sensors"]
|
||||
octoprint_api = hass.data[COMPONENT_DOMAIN][base_url]
|
||||
|
||||
devices = []
|
||||
for octo_type in monitored_conditions:
|
||||
new_sensor = OctoPrintBinarySensor(
|
||||
octoprint_api,
|
||||
octo_type,
|
||||
BINARY_SENSOR_TYPES[octo_type][2],
|
||||
name,
|
||||
BINARY_SENSOR_TYPES[octo_type][3],
|
||||
BINARY_SENSOR_TYPES[octo_type][0],
|
||||
BINARY_SENSOR_TYPES[octo_type][1],
|
||||
"flags",
|
||||
)
|
||||
devices.append(new_sensor)
|
||||
add_entities(devices, True)
|
||||
|
||||
|
||||
class OctoPrintBinarySensor(BinarySensorDevice):
|
||||
"""Representation an OctoPrint binary sensor."""
|
||||
|
||||
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 = f"{sensor_name} {condition}"
|
||||
else:
|
||||
self._name = f"{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
|
||||
_LOGGER.debug("Created OctoPrint binary sensor %r", self)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return true if binary sensor is on."""
|
||||
return bool(self._state)
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
"""Return the class of this sensor, from DEVICE_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
|
||||
7
custom_components/octoprint/manifest.json
Normal file
7
custom_components/octoprint/manifest.json
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"domain": "octoprint",
|
||||
"name": "OctoPrint",
|
||||
"documentation": "https://www.home-assistant.io/integrations/octoprint",
|
||||
"after_dependencies": ["discovery"],
|
||||
"codeowners": []
|
||||
}
|
||||
139
custom_components/octoprint/sensor.py
Normal file
139
custom_components/octoprint/sensor.py
Normal file
@@ -0,0 +1,139 @@
|
||||
"""Support for monitoring OctoPrint sensors."""
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.const import TEMP_CELSIUS, UNIT_PERCENTAGE
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from . import DOMAIN as COMPONENT_DOMAIN, SENSOR_TYPES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
NOTIFICATION_ID = "octoprint_notification"
|
||||
NOTIFICATION_TITLE = "OctoPrint sensor setup error"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the available OctoPrint sensors."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
name = discovery_info["name"]
|
||||
base_url = discovery_info["base_url"]
|
||||
monitored_conditions = discovery_info["sensors"]
|
||||
octoprint_api = hass.data[COMPONENT_DOMAIN][base_url]
|
||||
tools = octoprint_api.get_tools()
|
||||
|
||||
if "Temperatures" in monitored_conditions:
|
||||
if not tools:
|
||||
hass.components.persistent_notification.create(
|
||||
"Your printer appears to be offline.<br />"
|
||||
"If you do not want to have your printer on <br />"
|
||||
" at all times, and you would like to monitor <br /> "
|
||||
"temperatures, please add <br />"
|
||||
"bed and/or number_of_tools to your configuration <br />"
|
||||
"and restart.",
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID,
|
||||
)
|
||||
|
||||
devices = []
|
||||
types = ["actual", "target"]
|
||||
for octo_type in monitored_conditions:
|
||||
if octo_type == "Temperatures":
|
||||
for tool in tools:
|
||||
for temp_type in types:
|
||||
new_sensor = OctoPrintSensor(
|
||||
octoprint_api,
|
||||
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)
|
||||
else:
|
||||
new_sensor = OctoPrintSensor(
|
||||
octoprint_api,
|
||||
octo_type,
|
||||
SENSOR_TYPES[octo_type][2],
|
||||
name,
|
||||
SENSOR_TYPES[octo_type][3],
|
||||
SENSOR_TYPES[octo_type][0],
|
||||
SENSOR_TYPES[octo_type][1],
|
||||
None,
|
||||
SENSOR_TYPES[octo_type][4],
|
||||
)
|
||||
devices.append(new_sensor)
|
||||
add_entities(devices, True)
|
||||
|
||||
|
||||
class OctoPrintSensor(Entity):
|
||||
"""Representation of an OctoPrint sensor."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api,
|
||||
condition,
|
||||
sensor_type,
|
||||
sensor_name,
|
||||
unit,
|
||||
endpoint,
|
||||
group,
|
||||
tool=None,
|
||||
icon=None,
|
||||
):
|
||||
"""Initialize a new OctoPrint sensor."""
|
||||
self.sensor_name = sensor_name
|
||||
if tool is None:
|
||||
self._name = f"{sensor_name} {condition}"
|
||||
else:
|
||||
self._name = "{} {} {} {}".format(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
|
||||
self._icon = icon
|
||||
_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."""
|
||||
sensor_unit = self.unit_of_measurement
|
||||
if sensor_unit in (TEMP_CELSIUS, UNIT_PERCENTAGE):
|
||||
# API sometimes returns null and not 0
|
||||
if self._state is None:
|
||||
self._state = 0
|
||||
return round(self._state, 2)
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the 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
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Icon to use in the frontend."""
|
||||
return self._icon
|
||||
Reference in New Issue
Block a user