diff --git a/homeassistant/components/octoprint/__init__.py b/homeassistant/components/octoprint/__init__.py index 6f178f2657844..53bb44c9fb086 100644 --- a/homeassistant/components/octoprint/__init__.py +++ b/homeassistant/components/octoprint/__init__.py @@ -1,9 +1,8 @@ """Support for monitoring OctoPrint 3D printers.""" +from datetime import timedelta import logging -import time -from aiohttp.hdrs import CONTENT_TYPE -import requests +from pyoctoprintapi import OctoprintClient, PrinterOffline import voluptuous as vol from homeassistant.components.discovery import SERVICE_OCTOPRINT @@ -17,15 +16,15 @@ CONF_PORT, CONF_SENSORS, CONF_SSL, - CONTENT_TYPE_JSON, - PERCENTAGE, - TEMP_CELSIUS, - TIME_SECONDS, ) +from homeassistant.core import HomeAssistant from homeassistant.helpers import discovery +from homeassistant.helpers.aiohttp_client import async_get_clientsession import homeassistant.helpers.config_validation as cv -from homeassistant.helpers.discovery import load_platform +from homeassistant.helpers.discovery import async_load_platform +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.util import slugify as util_slugify +import homeassistant.util.dt as dt_util _LOGGER = logging.getLogger(__name__) @@ -53,11 +52,10 @@ def ensure_valid_path(value): return value -BINARY_SENSOR_TYPES = { - # API Endpoint, Group, Key, unit - "Printing": ["printer", "state", "printing", None], - "Printing Error": ["printer", "state", "error", None], -} +BINARY_SENSOR_TYPES = [ + "Printing", + "Printing Error", +] BINARY_SENSOR_SCHEMA = vol.Schema( { @@ -68,26 +66,13 @@ def ensure_valid_path(value): } ) -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"], -} +SENSOR_TYPES = [ + "Temperatures", + "Current State", + "Job Percentage", + "Time Remaining", + "Time Elapsed", +] SENSOR_SCHEMA = vol.Schema( { @@ -127,16 +112,16 @@ def ensure_valid_path(value): ) -def setup(hass, config): +async def async_setup(hass, config): """Set up the OctoPrint component.""" printers = hass.data[DOMAIN] = {} success = False - def device_discovered(service, info): + async 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) + discovery.async_listen(hass, SERVICE_OCTOPRINT, device_discovered) if DOMAIN not in config: # Skip the setup if there is no configuration present @@ -147,169 +132,87 @@ def device_discovered(service, info): protocol = "https" if printer[CONF_SSL] else "http" base_url = ( f"{protocol}://{printer[CONF_HOST]}:{printer[CONF_PORT]}" - f"{printer[CONF_PATH]}api/" + f"{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 + + session = async_get_clientsession(hass) + octoprint = OctoprintClient( + printer[CONF_HOST], + session, + printer[CONF_PORT], + printer[CONF_SSL], + printer[CONF_PATH], + ) + octoprint.set_api_key(printer[CONF_API_KEY]) + coordinator = OctoprintDataUpdateCoordinator(hass, octoprint, base_url, 30) + await coordinator.async_refresh() + + printers[base_url] = coordinator sensors = printer[CONF_SENSORS][CONF_MONITORED_CONDITIONS] - load_platform( - hass, - "sensor", - DOMAIN, - {"name": name, "base_url": base_url, "sensors": sensors}, - config, + hass.async_create_task( + async_load_platform( + hass, + "sensor", + DOMAIN, + { + "name": name, + "base_url": base_url, + "sensors": sensors, + CONF_NUMBER_OF_TOOLS: printer[CONF_NUMBER_OF_TOOLS], + CONF_BED: printer[CONF_BED], + }, + 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, + hass.async_create_task( + async_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.printer_error_logged = False - self.available = False - self.available_error_logged = False - self.job_error_logged = False - self.bed = bed - self.number_of_tools = number_of_tools +class OctoprintDataUpdateCoordinator(DataUpdateCoordinator): + """Class to manage fetching Octoprint data.""" - 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 __init__( + self, + hass: HomeAssistant, + octoprint: OctoprintClient, + device_id: str, + interval: int, + ): + """Initialize.""" + super().__init__( + hass, + _LOGGER, + name=f"octoprint-{device_id}", + update_interval=timedelta(seconds=interval), + ) + self._octoprint = octoprint + self.data = {"printer": None, "job": None, "last_read_time": None} - 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] + async def _async_update_data(self): + """Update data via API.""" + printer = None + job = await self._octoprint.get_job_info() - url = self.api_url + endpoint + # If octoprint is on, but the printer is disconnected + # printer will return a 409, so continue using the last + # reading if there is one 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 - self.available_error_logged = False - - return response.json() - - except requests.ConnectionError as exc_con: - log_string = "Failed to connect to Octoprint server. Error: %s" % exc_con - - if not self.available_error_logged: - _LOGGER.error(log_string) - self.job_available = False - self.printer_available = False - self.available_error_logged = True - - return None - - except requests.HTTPError as ex_http: - status_code = ex_http.response.status_code - - log_string = "Failed to update OctoPrint status. Error: %s" % ex_http - # 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": - if ( - status_code == 409 - ): # octoprint returns HTTP 409 when printer is not connected (and many other states) - self.printer_available = False - else: - 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 - - 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] + printer = await self._octoprint.get_printer_info() + except PrinterOffline: + _LOGGER.error("Unable to retrieve printer information: Printer offline") + if self.data and "printer" in self.data: + printer = self.data["printer"] - return None + return {"job": job, "printer": printer, "last_read_time": dt_util.utcnow()} diff --git a/homeassistant/components/octoprint/binary_sensor.py b/homeassistant/components/octoprint/binary_sensor.py index 0f740525f84cd..f67150a11e608 100644 --- a/homeassistant/components/octoprint/binary_sensor.py +++ b/homeassistant/components/octoprint/binary_sensor.py @@ -1,60 +1,56 @@ """Support for monitoring OctoPrint binary sensors.""" +from abc import abstractmethod import logging +from typing import Optional -import requests +from pyoctoprintapi import OctoprintPrinterInfo from homeassistant.components.binary_sensor import BinarySensorEntity +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, +) -from . import BINARY_SENSOR_TYPES, DOMAIN as COMPONENT_DOMAIN +from . import DOMAIN as COMPONENT_DOMAIN _LOGGER = logging.getLogger(__name__) -def setup_platform(hass, config, add_entities, discovery_info=None): +async def async_setup_platform(hass, config, async_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): + coordinator: DataUpdateCoordinator = hass.data[COMPONENT_DOMAIN][base_url] + + entities = [] + if "Printing" in monitored_conditions: + entities.append(OctoPrintPrintingBinarySensor(coordinator, name)) + + if "Printing Error" in monitored_conditions: + entities.append(OctoPrintPrintingErrorBinarySensor(coordinator, name)) + + async_add_entities(entities, True) + + +class OctoPrintBinarySensorBase(CoordinatorEntity, BinarySensorEntity): """Representation an OctoPrint binary sensor.""" def __init__( - self, api, condition, sensor_type, sensor_name, unit, endpoint, group, tool=None + self, + coordinator: DataUpdateCoordinator, + sensor_name: str, + sensor_type: str, ): """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator) self.sensor_name = sensor_name - if tool is None: - self._name = f"{sensor_name} {condition}" - else: - self._name = f"{sensor_name} {condition}" + self._name = f"{sensor_name} {sensor_type}" 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 @@ -65,19 +61,34 @@ def name(self): @property def is_on(self): """Return true if binary sensor is on.""" - return bool(self._state) + printer = self.coordinator.data["printer"] + if not printer: + return None - @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 + return bool(self._get_flag_state(printer)) + + @abstractmethod + def _get_flag_state(self, printer_info: OctoprintPrinterInfo) -> Optional[bool]: + """Return the value of the sensor flag.""" + + +class OctoPrintPrintingBinarySensor(OctoPrintBinarySensorBase): + """Representation an OctoPrint binary sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Printing") + + def _get_flag_state(self, printer_info: OctoprintPrinterInfo) -> Optional[bool]: + return bool(printer_info.state.flags.printing) + + +class OctoPrintPrintingErrorBinarySensor(OctoPrintBinarySensorBase): + """Representation an OctoPrint binary sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Printing Error") + + def _get_flag_state(self, printer_info: OctoprintPrinterInfo) -> Optional[bool]: + return bool(printer_info.state.flags.error) diff --git a/homeassistant/components/octoprint/manifest.json b/homeassistant/components/octoprint/manifest.json index 28e09cc7be957..8a426021b3c79 100644 --- a/homeassistant/components/octoprint/manifest.json +++ b/homeassistant/components/octoprint/manifest.json @@ -3,5 +3,6 @@ "name": "OctoPrint", "documentation": "https://www.home-assistant.io/integrations/octoprint", "after_dependencies": ["discovery"], + "requirements":["pyoctoprintapi==0.1.3"], "codeowners": [] } diff --git a/homeassistant/components/octoprint/sensor.py b/homeassistant/components/octoprint/sensor.py index 921f355edbe22..ee90f08f21853 100644 --- a/homeassistant/components/octoprint/sensor.py +++ b/homeassistant/components/octoprint/sensor.py @@ -1,20 +1,27 @@ """Support for monitoring OctoPrint sensors.""" +from datetime import timedelta import logging -import requests +from pyoctoprintapi import OctoprintJobInfo, OctoprintPrinterInfo -from homeassistant.const import PERCENTAGE, TEMP_CELSIUS +from homeassistant.const import ( + DEVICE_CLASS_TEMPERATURE, + DEVICE_CLASS_TIMESTAMP, + PERCENTAGE, + TEMP_CELSIUS, +) from homeassistant.helpers.entity import Entity +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, +) -from . import DOMAIN as COMPONENT_DOMAIN, SENSOR_TYPES +from . import CONF_BED, CONF_NUMBER_OF_TOOLS, DOMAIN as COMPONENT_DOMAIN _LOGGER = logging.getLogger(__name__) -NOTIFICATION_ID = "octoprint_notification" -NOTIFICATION_TITLE = "OctoPrint sensor setup error" - -def setup_platform(hass, config, add_entities, discovery_info=None): +async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): """Set up the available OctoPrint sensors.""" if discovery_info is None: return @@ -22,118 +29,204 @@ def setup_platform(hass, config, add_entities, discovery_info=None): 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.
" - "If you do not want to have your printer on
" - " at all times, and you would like to monitor
" - "temperatures, please add
" - "bed and/or number_of_tools to your configuration
" - "and restart.", - title=NOTIFICATION_TITLE, - notification_id=NOTIFICATION_ID, - ) - - devices = [] + number_of_tools = discovery_info[CONF_NUMBER_OF_TOOLS] + bed = discovery_info[CONF_BED] + coordinator: DataUpdateCoordinator = hass.data[COMPONENT_DOMAIN][base_url] + tools = [] + if coordinator.data["printer"]: + tools = [tool.name for tool in coordinator.data["printer"].temperatures] + else: + if number_of_tools > 0: + for tool_number in range(number_of_tools): + tools.append(f"tool{tool_number!s}") + if bed: + tools.append("bed") + + entities = [] 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): + + if "Temperatures" in monitored_conditions and tools: + for tool in tools: + for temp_type in types: + entities.append( + OctoPrintTemperatureSensor(coordinator, name, tool, temp_type) + ) + + if "Current State" in monitored_conditions: + entities.append(OctoPrintStatusSensor(coordinator, name)) + if "Job Percentage" in monitored_conditions: + entities.append(OctoPrintJobPercentageSensor(coordinator, name)) + if "Time Remaining" in monitored_conditions: + entities.append(OctoPrintEstimatedFinishTimeSensor(coordinator, name)) + if "Time Elapsed" in monitored_conditions: + entities.append(OctoPrintStartTimeSensor(coordinator, name)) + + async_add_entities(entities, True) + + +class OctoPrintSensorBase(CoordinatorEntity, Entity): """Representation of an OctoPrint sensor.""" def __init__( self, - api, - condition, - sensor_type, - sensor_name, - unit, - endpoint, - group, - tool=None, - icon=None, + coordinator: DataUpdateCoordinator, + sensor_name: str, + sensor_type: str, ): """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 + super().__init__(coordinator) 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) + self._sensor_name = sensor_name + self._name = f"{sensor_name} {sensor_type}" @property def name(self): """Return the name of the sensor.""" return self._name + +class OctoPrintStatusSensor(OctoPrintSensorBase): + """Representation of an OctoPrint sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Current State") + + @property + def state(self): + """Return sensor state.""" + printer: OctoprintPrinterInfo = self.coordinator.data["printer"] + if not printer: + return None + + return printer.state.text + + @property + def icon(self): + """Icon to use in the frontend.""" + return "mdi:printer-3d" + + +class OctoPrintJobPercentageSensor(OctoPrintSensorBase): + """Representation of an OctoPrint sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Job Percentage") + @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 + """Return sensor state.""" + job: OctoprintJobInfo = self.coordinator.data["job"] + if not job: + return None + + state = job.progress.completion + if not state: + return 0 + + return round(state, 2) @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 + return PERCENTAGE @property def icon(self): """Icon to use in the frontend.""" - return self._icon + return "mdi:file-percent" + + +class OctoPrintEstimatedFinishTimeSensor(OctoPrintSensorBase): + """Representation of an OctoPrint sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Estimated Finish Time") + + @property + def state(self): + """Return sensor state.""" + job: OctoprintJobInfo = self.coordinator.data["job"] + if not job or not job.progress.print_time_left: + return None + + read_time = self.coordinator.data["last_read_time"] + + return (read_time + timedelta(seconds=job.progress.print_time_left)).isoformat() + + @property + def device_class(self): + """Return the device class of the sensor.""" + return DEVICE_CLASS_TIMESTAMP + + +class OctoPrintStartTimeSensor(OctoPrintSensorBase): + """Representation of an OctoPrint sensor.""" + + def __init__(self, coordinator: DataUpdateCoordinator, sensor_name: str): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, "Start Time") + + @property + def state(self): + """Return sensor state.""" + job: OctoprintJobInfo = self.coordinator.data["job"] + + if not job or not job.progress.print_time: + return None + + read_time = self.coordinator.data["last_read_time"] + + return (read_time - timedelta(seconds=job.progress.print_time)).isoformat() + + @property + def device_class(self): + """Return the device class of the sensor.""" + return DEVICE_CLASS_TIMESTAMP + + +class OctoPrintTemperatureSensor(OctoPrintSensorBase): + """Representation of an OctoPrint sensor.""" + + def __init__( + self, + coordinator: DataUpdateCoordinator, + sensor_name: str, + tool: str, + temp_type: str, + ): + """Initialize a new OctoPrint sensor.""" + super().__init__(coordinator, sensor_name, f"{temp_type} {tool} temp") + self._temp_type = temp_type + self._api_tool = tool + self._state = 0 + + @property + def unit_of_measurement(self): + """Return the unit of measurement of this entity, if any.""" + return TEMP_CELSIUS + + @property + def device_class(self): + """Return the device class of this entity.""" + return DEVICE_CLASS_TEMPERATURE + + @property + def state(self): + """Return sensor state.""" + printer: OctoprintPrinterInfo = self.coordinator.data["printer"] + if not printer: + return None + + for temp in printer.temperatures: + if temp.name == self._api_tool: + return round( + temp.actual_temp + if self._temp_type == "actual" + else temp.target_temp, + 2, + ) + + return None diff --git a/requirements_all.txt b/requirements_all.txt index 4dbee8b0a65e3..183943c1b061f 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1580,6 +1580,9 @@ pynzbgetapi==0.2.0 # homeassistant.components.obihai pyobihai==1.3.1 +# homeassistant.components.octoprint +pyoctoprintapi==0.1.3 + # homeassistant.components.ombi pyombi==0.1.10 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 88e8dffff1cef..04f682f5640f8 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -831,6 +831,9 @@ pynx584==0.5 # homeassistant.components.nzbget pynzbgetapi==0.2.0 +# homeassistant.components.octoprint +pyoctoprintapi==0.1.3 + # homeassistant.components.openuv pyopenuv==1.0.9 diff --git a/tests/components/octoprint/__init__.py b/tests/components/octoprint/__init__.py new file mode 100644 index 0000000000000..f1e8831461a23 --- /dev/null +++ b/tests/components/octoprint/__init__.py @@ -0,0 +1,45 @@ +"""Tests for the OctoPrint integration.""" +from unittest.mock import patch + +from pyoctoprintapi import OctoprintJobInfo, OctoprintPrinterInfo + +from homeassistant.components.octoprint import DOMAIN +from homeassistant.setup import async_setup_component + +DEFAULT_JOB = { + "job": {}, + "progress": {"completion": 50}, +} + +DEFAULT_PRINTER = { + "state": { + "flags": {"printing": True, "error": False}, + "text": "Operational", + }, + "temperature": [], +} + + +async def init_integration( + hass, platform, printer: dict = DEFAULT_PRINTER, job: dict = DEFAULT_JOB +): + """Set up the octoprint integration in Home Assistant.""" + config = { + DOMAIN: { + "host": "192.168.1.5", + "api_key": "test-key", + "ssl": False, + "port": 80, + "path": "/", + "name": "Octoprint", + } + } + with patch( + "pyoctoprintapi.OctoprintClient.get_printer_info", + return_value=OctoprintPrinterInfo(printer), + ), patch( + "pyoctoprintapi.OctoprintClient.get_job_info", + return_value=OctoprintJobInfo(job), + ): + assert await async_setup_component(hass, DOMAIN, config) + await hass.async_block_till_done() diff --git a/tests/components/octoprint/test_binary_sensor.py b/tests/components/octoprint/test_binary_sensor.py new file mode 100644 index 0000000000000..685479a7a62f6 --- /dev/null +++ b/tests/components/octoprint/test_binary_sensor.py @@ -0,0 +1,27 @@ +"""The tests for Octoptint binary sensor module.""" + +from homeassistant.const import STATE_OFF, STATE_ON + +from . import init_integration + + +async def test_sensors(hass): + """Test the underlying sensors.""" + printer = { + "state": { + "flags": {"printing": True, "error": False}, + "text": "Operational", + }, + "temperature": [], + } + await init_integration(hass, "binary_sensor", printer=printer) + + state = hass.states.get("binary_sensor.octoprint_printing") + assert state is not None + assert state.state == STATE_ON + assert state.name == "Octoprint Printing" + + state = hass.states.get("binary_sensor.octoprint_printing_error") + assert state is not None + assert state.state == STATE_OFF + assert state.name == "Octoprint Printing Error" diff --git a/tests/components/octoprint/test_sensor.py b/tests/components/octoprint/test_sensor.py new file mode 100644 index 0000000000000..7bf2d6b580549 --- /dev/null +++ b/tests/components/octoprint/test_sensor.py @@ -0,0 +1,59 @@ +"""The tests for Octoptint binary sensor module.""" +from datetime import datetime +from unittest.mock import patch + +from . import init_integration + + +async def test_sensors(hass): + """Test the underlying sensors.""" + printer = { + "state": { + "flags": {}, + "text": "Operational", + }, + "temperature": {"tool1": {"actual": 18.83136, "target": 37.83136}}, + } + job = { + "job": {}, + "progress": {"completion": 50, "printTime": 600, "printTimeLeft": 6000}, + } + with patch( + "homeassistant.util.dt.utcnow", return_value=datetime(2020, 2, 20, 9, 10, 0) + ): + await init_integration(hass, "sensor", printer=printer, job=job) + + state = hass.states.get("sensor.octoprint_job_percentage") + assert state is not None + assert state.state == "50" + assert state.name == "Octoprint Job Percentage" + + state = hass.states.get("sensor.octoprint_current_state") + assert state is not None + assert state.state == "Operational" + assert state.name == "Octoprint Current State" + + state = hass.states.get("sensor.octoprint_actual_tool1_temp") + assert state is not None + assert state.state == "18.83" + assert state.name == "Octoprint actual tool1 temp" + + state = hass.states.get("sensor.octoprint_target_tool1_temp") + assert state is not None + assert state.state == "37.83" + assert state.name == "Octoprint target tool1 temp" + + state = hass.states.get("sensor.octoprint_target_tool1_temp") + assert state is not None + assert state.state == "37.83" + assert state.name == "Octoprint target tool1 temp" + + state = hass.states.get("sensor.octoprint_start_time") + assert state is not None + assert state.state == "2020-02-20T09:00:00" + assert state.name == "Octoprint Start Time" + + state = hass.states.get("sensor.octoprint_estimated_finish_time") + assert state is not None + assert state.state == "2020-02-20T10:50:00" + assert state.name == "Octoprint Estimated Finish Time"