evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Enable strict typing for greeneye_monitor (#58571)

Browse source 
* Enable strict typing for greeneye_monitor

* Fix pylint
This commit is contained in:
Jonathan Keljo 2021-10-29 18:54:40 -07:00 • committed by GitHub
parent 061b1abd1b
commit 687c40a622
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 110 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -52,6 +52,7 @@ homeassistant.components.fritz.*
homeassistant.components.geo_location.*
homeassistant.components.gios.*
homeassistant.components.goalzero.*
homeassistant.components.greeneye_monitor.*
homeassistant.components.group.*
homeassistant.components.guardian.*
homeassistant.components.history.*

11
homeassistant/components/greeneye_monitor/__init__.py
View file

@ -1,5 +1,8 @@
"""Support for monitoring a GreenEye Monitor energy monitor."""
from __future__ import annotations
import logging
from typing import Any
from greeneye import Monitors
import voluptuous as vol
@ -15,8 +18,10 @@ from homeassistant.const import (
TIME_MINUTES,
TIME_SECONDS,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.typing import ConfigType
_LOGGER = logging.getLogger(__name__)
@ -117,7 +122,7 @@ COMPONENT_SCHEMA = vol.Schema(
CONFIG_SCHEMA = vol.Schema({DOMAIN: COMPONENT_SCHEMA}, extra=vol.ALLOW_EXTRA)
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the GreenEye Monitor component."""
monitors = Monitors()
hass.data[DATA_GREENEYE_MONITOR] = monitors
@ -125,7 +130,7 @@ async def async_setup(hass, config):
server_config = config[DOMAIN]
server = await monitors.start_server(server_config[CONF_PORT])
async def close_server(*args):
async def close_server(*args: list[Any]) -> None:
"""Close the monitoring server."""
await server.close()
@ -189,7 +194,7 @@ async def async_setup(hass, config):
return False
hass.async_create_task(
async_load_platform(hass, "sensor", DOMAIN, all_sensors, config)
async_load_platform(hass, "sensor", DOMAIN, {CONF_SENSORS: all_sensors}, config)
)
return True

134
homeassistant/components/greeneye_monitor/sensor.py
View file

@ -1,8 +1,16 @@
"""Support for the sensors in a GreenEye Monitor."""
from __future__ import annotations
from typing import Any, Generic, TypeVar
import greeneye
from greeneye import Monitors
from homeassistant.components.sensor import SensorEntity
from homeassistant.const import (
CONF_NAME,
CONF_SENSOR_TYPE,
CONF_SENSORS,
CONF_TEMPERATURE_UNIT,
DEVICE_CLASS_TEMPERATURE,
ELECTRIC_POTENTIAL_VOLT,
@ -11,6 +19,9 @@ from homeassistant.const import (
TIME_MINUTES,
TIME_SECONDS,
)
from homeassistant.core import Config, HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import DiscoveryInfoType
from . import (
CONF_COUNTED_QUANTITY,
@ -37,13 +48,15 @@ TEMPERATURE_ICON = "mdi:thermometer"
VOLTAGE_ICON = "mdi:current-ac"
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant,
config: Config,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType,
) -> None:
"""Set up a single GEM temperature sensor."""
if not discovery_info:
return
entities = []
for sensor in discovery_info:
entities: list[GEMSensor] = []
for sensor in discovery_info[CONF_SENSORS]:
sensor_type = sensor[CONF_SENSOR_TYPE]
if sensor_type == SENSOR_TYPE_CURRENT:
entities.append(
@ -86,42 +99,53 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
async_add_entities(entities)
class GEMSensor(SensorEntity):
T = TypeVar(
"T",
greeneye.monitor.Channel,
greeneye.monitor.Monitor,
greeneye.monitor.PulseCounter,
greeneye.monitor.TemperatureSensor,
)
class GEMSensor(Generic[T], SensorEntity):
"""Base class for GreenEye Monitor sensors."""
_attr_should_poll = False
def __init__(self, monitor_serial_number, name, sensor_type, number):
def __init__(
self, monitor_serial_number: int, name: str, sensor_type: str, number: int
) -> None:
"""Construct the entity."""
self._monitor_serial_number = monitor_serial_number
self._name = name
self._sensor = None
self._sensor: T | None = None
self._sensor_type = sensor_type
self._number = number
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique ID for this sensor."""
return f"{self._monitor_serial_number}-{self._sensor_type}-{self._number}"
@property
def name(self):
def name(self) -> str:
"""Return the name of the channel."""
return self._name
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Wait for and connect to the sensor."""
monitors = self.hass.data[DATA_GREENEYE_MONITOR]
if not self._try_connect_to_monitor(monitors):
monitors.add_listener(self._on_new_monitor)
def _on_new_monitor(self, *args):
def _on_new_monitor(self, *args: list[Any]) -> None:
monitors = self.hass.data[DATA_GREENEYE_MONITOR]
if self._try_connect_to_monitor(monitors):
monitors.remove_listener(self._on_new_monitor)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Remove listener from the sensor."""
if self._sensor:
self._sensor.remove_listener(self.async_write_ha_state)
@ -129,7 +153,7 @@ class GEMSensor(SensorEntity):
monitors = self.hass.data[DATA_GREENEYE_MONITOR]
monitors.remove_listener(self._on_new_monitor)
def _try_connect_to_monitor(self, monitors):
def _try_connect_to_monitor(self, monitors: Monitors) -> bool:
monitor = monitors.monitors.get(self._monitor_serial_number)
if not monitor:
return False
@ -140,34 +164,39 @@ class GEMSensor(SensorEntity):
return True
def _get_sensor(self, monitor):
def _get_sensor(self, monitor: greeneye.monitor.Monitor) -> T:
raise NotImplementedError()
class CurrentSensor(GEMSensor):
class CurrentSensor(GEMSensor[greeneye.monitor.Channel]):
"""Entity showing power usage on one channel of the monitor."""
_attr_icon = CURRENT_SENSOR_ICON
_attr_native_unit_of_measurement = UNIT_WATTS
def __init__(self, monitor_serial_number, number, name, net_metering):
def __init__(
self, monitor_serial_number: int, number: int, name: str, net_metering: bool
) -> None:
"""Construct the entity."""
super().__init__(monitor_serial_number, name, "current", number)
self._net_metering = net_metering
def _get_sensor(self, monitor):
def _get_sensor(
self, monitor: greeneye.monitor.Monitor
) -> greeneye.monitor.Channel:
return monitor.channels[self._number - 1]
@property
def native_value(self):
def native_value(self) -> float | None:
"""Return the current number of watts being used by the channel."""
if not self._sensor:
return None
assert isinstance(self._sensor.watts, float)
return self._sensor.watts
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return total wattseconds in the state dictionary."""
if not self._sensor:
return None
@ -180,43 +209,47 @@ class CurrentSensor(GEMSensor):
return {DATA_WATT_SECONDS: watt_seconds}
class PulseCounter(GEMSensor):
class PulseCounter(GEMSensor[greeneye.monitor.PulseCounter]):
"""Entity showing rate of change in one pulse counter of the monitor."""
_attr_icon = COUNTER_ICON
def __init__(
self,
monitor_serial_number,
number,
name,
counted_quantity,
time_unit,
counted_quantity_per_pulse,
):
monitor_serial_number: int,
number: int,
name: str,
counted_quantity: str,
time_unit: str,
counted_quantity_per_pulse: float,
) -> None:
"""Construct the entity."""
super().__init__(monitor_serial_number, name, "pulse", number)
self._counted_quantity = counted_quantity
self._counted_quantity_per_pulse = counted_quantity_per_pulse
self._time_unit = time_unit
def _get_sensor(self, monitor):
def _get_sensor(
self, monitor: greeneye.monitor.Monitor
) -> greeneye.monitor.PulseCounter:
return monitor.pulse_counters[self._number - 1]
@property
def native_value(self):
def native_value(self) -> float | None:
"""Return the current rate of change for the given pulse counter."""
if not self._sensor or self._sensor.pulses_per_second is None:
return None
return (
result = (
self._sensor.pulses_per_second
* self._counted_quantity_per_pulse
* self._seconds_per_time_unit
)
assert isinstance(result, float)
return result
@property
def _seconds_per_time_unit(self):
def _seconds_per_time_unit(self) -> int:
"""Return the number of seconds in the given display time unit."""
if self._time_unit == TIME_SECONDS:
return 1
@ -225,13 +258,18 @@ class PulseCounter(GEMSensor):
if self._time_unit == TIME_HOURS:
return 3600
# Config schema should have ensured it is one of the above values
raise Exception(
f"Invalid value for time unit: {self._time_unit}. Expected one of {TIME_SECONDS}, {TIME_MINUTES}, or {TIME_HOURS}"
)
@property
def native_unit_of_measurement(self):
def native_unit_of_measurement(self) -> str:
"""Return the unit of measurement for this pulse counter."""
return f"{self._counted_quantity}/{self._time_unit}"
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return total pulses in the data dictionary."""
if not self._sensor:
return None
@ -239,52 +277,60 @@ class PulseCounter(GEMSensor):
return {DATA_PULSES: self._sensor.pulses}
class TemperatureSensor(GEMSensor):
class TemperatureSensor(GEMSensor[greeneye.monitor.TemperatureSensor]):
"""Entity showing temperature from one temperature sensor."""
_attr_device_class = DEVICE_CLASS_TEMPERATURE
_attr_icon = TEMPERATURE_ICON
def __init__(self, monitor_serial_number, number, name, unit):
def __init__(
self, monitor_serial_number: int, number: int, name: str, unit: str
) -> None:
"""Construct the entity."""
super().__init__(monitor_serial_number, name, "temp", number)
self._unit = unit
def _get_sensor(self, monitor):
def _get_sensor(
self, monitor: greeneye.monitor.Monitor
) -> greeneye.monitor.TemperatureSensor:
return monitor.temperature_sensors[self._number - 1]
@property
def native_value(self):
def native_value(self) -> float | None:
"""Return the current temperature being reported by this sensor."""
if not self._sensor:
return None
assert isinstance(self._sensor.temperature, float)
return self._sensor.temperature
@property
def native_unit_of_measurement(self):
def native_unit_of_measurement(self) -> str:
"""Return the unit of measurement for this sensor (user specified)."""
return self._unit
class VoltageSensor(GEMSensor):
class VoltageSensor(GEMSensor[greeneye.monitor.Monitor]):
"""Entity showing voltage."""
_attr_icon = VOLTAGE_ICON
_attr_native_unit_of_measurement = ELECTRIC_POTENTIAL_VOLT
def __init__(self, monitor_serial_number, number, name):
def __init__(self, monitor_serial_number: int, number: int, name: str) -> None:
"""Construct the entity."""
super().__init__(monitor_serial_number, name, "volts", number)
def _get_sensor(self, monitor):
def _get_sensor(
self, monitor: greeneye.monitor.Monitor
) -> greeneye.monitor.Monitor:
"""Wire the updates to the monitor itself, since there is no voltage element in the API."""
return monitor
@property
def native_value(self):
def native_value(self) -> float | None:
"""Return the current voltage being reported by this sensor."""
if not self._sensor:
return None
assert isinstance(self._sensor.voltage, float)
return self._sensor.voltage

11
mypy.ini
View file

@ -583,6 +583,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.greeneye_monitor.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.group.*]
check_untyped_defs = true
disallow_incomplete_defs = true