Skip to content

Commit

Permalink
Update typing and remove depecated device classes.
Browse files Browse the repository at this point in the history
Cleaned up some un-needed functions in favor of defaults.
Fixes #10.
  • Loading branch information
eseglem committed Jan 14, 2023
1 parent 61f4838 commit 7637ed8
Show file tree
Hide file tree
Showing 6 changed files with 175 additions and 169 deletions.
47 changes: 17 additions & 30 deletions custom_components/wattbox/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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(
{
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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.
Expand All @@ -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.
Expand All @@ -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
42 changes: 19 additions & 23 deletions custom_components/wattbox/binary_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,22 @@

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

_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 = []
Expand All @@ -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
113 changes: 78 additions & 35 deletions custom_components/wattbox/const.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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}
Expand All @@ -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",
},
}
32 changes: 11 additions & 21 deletions custom_components/wattbox/entity.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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."""
Expand All @@ -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
Loading

0 comments on commit 7637ed8

Please sign in to comment.