Use strict data model checks with pydantic

.pre-commit-config.yaml

@@ -7,6 +7,7 @@ repos:
       - id: end-of-file-fixer
       - id: mixed-line-ending
         args: ["--fix=lf"]
+
   # - repo: https://github.com/astral-sh/ruff-pre-commit
   #   rev: v0.0.284
   #   hooks:
@@ -22,3 +23,12 @@ repos:
     rev: 5.12.0
     hooks:
     - id: isort
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.5.1
+    hooks:
+      - id: mypy
+        language_version: python
+        additional_dependencies:
+        - pydantic~=2.0
+        - types-pytz

custom_components/eyeonwater/binary_sensor.py

@@ -1,4 +1,6 @@
 """Support for EyeOnWater binary sensors."""
+from dataclasses import dataclass
+
 from pyonwater import Meter
 
 from homeassistant.components.binary_sensor import (
@@ -16,38 +18,56 @@
 
 from .const import DATA_COORDINATOR, DATA_SMART_METER, DOMAIN
 
+
+@dataclass
+class Description:
+    name: str
+    key: str
+    translation_key: str
+    device_class: BinarySensorDeviceClass
+
+
 FLAG_SENSORS = [
-    BinarySensorEntityDescription(
-        key="Leak",
+    Description(
+        name="Leak",
+        key="leak",
         translation_key="leak",
         device_class=BinarySensorDeviceClass.MOISTURE,
     ),
-    BinarySensorEntityDescription(
-        key="EmptyPipe",
+    Description(
+        name="EmptyPipe",
+        key="empty_pipe",
         translation_key="emptypipe",
         device_class=BinarySensorDeviceClass.PROBLEM,
     ),
-    BinarySensorEntityDescription(
-        key="Tamper",
+    Description(
+        name="Tamper",
+        key="tamper",
         translation_key="tamper",
         device_class=BinarySensorDeviceClass.TAMPER,
     ),
-    BinarySensorEntityDescription(
-        key="CoverRemoved",
+    Description(
+        name="CoverRemoved",
+        key="cover_removed",
         translation_key="coverremoved",
         device_class=BinarySensorDeviceClass.TAMPER,
     ),
-    BinarySensorEntityDescription(
-        key="ReverseFlow",
+    Description(
+        name="ReverseFlow",
+        key="reverse_flow",
         translation_key="reverseflow",
         device_class=BinarySensorDeviceClass.PROBLEM,
     ),
-    BinarySensorEntityDescription(
-        key="LowBattery",
+    Description(
+        name="LowBattery",
+        key="low_battery",
+        translation_key="lowbattery",
         device_class=BinarySensorDeviceClass.BATTERY,
     ),
-    BinarySensorEntityDescription(
-        key="BatteryCharging",
+    Description(
+        name="BatteryCharging",
+        key="battery_charging",
+        translation_key="batterycharging",
         device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
     ),
 ]
@@ -75,27 +95,34 @@ def __init__(
         self,
         meter: Meter,
         coordinator: DataUpdateCoordinator,
-        description: BinarySensorEntityDescription,
+        description: Description,
     ) -> None:
         """Initialize the sensor."""
         super().__init__(coordinator)
-        self.entity_description = description
+        self.entity_description = BinarySensorEntityDescription(
+            key=description.key,
+            device_class=description.device_class,
+            translation_key=description.translation_key,
+        )
         self.meter = meter
         self._state = False
         self._available = False
-        self._attr_unique_id = f"{description.key}_{self.meter.meter_uuid}"
+        self._attr_unique_id = f"{description.name}_{self.meter.meter_uuid}"
         self._attr_is_on = self._state
         self._attr_device_info = DeviceInfo(
             identifiers={(DOMAIN, self.meter.meter_uuid)},
             name=f"Water Meter {self.meter.meter_id}",
         )
 
+    def get_flag(self) -> bool:
+        return self.meter.meter_info.reading.flags.__dict__[self.entity_description.key]
+
     @callback
     def _state_update(self):
         """Call when the coordinator has an update."""
         self._available = self.coordinator.last_update_success
         if self._available:
-            self._state = self.meter.get_flags(self.entity_description.key)
+            self._state = self.get_flag()
         self.async_write_ha_state()
 
     async def async_added_to_hass(self):

custom_components/eyeonwater/config_flow.py

@@ -78,7 +78,7 @@ async def validate_input(hass: core.HomeAssistant, data):
     return {"title": account.username}
 
 
-class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):  # type: ignore
     """Handle a config flow for EyeOnWater."""
 
     VERSION = 1

custom_components/eyeonwater/manifest.json

@@ -6,6 +6,6 @@
   "dependencies": ["recorder"],
   "documentation": "https://github.com/kdeyev/eyeonwater",
   "iot_class": "cloud_polling",
-  "requirements": ["pyonwater==0.1.2"],
-  "version": "2.0.2"
+  "requirements": ["pyonwater==0.2.4"],
+  "version": "2.0.4"
 }

custom_components/eyeonwater/sensor.py

@@ -1,8 +1,9 @@
 """Support for EyeOnWater sensors."""
 import datetime
 import logging
+from typing import Any
 
-from pyonwater import Meter
+from pyonwater import DataPoint, Meter
 import pytz
 
 from homeassistant.components.recorder import get_instance
@@ -30,7 +31,7 @@
 _LOGGER.addHandler(logging.StreamHandler())
 
 
-def get_statistics_id(meter) -> str:
+def get_statistics_id(meter: Meter) -> str:
     return f"sensor.water_meter_{meter.meter_id}"
 
 
@@ -80,7 +81,10 @@ class EyeOnWaterSensor(CoordinatorEntity, SensorEntity):
     # _attr_state_class = SensorStateClass.TOTAL_INCREASING
 
     def __init__(
-        self, meter: Meter, last_imported_time, coordinator: DataUpdateCoordinator
+        self,
+        meter: Meter,
+        last_imported_time: datetime.datetime,
+        coordinator: DataUpdateCoordinator,
     ) -> None:
         """Initialize the sensor."""
         super().__init__(coordinator)
@@ -93,7 +97,7 @@ def __init__(
             identifiers={(DOMAIN, self.meter.meter_uuid)},
             name=f"{WATER_METER_NAME} {self.meter.meter_id}",
         )
-        self._last_historical_data = []
+        self._last_historical_data: list[DataPoint] = []
         self._last_imported_time = last_imported_time
 
     @property
@@ -107,9 +111,9 @@ def native_value(self):
         return self._state
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> dict[str, Any]:
         """Return the device specific state attributes."""
-        return self.meter.attributes["register_0"]
+        return self.meter.meter_info.reading.dict()
 
     @callback
     def _state_update(self):
@@ -121,11 +125,11 @@ def _state_update(self):
             self._last_historical_data = self.meter.last_historical_data.copy()
             if self._last_imported_time and self._last_historical_data:
                 _LOGGER.info(
-                    f"_last_imported_time {self._last_imported_time} - self._last_historical_data {self._last_historical_data[-1]['dt']}"
+                    f"_last_imported_time {self._last_imported_time} - self._last_historical_data {self._last_historical_data[-1].dt}"
                 )
                 self._last_historical_data = list(
                     filter(
-                        lambda r: r["dt"] > self._last_imported_time,
+                        lambda r: r.dt > self._last_imported_time,
                         self._last_historical_data,
                     )
                 )
@@ -136,7 +140,7 @@ def _state_update(self):
             if self._last_historical_data:
                 self.import_historical_data()
 
-                self._last_imported_time = self._last_historical_data[-1]["dt"]
+                self._last_imported_time = self._last_historical_data[-1].dt
 
         self.async_write_ha_state()
 
@@ -163,9 +167,9 @@ def import_historical_data(self):
 
         statistics = [
             StatisticData(
-                start=row["dt"],
-                sum=row["reading"],
-                state=row["reading"],
+                start=row.dt,
+                sum=row.reading,
+                state=row.reading,
             )
             for row in self._last_historical_data
         ]

mypy.ini

@@ -0,0 +1,11 @@
+[tool.mypy]
+plugins = ["pydantic.mypy"]
+strict = true
+disallow_untyped_calls = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+no_implicit_optional = true
+show_error_codes = true
+implicit_reexport = true