mirror of
https://github.com/skalavala/mysmarthome.git
synced 2025-08-27 14:57:19 +00:00
minor updates
This commit is contained in:
Mahasri Kalavala
@@ -1,161 +0,0 @@
|
||||
"""
|
||||
@ Author : Suresh Kalavala
|
||||
@ Date : 09/14/2017
|
||||
@ Description : Input Label - A label that holds data
|
||||
|
||||
@ Notes: Copy this file and services.yaml files and place it in your
|
||||
"Home Assistant Config folder\custom_components\" folder
|
||||
|
||||
To use the component, have the following in your .yaml file:
|
||||
The 'value' is optional, by default, it is set to 0
|
||||
|
||||
input_label:
|
||||
some_string1:
|
||||
name: Some String 1
|
||||
icon: mdi:alphabetical
|
||||
|
||||
input_label:
|
||||
name: Some String 2
|
||||
value: 'Hello, Home Assistant!'
|
||||
icon: mdi:alphabetical
|
||||
|
||||
"""
|
||||
"""
|
||||
Component to provide input_label.
|
||||
|
||||
For more details about this component, please contact Suresh Kalavala
|
||||
"""
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.const import (ATTR_ENTITY_ID, CONF_ICON, CONF_NAME)
|
||||
from homeassistant.helpers.entity_component import EntityComponent
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = 'input_label'
|
||||
ENTITY_ID_FORMAT = DOMAIN + '.{}'
|
||||
|
||||
CONF_INITIAL = 'initial'
|
||||
ATTR_VALUE = "value"
|
||||
DEFAULT_ICON = "mdi:label"
|
||||
|
||||
SERVICE_SETNAME = 'set_name'
|
||||
SERVICE_SETVALUE = 'set_value'
|
||||
SERVICE_SETICON = 'set_icon'
|
||||
|
||||
SERVICE_SCHEMA = vol.Schema({
|
||||
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
|
||||
vol.Optional(ATTR_VALUE): cv.string,
|
||||
vol.Optional(CONF_NAME): cv.icon,
|
||||
vol.Optional(CONF_ICON): cv.icon,
|
||||
})
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
cv.slug: vol.Any({
|
||||
vol.Optional(CONF_ICON, default=DEFAULT_ICON): cv.icon,
|
||||
vol.Optional(ATTR_VALUE, ''): cv.string,
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
}, None)
|
||||
})
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
async def async_setup(hass, config):
|
||||
"""Set up a input_label."""
|
||||
component = EntityComponent(_LOGGER, DOMAIN, hass)
|
||||
|
||||
entities = []
|
||||
|
||||
for object_id, cfg in config[DOMAIN].items():
|
||||
if not cfg:
|
||||
cfg = {}
|
||||
name = cfg.get(CONF_NAME)
|
||||
initial = cfg.get(ATTR_VALUE)
|
||||
icon = cfg.get(CONF_ICON)
|
||||
|
||||
entities.append(InputLabel(object_id, name, initial, icon))
|
||||
|
||||
if not entities:
|
||||
return False
|
||||
|
||||
component.async_register_entity_service(
|
||||
SERVICE_SETNAME, SERVICE_SCHEMA,
|
||||
'async_set_name'
|
||||
)
|
||||
|
||||
component.async_register_entity_service(
|
||||
SERVICE_SETVALUE, SERVICE_SCHEMA,
|
||||
'async_set_value'
|
||||
)
|
||||
|
||||
component.async_register_entity_service(
|
||||
SERVICE_SETICON, SERVICE_SCHEMA,
|
||||
'async_set_icon'
|
||||
)
|
||||
|
||||
await component.async_add_entities(entities)
|
||||
return True
|
||||
|
||||
class InputLabel(RestoreEntity):
|
||||
"""Representation of a input_label."""
|
||||
|
||||
def __init__(self, object_id, name, initial, icon):
|
||||
"""Initialize a input_label."""
|
||||
self.entity_id = ENTITY_ID_FORMAT.format(object_id)
|
||||
self._name = name
|
||||
self._current_value = initial
|
||||
self._icon = icon
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""If entity should be polled."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return name of the input_label."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon to be used for this entity."""
|
||||
return self._icon
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the current value of the input_label."""
|
||||
return self._current_value
|
||||
|
||||
@property
|
||||
def state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
return {
|
||||
ATTR_VALUE: self._current_value,
|
||||
}
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Run when entity about to be added to hass."""
|
||||
|
||||
await super().async_added_to_hass()
|
||||
if self._current_value is not None:
|
||||
return
|
||||
|
||||
state = await self.async_get_last_state()
|
||||
value = state and state.state
|
||||
self._current_value = value
|
||||
|
||||
async def async_set_name(self, value):
|
||||
self._name = value
|
||||
await self.async_update_ha_state()
|
||||
|
||||
async def async_set_icon(self, value):
|
||||
self._icon = value
|
||||
await self.async_update_ha_state()
|
||||
|
||||
async def async_set_value(self, value):
|
||||
self._current_value = value
|
||||
await self.async_update_ha_state()
|
||||
@@ -1,296 +0,0 @@
|
||||
#
|
||||
# API URLS: http://192.168.xxx.xxx/api/printer and http://192.168.xxx.xxx/api/job
|
||||
#
|
||||
|
||||
"""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,
|
||||
PERCENTAGE,
|
||||
TEMP_CELSIUS,
|
||||
TIME_SECONDS,
|
||||
)
|
||||
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",
|
||||
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]
|
||||
protocol = "https" if printer[CONF_SSL] else "http"
|
||||
base_url = (
|
||||
f"{protocol}://{printer[CONF_HOST]}:{printer[CONF_PORT]}"
|
||||
f"{printer[CONF_PATH]}api/"
|
||||
)
|
||||
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
|
||||
@@ -1,83 +0,0 @@
|
||||
"""Support for monitoring OctoPrint binary sensors."""
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorEntity
|
||||
|
||||
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(BinarySensorEntity):
|
||||
"""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
|
||||
@@ -1,7 +0,0 @@
|
||||
{
|
||||
"domain": "octoprint",
|
||||
"name": "OctoPrint",
|
||||
"documentation": "https://www.home-assistant.io/integrations/octoprint",
|
||||
"after_dependencies": ["discovery"],
|
||||
"codeowners": []
|
||||
}
|
||||
@@ -1,139 +0,0 @@
|
||||
"""Support for monitoring OctoPrint sensors."""
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.const import PERCENTAGE, TEMP_CELSIUS
|
||||
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 = f"{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, 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