diff --git a/custom_components/wattbox/__init__.py b/custom_components/wattbox/__init__.py index 809a7d7..5a3c5d8 100644 --- a/custom_components/wattbox/__init__.py +++ b/custom_components/wattbox/__init__.py @@ -6,8 +6,9 @@ """ import logging import os -from datetime import timedelta +from datetime import datetime from functools import partial +from typing import Final, List import homeassistant.helpers.config_validation as cv import voluptuous as vol @@ -20,9 +21,11 @@ CONF_SCAN_INTERVAL, CONF_USERNAME, ) +from homeassistant.core import HomeAssistant from homeassistant.helpers import discovery from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.event import async_track_time_interval +from homeassistant.helpers.typing import ConfigType from .const import ( BINARY_SENSOR_TYPES, @@ -40,13 +43,11 @@ TOPIC_UPDATE, ) -REQUIREMENTS = ["pywattbox>=0.4.0"] +REQUIREMENTS: Final[List[str]] = ["pywattbox>=0.4.0"] _LOGGER = logging.getLogger(__name__) -MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=1) - -ALL_SENSOR_TYPES = list({**BINARY_SENSOR_TYPES, **SENSOR_TYPES}.keys()) +ALL_SENSOR_TYPES: Final[List[str]] = [*BINARY_SENSOR_TYPES.keys(), *SENSOR_TYPES.keys()] WATTBOX_HOST_SCHEMA = vol.Schema( { @@ -70,9 +71,9 @@ ) -async def async_setup(hass, config): +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up this component.""" - from pywattbox import WattBox + from pywattbox import WattBox # pylint: disable=import-outside-toplevel # Print startup message _LOGGER.info(STARTUP) @@ -85,6 +86,7 @@ async def async_setup(hass, config): hass.data[DOMAIN_DATA] = dict() for wattbox_host in config[DOMAIN]: + _LOGGER.debug(repr(wattbox_host)) # Create DATA dict host = wattbox_host.get(CONF_HOST) password = wattbox_host.get(CONF_PASSWORD) @@ -107,33 +109,21 @@ async def async_setup(hass, config): scan_interval = wattbox_host.get(CONF_SCAN_INTERVAL) async_track_time_interval( - hass, partial(scan_update_data, hass=hass, name=name), scan_interval + hass, partial(update_data, hass=hass, name=name), scan_interval ) # Extra logging to ensure the right outlets are set up. - _LOGGER.debug(", ".join([str(v) for k, v in hass.data[DOMAIN_DATA].items()])) + _LOGGER.debug(", ".join([str(v) for _, v in hass.data[DOMAIN_DATA].items()])) _LOGGER.debug(repr(hass.data[DOMAIN_DATA])) for _, wattbox in hass.data[DOMAIN_DATA].items(): _LOGGER.debug("%s has %s outlets", wattbox, len(wattbox.outlets)) - for o in wattbox.outlets: - _LOGGER.debug("Outlet: %s - %s", o, repr(o)) + for outlet in wattbox.outlets: + _LOGGER.debug("Outlet: %s - %s", outlet, repr(outlet)) return True -# Setup scheduled updates -async def scan_update_data(_, hass, name): - """Scan update data wrapper.""" - - _LOGGER.debug( - "Scan Update Data: %s - %s", - hass.data[DOMAIN_DATA][name], - repr(hass.data[DOMAIN_DATA][name]), - ) - await update_data(hass, name) - - -async def update_data(hass, name): +async def update_data(_: datetime, hass: HomeAssistant, name: str) -> None: """Update data.""" # This is where the main logic to update platform data goes. @@ -150,7 +140,7 @@ async def update_data(hass, name): _LOGGER.error("Could not update data - %s", error) -async def check_files(hass): +async def check_files(hass: HomeAssistant) -> bool: """Return bool that indicates if all files are present.""" # Verify that the user downloaded all files. @@ -163,8 +153,5 @@ async def check_files(hass): if missing: _LOGGER.critical("The following files are missing: %s", str(missing)) - returnvalue = False - else: - returnvalue = True - - return returnvalue + return False + return True diff --git a/custom_components/wattbox/binary_sensor.py b/custom_components/wattbox/binary_sensor.py index 896658c..e21dd62 100644 --- a/custom_components/wattbox/binary_sensor.py +++ b/custom_components/wattbox/binary_sensor.py @@ -4,6 +4,9 @@ from homeassistant.components.binary_sensor import BinarySensorEntity from homeassistant.const import CONF_NAME, CONF_RESOURCES +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from .const import BINARY_SENSOR_TYPES, DOMAIN_DATA from .entity import WattBoxEntity @@ -11,9 +14,12 @@ _LOGGER = logging.getLogger(__name__) -async def async_setup_platform( - hass, config, async_add_entities, discovery_info=None -): # pylint: disable=unused-argument +async def async_setup_platform( # pylint: disable=unused-argument + hass: HomeAssistant, + config: ConfigType, + async_add_entities: AddEntitiesCallback, + discovery_info: DiscoveryInfoType, +) -> None: """Setup binary_sensor platform.""" name = discovery_info[CONF_NAME] entities = [] @@ -32,30 +38,20 @@ async def async_setup_platform( class WattBoxBinarySensor(WattBoxEntity, BinarySensorEntity): """WattBox binary_sensor class.""" - def __init__(self, hass, name, sensor_type): + def __init__(self, hass: HomeAssistant, name: str, sensor_type: str) -> None: super().__init__(hass, name, sensor_type) - self.type = sensor_type - self._status = False - self._name = name + " " + BINARY_SENSOR_TYPES[sensor_type]["name"] + self.type: str = sensor_type + self.flipped: bool = BINARY_SENSOR_TYPES[self.type]["flipped"] + self._attr_name = name + " " + BINARY_SENSOR_TYPES[sensor_type]["name"] + self._attr_device_class = BINARY_SENSOR_TYPES[self.type]["device_class"] - async def async_update(self): + async def async_update(self) -> None: """Update the sensor.""" # Get domain data wattbox = self.hass.data[DOMAIN_DATA][self.wattbox_name] # Check the data and update the value. - self._status = getattr(wattbox, self.type) - - @property - def device_class(self): - """Return the class of this binary_sensor.""" - return BINARY_SENSOR_TYPES[self.type]["device_class"] - - @property - def is_on(self): - """Return true if the binary_sensor is on.""" - return ( - not self._status - if BINARY_SENSOR_TYPES[self.type]["flipped"] - else self._status - ) + value: bool | None = getattr(wattbox, self.type) + if value is not None and self.flipped: + value = not value + self._attr_is_on = value diff --git a/custom_components/wattbox/const.py b/custom_components/wattbox/const.py index 144f41a..285cd43 100644 --- a/custom_components/wattbox/const.py +++ b/custom_components/wattbox/const.py @@ -1,13 +1,9 @@ """Constants for wattbox.""" from datetime import timedelta +from typing import Dict, Final, List, TypedDict -from homeassistant.components.binary_sensor import ( - DEVICE_CLASS_CONNECTIVITY, - DEVICE_CLASS_PLUG, - DEVICE_CLASS_PROBLEM, - DEVICE_CLASS_SAFETY, -) +from homeassistant.components.binary_sensor import BinarySensorDeviceClass from homeassistant.const import ( ELECTRIC_POTENTIAL_VOLT, PERCENTAGE, @@ -16,14 +12,21 @@ ) # Base component constants -DOMAIN = "wattbox" -DOMAIN_DATA = f"{DOMAIN}_data" -VERSION = "0.8.0" -PLATFORMS = ["binary_sensor", "sensor", "switch"] -REQUIRED_FILES = ["binary_sensor.py", "const.py", "sensor.py", "switch.py"] -ISSUE_URL = "https://github.com/eseglem/hass-wattbox/issues" - -STARTUP = f""" +DOMAIN: Final[str] = "wattbox" +DOMAIN_DATA: Final[str] = f"{DOMAIN}_data" +VERSION: Final[str] = "0.8.1" +PLATFORMS: Final[List[str]] = ["binary_sensor", "sensor", "switch"] +REQUIRED_FILES: Final[List[str]] = [ + "binary_sensor.py", + "const.py", + "sensor.py", + "switch.py", +] +ISSUE_URL: Final[str] = "https://github.com/eseglem/hass-wattbox/issues" + +STARTUP: Final[ + str +] = f""" ------------------------------------------------------------------- {DOMAIN} Version: {VERSION} @@ -34,47 +37,87 @@ """ # Icons -ICON = "mdi:power" -PLUG_ICON = "mdi:power-socket-us" +ICON: Final[str] = "mdi:power" +PLUG_ICON: Final[str] = "mdi:power-socket-us" # Defaults -DEFAULT_NAME = "WattBox" -DEFAULT_PASSWORD = DOMAIN -DEFAULT_PORT = 80 -DEFAULT_USER = DOMAIN -DEFAULT_SCAN_INTERVAL = timedelta(seconds=30) +DEFAULT_NAME: Final[str] = "WattBox" +DEFAULT_PASSWORD: Final[str] = DOMAIN +DEFAULT_PORT: Final[int] = 80 +DEFAULT_USER: Final[str] = DOMAIN +DEFAULT_SCAN_INTERVAL: Final[timedelta] = timedelta(seconds=30) + +TOPIC_UPDATE: Final[str] = "{}_data_update_{}" + + +class _BinarySensorDict(TypedDict): + """TypedDict for use in BINARY_SENSOR_TYPES""" -TOPIC_UPDATE = "{}_data_update_{}" + name: str + device_class: BinarySensorDeviceClass | None + flipped: bool -BINARY_SENSOR_TYPES = { - "audible_alarm": {"name": "Audible Alarm", "device_class": None, "flipped": False}, + +BINARY_SENSOR_TYPES: Final[Dict[str, _BinarySensorDict]] = { + "audible_alarm": { + "name": "Audible Alarm", + "device_class": BinarySensorDeviceClass.SOUND, + "flipped": False, + }, "auto_reboot": {"name": "Auto Reboot", "device_class": None, "flipped": False}, "battery_health": { "name": "Battery Health", - "device_class": DEVICE_CLASS_PROBLEM, + "device_class": BinarySensorDeviceClass.PROBLEM, "flipped": True, }, "battery_test": {"name": "Battery Test", "device_class": None, "flipped": False}, "cloud_status": { "name": "Cloud Status", - "device_class": DEVICE_CLASS_CONNECTIVITY, + "device_class": BinarySensorDeviceClass.CONNECTIVITY, "flipped": False, }, "has_ups": {"name": "Has UPS", "device_class": None, "flipped": False}, "mute": {"name": "Mute", "device_class": None, "flipped": False}, - "power_lost": {"name": "Power", "device_class": DEVICE_CLASS_PLUG, "flipped": True}, + "power_lost": { + "name": "Power", + "device_class": BinarySensorDeviceClass.PLUG, + "flipped": True, + }, "safe_voltage_status": { "name": "Safe Voltage Status", - "device_class": DEVICE_CLASS_SAFETY, + "device_class": BinarySensorDeviceClass.SAFETY, "flipped": True, }, } -SENSOR_TYPES = { - "battery_charge": ["Battery Charge", PERCENTAGE, "mdi:gauge"], - "battery_load": ["Battery Load", PERCENTAGE, "mdi:gauge"], - "current_value": ["Current", "A", "mdi:current-ac"], - "est_run_time": ["Estimated Run Time", TIME_MINUTES, "mdi:update"], - "power_value": ["Power", POWER_WATT, "mdi:lightbulb-outline"], - "voltage_value": ["Voltage", ELECTRIC_POTENTIAL_VOLT, "mdi:flash-circle"], + +class _SensorTypeDict(TypedDict): + name: str + unit: str + icon: str + + +SENSOR_TYPES: Final[Dict[str, _SensorTypeDict]] = { + "battery_charge": { + "name": "Battery Charge", + "unit": PERCENTAGE, + "icon": "mdi:battery", + }, + "battery_load": {"name": "Battery Load", "unit": PERCENTAGE, "icon": "mdi:gauge"}, + "current_value": {"name": "Current", "unit": "A", "icon": "mdi:current-ac"}, + "est_run_time": { + "name": "Estimated Run Time", + "unit": TIME_MINUTES, + "icon": "mdi:timer", + }, + "power_value": { + "name": "Power", + "unit": POWER_WATT, + "icon": "mdi:lightbulb-outline", + }, + "voltage_value": { + "name": "Voltage", + "unit": ELECTRIC_POTENTIAL_VOLT, + "icon": "mdi:lightning-bolt-circle", + }, } diff --git a/custom_components/wattbox/entity.py b/custom_components/wattbox/entity.py index 062cc08..daeaf74 100644 --- a/custom_components/wattbox/entity.py +++ b/custom_components/wattbox/entity.py @@ -1,6 +1,7 @@ """Base Entity component for wattbox.""" +from typing import Any, Callable, Dict, Literal -from homeassistant.core import callback +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity import Entity @@ -10,12 +11,16 @@ class WattBoxEntity(Entity): """WattBox Entity class.""" - def __init__(self, hass, name, *args): + _async_unsub_dispatcher_connect: Callable + _attr_should_poll: Literal[False] = False + + def __init__( # pylint: disable=unused-argument + self, hass: HomeAssistant, name: str, *args + ) -> None: self.hass = hass - self.attr = dict() - self.wattbox_name = name - self._name = "" - self.topic = TOPIC_UPDATE.format(DOMAIN, self.wattbox_name) + self._attr_extra_state_attributes: Dict[str, Any] = dict() + self.wattbox_name: str = name + self.topic: str = TOPIC_UPDATE.format(DOMAIN, self.wattbox_name) async def async_added_to_hass(self) -> None: """Register callbacks.""" @@ -33,18 +38,3 @@ async def async_will_remove_from_hass(self) -> None: """Disconnect dispatcher listener when removed.""" if hasattr(self, "_async_unsub_dispatcher_connect"): self._async_unsub_dispatcher_connect() - - @property - def name(self): - """Return the name of the switch.""" - return self._name - - @property - def extra_state_attributes(self): - """Return the state attributes.""" - return self.attr - - @property - def should_poll(self) -> bool: - """Return true.""" - return False diff --git a/custom_components/wattbox/sensor.py b/custom_components/wattbox/sensor.py index ec3762b..92100a4 100644 --- a/custom_components/wattbox/sensor.py +++ b/custom_components/wattbox/sensor.py @@ -1,8 +1,12 @@ """Sensor platform for wattbox.""" import logging +from typing import List -from homeassistant.const import CONF_NAME, CONF_RESOURCES +from homeassistant.const import CONF_NAME, CONF_RESOURCES, STATE_UNKNOWN +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from .const import DOMAIN_DATA, SENSOR_TYPES from .entity import WattBoxEntity @@ -10,12 +14,15 @@ _LOGGER = logging.getLogger(__name__) -async def async_setup_platform( - hass, config, async_add_entities, discovery_info=None -): # pylint: disable=unused-argument +async def async_setup_platform( # pylint: disable=unused-argument + hass: HomeAssistant, + config: ConfigType, + async_add_entities: AddEntitiesCallback, + discovery_info: DiscoveryInfoType, +) -> None: """Setup sensor platform.""" - name = discovery_info[CONF_NAME] - entities = [] + name: str = discovery_info[CONF_NAME] + entities: List[WattBoxSensor] = [] for resource in discovery_info[CONF_RESOURCES]: sensor_type = resource.lower() @@ -31,32 +38,17 @@ async def async_setup_platform( class WattBoxSensor(WattBoxEntity): """WattBox Sensor class.""" - def __init__(self, hass, name, sensor_type): + def __init__(self, hass: HomeAssistant, name: str, sensor_type: str) -> None: super().__init__(hass, name, sensor_type) - self.type = sensor_type - self._name = name + " " + SENSOR_TYPES[self.type][0] - self._state = None - self._unit = SENSOR_TYPES[self.type][1] + self.sensor_type: str = sensor_type + self._attr_name = name + " " + SENSOR_TYPES[self.sensor_type]["name"] + self._attr_unit_of_measurement = SENSOR_TYPES[self.sensor_type]["unit"] + self._attr_icon = SENSOR_TYPES[self.sensor_type]["icon"] - async def async_update(self): + async def async_update(self) -> None: """Update the sensor.""" # Get new data (if any) wattbox = self.hass.data[DOMAIN_DATA][self.wattbox_name] # Check the data and update the value. - self._state = getattr(wattbox, self.type) - - @property - def state(self): - """Return the state of the sensor.""" - return self._state - - @property - def icon(self): - """Return the icon of the sensor.""" - return SENSOR_TYPES[self.type][2] - - @property - def unit_of_measurement(self): - """Return the unit of measurement of this entity, if any.""" - return self._unit + self._attr_state = getattr(wattbox, self.sensor_type, STATE_UNKNOWN) diff --git a/custom_components/wattbox/switch.py b/custom_components/wattbox/switch.py index 9bf2809..883ad04 100644 --- a/custom_components/wattbox/switch.py +++ b/custom_components/wattbox/switch.py @@ -1,9 +1,13 @@ """Switch platform for wattbox.""" import logging +from typing import Final, List -from homeassistant.components.switch import DEVICE_CLASS_OUTLET, SwitchEntity +from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity from homeassistant.const import CONF_NAME +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from .const import DOMAIN_DATA, PLUG_ICON from .entity import WattBoxEntity @@ -11,14 +15,17 @@ _LOGGER = logging.getLogger(__name__) -async def async_setup_platform( - hass, config, async_add_entities, discovery_info=None -): # pylint: disable=unused-argument +async def async_setup_platform( # pylint: disable=unused-argument + hass: HomeAssistant, + config: ConfigType, + async_add_entities: AddEntitiesCallback, + discovery_info: DiscoveryInfoType, +) -> None: """Setup switch platform.""" - name = discovery_info[CONF_NAME] - entities = [] + name: str = discovery_info[CONF_NAME] + entities: List[WattBoxEntity] = [] - num_switches = hass.data[DOMAIN_DATA][name].number_outlets + num_switches: int = hass.data[DOMAIN_DATA][name].number_outlets entities.append(WattBoxMasterSwitch(hass, name)) for i in range(1, num_switches + 1): @@ -30,11 +37,12 @@ async def async_setup_platform( class WattBoxBinarySwitch(WattBoxEntity, SwitchEntity): """WattBox switch class.""" - def __init__(self, hass, name, index): + _attr_device_class: Final[str] = SwitchDeviceClass.OUTLET + + def __init__(self, hass: HomeAssistant, name: str, index: int): super().__init__(hass, name, index) - self.index = index - self._status = False - self._name = name + " Outlet " + str(index) + self.index: int = index + self._attr_name = name + " Outlet " + str(index) async def async_update(self): """Update the sensor.""" @@ -42,14 +50,14 @@ async def async_update(self): outlet = self.hass.data[DOMAIN_DATA][self.wattbox_name].outlets[self.index] # Check the data and update the value. - self._status = outlet.status + self._attr_is_on = outlet.status # Set/update attributes - self.attr["name"] = outlet.name - self.attr["method"] = outlet.method - self.attr["index"] = outlet.index + self._attr_extra_state_attributes["name"] = outlet.name + self._attr_extra_state_attributes["method"] = outlet.method + self._attr_extra_state_attributes["index"] = outlet.index - async def async_turn_on(self, **kwargs): # pylint: disable=unused-argument + async def async_turn_on(self, **kwargs) -> None: # pylint: disable=unused-argument """Turn on the switch.""" _LOGGER.debug( "Turning On: %s - %s", @@ -62,14 +70,14 @@ async def async_turn_on(self, **kwargs): # pylint: disable=unused-argument repr(self.hass.data[DOMAIN_DATA][self.wattbox_name].outlets[self.index]), ) # Update state first so it is not stale. - self._status = True + self._attr_is_on = True self.async_write_ha_state() # Trigger the action on the wattbox. await self.hass.async_add_executor_job( self.hass.data[DOMAIN_DATA][self.wattbox_name].outlets[self.index].turn_on ) - async def async_turn_off(self, **kwargs): # pylint: disable=unused-argument + async def async_turn_off(self, **kwargs) -> None: # pylint: disable=unused-argument """Turn off the switch.""" _LOGGER.debug( "Turning Off: %s - %s", @@ -82,7 +90,7 @@ async def async_turn_off(self, **kwargs): # pylint: disable=unused-argument repr(self.hass.data[DOMAIN_DATA][self.wattbox_name].outlets[self.index]), ) # Update state first so it is not stale. - self._status = False + self._attr_is_on = False self.async_write_ha_state() # Trigger the action on the wattbox. await self.hass.async_add_executor_job( @@ -90,24 +98,14 @@ async def async_turn_off(self, **kwargs): # pylint: disable=unused-argument ) @property - def icon(self): + def icon(self) -> str | None: """Return the icon of this switch.""" return PLUG_ICON - @property - def is_on(self): - """Return true if the switch is on.""" - return self._status - - @property - def device_class(self): - """Return the class of this device, from component DEVICE_CLASSES.""" - return DEVICE_CLASS_OUTLET - class WattBoxMasterSwitch(WattBoxBinarySwitch): """WattBox master switch class.""" - def __init__(self, hass, name): + def __init__(self, hass: HomeAssistant, name: str) -> None: super().__init__(hass, name, 0) - self._name = name + " Master Switch" + self._attr_name = name + " Master Switch"