From 6bdd3990fe8c5a56bf4545e486a38418c5c4aeca Mon Sep 17 00:00:00 2001 From: Harley Watson Date: Mon, 17 May 2021 22:29:18 +0100 Subject: [PATCH] address mypy errors --- .../hildebrandglow/config_flow.py | 14 ++--- custom_components/hildebrandglow/glow.py | 19 ++++--- .../hildebrandglow/mqttpayload.py | 54 +++++++++---------- custom_components/hildebrandglow/sensor.py | 13 +++-- setup.cfg | 3 ++ 5 files changed, 56 insertions(+), 47 deletions(-) diff --git a/custom_components/hildebrandglow/config_flow.py b/custom_components/hildebrandglow/config_flow.py index 05110f1..4d7be9a 100644 --- a/custom_components/hildebrandglow/config_flow.py +++ b/custom_components/hildebrandglow/config_flow.py @@ -1,4 +1,6 @@ """Config flow for Hildebrand Glow integration.""" +from __future__ import annotations + import logging from typing import Any, Dict @@ -19,8 +21,6 @@ def config_object(data: dict, glow: Dict[str, Any]) -> Dict[str, Any]: "name": glow["name"], "username": data["username"], "password": data["password"], - "token": glow["token"], - "token_exp": glow["exp"], } @@ -29,12 +29,12 @@ async def validate_input(hass: core.HomeAssistant, data: dict) -> Dict[str, Any] Data has the keys from DATA_SCHEMA with values provided by the user. """ - glow = await hass.async_add_executor_job( - Glow.authenticate, APP_ID, data["username"], data["password"] - ) + glow = Glow(APP_ID, data["username"], data["password"]) + + auth_data: Dict[str, Any] = await hass.async_add_executor_job(glow.authenticate) # Return some info we want to store in the config entry. - return config_object(data, glow) + return config_object(data, auth_data) class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): @@ -44,7 +44,7 @@ class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): CONNECTION_CLASS = config_entries.SOURCE_USER async def async_step_user( - self, user_input: Dict = None + self, user_input: dict[str, Any] | None = None ) -> data_entry_flow.FlowResult: """Handle the initial step.""" errors = {} diff --git a/custom_components/hildebrandglow/glow.py b/custom_components/hildebrandglow/glow.py index 4dc282c..eda0785 100644 --- a/custom_components/hildebrandglow/glow.py +++ b/custom_components/hildebrandglow/glow.py @@ -45,7 +45,7 @@ def __init__(self, app_id: str, username: str, password: str): self.broker_active = False - def authenticate(self) -> None: + def authenticate(self) -> Dict[str, Any]: """Attempt to authenticate with Glowmarkt.""" url = f"{self.BASE_URL}/auth" auth = {"username": self.username, "password": self.password} @@ -60,6 +60,7 @@ def authenticate(self) -> None: if data["valid"]: self.token = data["token"] + return data else: pprint(data) raise InvalidAuth @@ -87,7 +88,7 @@ def retrieve_cad_hardwareId(self) -> str: devices = self.retrieve_devices() cad: Dict[str, Any] = next( - (dev for dev in devices if dev["deviceTypeId"] == ZIGBEE_GLOW_STICK), None + (dev for dev in devices if dev["deviceTypeId"] == ZIGBEE_GLOW_STICK), {} ) self.hardwareId = cad["hardwareId"] @@ -100,7 +101,9 @@ def connect_mqtt(self) -> None: self.broker.loop_start() - def _cb_on_connect(self, client, userdata, flags, rc): + def _cb_on_connect( + self, client: mqtt, userdata: Any, flags: Dict[str, Any], rc: int + ) -> None: """Receive a CONNACK message from the server.""" client.subscribe( [ @@ -114,11 +117,13 @@ def _cb_on_connect(self, client, userdata, flags, rc): self.broker_active = True - def _cb_on_disconnect(self, client, userdata, rc): + def _cb_on_disconnect(self, client: mqtt, userdata: Any, rc: int) -> None: """Receive notice the MQTT connection has disconnected.""" self.broker_active = False - def _cb_on_message(self, client, userdata, msg): + def _cb_on_message( + self, client: mqtt, userdata: Any, msg: mqtt.MQTTMessage + ) -> None: """Receive a PUBLISH message from the server.""" payload = MQTTPayload(msg.payload) @@ -160,7 +165,9 @@ def current_usage(self, resource: Dict[str, Any]) -> Dict[str, Any]: data = response.json() return data - def register_sensor(self, sensor, resource): + def register_sensor( + self, sensor: GlowConsumptionCurrent, resource: Dict[str, Any] + ) -> None: """Register a live sensor for dispatching MQTT messages.""" self.sensors[resource["classifier"]] = sensor diff --git a/custom_components/hildebrandglow/mqttpayload.py b/custom_components/hildebrandglow/mqttpayload.py index 70bd4a4..3942ae5 100644 --- a/custom_components/hildebrandglow/mqttpayload.py +++ b/custom_components/hildebrandglow/mqttpayload.py @@ -1,7 +1,7 @@ """Helper classes for Zigbee Smart Energy Profile data.""" import json from enum import Enum -from typing import Any, Dict +from typing import Any, Dict, Optional class Meter: @@ -17,19 +17,19 @@ class SupplyStatus(Enum): ARMED = "01" ON = "02" - current_summation_delivered: int + current_summation_delivered: Optional[int] """Import energy usage""" - current_summation_received: int + current_summation_received: Optional[int] """Export energy usage""" - current_max_demand_delivered: int + current_max_demand_delivered: Optional[int] """Maximum import energy usage rate""" - reading_snapshot_time: str + reading_snapshot_time: Optional[int] """Last time all of the reported attributed were updated""" - supply_status: SupplyStatus + supply_status: Optional[SupplyStatus] """Current state of the meter's supply.""" def __init__(self, payload: Dict[str, Any]): @@ -65,7 +65,7 @@ def __init__(self, payload: Dict[str, Any]): class MeterStatus: """Information about the meter's error conditions.""" - status: str + status: Optional[str] """Meter error conditions""" def __init__(self, payload: Dict[str, Any]): @@ -89,31 +89,31 @@ class MeteringDeviceType(Enum): ELECTRIC = "00" GAS = "80" - unit_of_measure: UnitofMeasure + unit_of_measure: Optional[UnitofMeasure] """Unit for the measured value.""" - multiplier: int + multiplier: Optional[int] """Multiplier value for smart meter readings.""" - divisor: int + divisor: Optional[int] """Divisor value for smart meter readings.""" - summation_formatting: str + summation_formatting: Optional[str] """Bitmap representing decimal places in Summation readings.""" - demand_formatting: str + demand_formatting: Optional[str] """Bitmap representing decimal places in Demand readings.""" - metering_device_type: MeteringDeviceType + metering_device_type: Optional[MeteringDeviceType] """Smart meter device type.""" - siteID: str + siteID: Optional[str] """Electricicity MPAN / Gas MPRN.""" - meter_serial_number: str + meter_serial_number: Optional[str] """Smart meter serial number.""" - alternative_unit_of_measure: UnitofMeasure + alternative_unit_of_measure: Optional[UnitofMeasure] """Alternative unit for the measured value.""" def __init__(self, payload: Dict[str, Any]): @@ -139,16 +139,16 @@ def __init__(self, payload: Dict[str, Any]): class HistoricalConsumption: """Information about the meter's historical readings.""" - instantaneous_demand: int + instantaneous_demand: Optional[int] """Instantaneous import energy usage rate""" - current_day_consumption_delivered: int + current_day_consumption_delivered: Optional[int] """Import energy used in the current day.""" - current_week_consumption_delivered: int + current_week_consumption_delivered: Optional[int] """Import energy used in the current week.""" - current_month_consumption_delivered: int + current_month_consumption_delivered: Optional[int] """Import energy used in the current month.""" def __init__(self, payload: Dict[str, Any]): @@ -181,13 +181,13 @@ def __init__(self, payload: Dict[str, Any]): class AlternativeHistoricalConsumption: """Information about the meter's altenative historical readings.""" - current_day_consumption_delivered: int + current_day_consumption_delivered: Optional[int] """Import energy used in the current day.""" - current_week_consumption_delivered: int + current_week_consumption_delivered: Optional[int] """Import energy used in the current week.""" - current_month_consumption_delivered: int + current_month_consumption_delivered: Optional[int] """Import energy used in the current month.""" def __init__(self, payload: Dict[str, Any]): @@ -232,15 +232,15 @@ def __init__(self, payload: Dict[str, Any]): class MQTTPayload: """Object representing a payload received over MQTT.""" - electricity: Meter + electricity: Optional[Meter] """Data interpreted from an electricity meter.""" - gas: Meter + gas: Optional[Meter] """Data interpreted from a gas meter.""" - def __init__(self, payload: str): + def __init__(self, input: str): """Create internal Meter instances based off the unprocessed payload.""" - payload = json.loads(payload) + payload: Dict[str, Any] = json.loads(input) self.electricity = ( Meter(payload["elecMtr"]) if "03" in payload["elecMtr"]["0702"] else None ) diff --git a/custom_components/hildebrandglow/sensor.py b/custom_components/hildebrandglow/sensor.py index b493e63..683487a 100644 --- a/custom_components/hildebrandglow/sensor.py +++ b/custom_components/hildebrandglow/sensor.py @@ -8,7 +8,7 @@ from .const import DOMAIN from .glow import Glow, InvalidAuth -from .mqttpayload import Meter +from .mqttpayload import Meter, MQTTPayload async def async_setup_entry( @@ -89,7 +89,7 @@ def device_info(self) -> Optional[Dict[str, Any]]: } @property - def state(self) -> Optional[str]: + def state(self) -> Optional[int]: """Return the state of the sensor.""" if self._state: if self.resource["dataSourceResourceTypeInfo"]["type"] == "ELEC": @@ -97,10 +97,9 @@ def state(self) -> Optional[str]: elif self.resource["dataSourceResourceTypeInfo"]["type"] == "GAS": alt = self._state.alternative_historical_consumption return alt.current_day_consumption_delivered - else: - return None + return None - def update_state(self, meter) -> None: + def update_state(self, meter: MQTTPayload) -> None: """Receive an MQTT update from Glow and update the internal state.""" self._state = meter.electricity self.async_write_ha_state() @@ -118,5 +117,5 @@ def unit_of_measurement(self) -> Optional[str]: return POWER_WATT elif self.resource["dataSourceResourceTypeInfo"]["type"] == "GAS": return VOLUME_CUBIC_METERS - else: - return None + + return None diff --git a/setup.cfg b/setup.cfg index bc86215..2adbb5a 100644 --- a/setup.cfg +++ b/setup.cfg @@ -22,3 +22,6 @@ ignore_errors = True [mypy-voluptuous] ignore_missing_imports = True + +[mypy-paho.*] +ignore_missing_imports = True