address mypy errors

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 = {}

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
 

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
         )

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,18 +89,17 @@ 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":
                 return self._state.historical_consumption.instantaneous_demand
             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

setup.cfg

@@ -22,3 +22,6 @@ ignore_errors = True
 
 [mypy-voluptuous]
 ignore_missing_imports = True
+
+[mypy-paho.*]
+ignore_missing_imports = True