Merge pull request #20 from dontinelli/fyta-models

Add models to enable strict typing

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fyta_cli"
-version = "0.5.1"
+version = "0.6.0"
 authors = [
   { name="dontinelli", email="[email protected]" },
 ]

requirements.txt

@@ -1 +1,2 @@
-aiohttp==3.10.2
+aiohttp==3.10.2
+mashumaro>=3.13

src/fyta_cli/fyta_client.py

@@ -18,6 +18,7 @@
     FytaPasswordError,
     FytaPlantError,
 )
+from .fyta_models import Credentials
 
 FYTA_AUTH_URL = "https://web.fyta.de/api/auth/login"
 FYTA_PLANT_URL = "https://web.fyta.de/api/user-plant"
@@ -63,7 +64,7 @@ async def test_connection(self) -> bool:
 
         return False
 
-    async def login(self) -> dict[str, str | datetime]:
+    async def login(self) -> Credentials:
         """Handle a request to FYTA."""
 
         if (
@@ -108,7 +109,7 @@ async def login(self) -> dict[str, str | datetime]:
             seconds=int(json_response["expires_in"])
         )
 
-        return {"access_token": self.access_token, "expiration": self.expiration}
+        return Credentials(access_token = self.access_token, expiration = self.expiration)
 
     async def get_plants(self) -> dict[int, str]:
         """Get a list of all available plants from FYTA"""

src/fyta_cli/fyta_connector.py

@@ -5,24 +5,7 @@
 from zoneinfo import ZoneInfo
 
 from .fyta_client import Client
-from .utils import safe_get
-
-# status = 0 if user_plant deleted, 1 for user_plant in green, 2 for yellow, 3 for red
-PLANT_STATUS = {
-    0: "deleted",
-    1: "doing_great",
-    2: "need_attention",
-    3: "no_sensor",
-}
-
-PLANT_MEASUREMENT_STATUS = {
-    0: "no_data",
-    1: "too_low",
-    2: "low",
-    3: "perfect",
-    4: "high",
-    5: "too_high",
-}
+from .fyta_models import Credentials, Plant
 
 
 class FytaConnector:
@@ -58,7 +41,7 @@ async def test_connection(self) -> bool:
 
         return await self.client.test_connection()
 
-    async def login(self) -> dict[str, str | datetime]:
+    async def login(self) -> Credentials:
         """Login with credentials to get access token."""
 
         login = await self.client.login()
@@ -75,104 +58,34 @@ async def update_plant_list(self) -> dict[int, str]:
 
         return self.plant_list
 
-    async def update_all_plants(self) -> dict[int, dict[str, Any]]:
+    async def update_all_plants(self) -> dict[int, Plant]:
         """Get data of all available plants."""
 
-        plants: dict[int, dict[str, Any]] = {}
+        plants: dict[int, Plant] = {}
 
-        plant_list = await self.update_plant_list()
+        plant_list: dict[int, str] = await self.update_plant_list()
 
         for plant in plant_list:
-            current_plant = await self.update_plant_data(plant)
-            if current_plant != {}:
-                plants |= {plant: current_plant}
+            current_plant: Plant = await self.update_plant_data(plant)
+            plants |= {plant: current_plant}
 
         self.plants = plants
 
         return plants
 
-    async def update_plant_data(self, plant_id: int) -> dict[str, Any]:
+    async def update_plant_data(self, plant_id: int) -> Plant:
         """Get data of specific plant."""
 
         p: dict = await self.client.get_plant_data(plant_id)
 
-        current_plant = {}
+        current_plant = Plant
 
         if ("plant" not in p) or (p["plant"]["sensor"] is None):
-            current_plant |= {"sensor_available": False}
+            current_plant.sensor_available = False
         else:
             plant_data: dict = p["plant"]
-            current_plant |= {"online": True}
-            current_plant |= {"sensor_available": True}
-            current_plant |= {
-                "battery_status:": safe_get(plant_data, "sensor.is_battery_low", bool)
-            }
-            current_plant |= {"sw_version": safe_get(plant_data, "sensor.version", str)}
-            current_plant |= {"plant_id": safe_get(plant_data, "plant_id", int)}
-            current_plant |= {"name": safe_get(plant_data, "nickname", str)}
-            current_plant |= {
-                "scientific_name": safe_get(plant_data, "scientific_name", str)
-            }
-            current_plant |= {"status": safe_get(plant_data, "status", int)}
-            current_plant |= {
-                "plant_thumb_path": safe_get(plant_data, "plant_thumb_path", str)
-            }
-            current_plant |= {
-                "plant_origin_path": safe_get(plant_data, "plant_origin_path", str)
-            }
-            current_plant |= {
-                "temperature_status": safe_get(
-                    plant_data, "measurements.temperature.status", int
-                )
-            }
-            current_plant |= {
-                "light_status": safe_get(plant_data, "measurements.light.status", int)
-            }
-            current_plant |= {
-                "moisture_status": safe_get(
-                    plant_data, "measurements.moisture.status", int
-                )
-            }
-            current_plant |= {
-                "salinity_status": safe_get(
-                    plant_data, "measurements.salinity.status", int
-                )
-            }
-            current_plant |= {
-                "ph": safe_get(plant_data, "measurements.ph.values.current", float)
-            }
-            current_plant |= {
-                "temperature": safe_get(
-                    plant_data, "measurements.temperature.values.current", float
-                )
-            }
-            current_plant |= {
-                "light": safe_get(
-                    plant_data, "measurements.light.values.current", float
-                )
-            }
-            current_plant |= {"light_dli": safe_get(plant_data, "dli_light", float)}
-            current_plant |= {
-                "moisture": safe_get(
-                    plant_data, "measurements.moisture.values.current", float
-                )
-            }
-            current_plant |= {
-                "salinity": safe_get(
-                    plant_data, "measurements.salinity.values.current", float
-                )
-            }
-            current_plant |= {
-                "battery_level": safe_get(plant_data, "measurements.battery", float)
-            }
-            current_plant |= {
-                "last_updated": safe_get(
-                    plant_data,
-                    "sensor.received_data_at",
-                    datetime,
-                    self.client.timezone,
-                )
-            }
+
+            current_plant = Plant.from_dict(plant_data)
 
         return current_plant
 

src/fyta_cli/fyta_models.py

@@ -0,0 +1,87 @@
+"""Models for FYTA."""
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import IntEnum
+from typing import Any
+
+from mashumaro import DataClassDictMixin, field_options
+
+
+@dataclass
+class Credentials():
+    """Fyta login model."""
+
+    access_token: str
+    expiration: datetime
+
+
+class PlantStatus(IntEnum):
+    """Plant status enum."""
+    DELETED = 0
+    DOING_GREAT = 1
+    NEED_ATTENTION = 2
+    NO_SENSOR = 3
+
+
+class PlantMeasurementStatus(IntEnum):
+    """Plant measurement status enum."""
+    NO_DATA = 0
+    TOO_LOW = 1
+    LOW = 2
+    PERFECT = 3
+    HIGH = 4
+    TOO_HIGH = 5
+
+
+@dataclass
+class Plant(DataClassDictMixin):
+    """Plant model."""
+
+    # pylint: disable=too-many-instance-attributes
+
+    battery_level: float
+    battery_status: bool
+    last_updated: datetime
+    light: float
+    light_status: PlantMeasurementStatus
+    name: str = field(metadata=field_options(alias="nickname"))
+    moisture: float
+    moisture_status: PlantMeasurementStatus
+    sensor_available: bool
+    sw_version: str
+    status: PlantStatus
+    online: bool
+    ph: float | None
+    plant_id: int
+    plant_origin_path: str
+    plant_thumb_path: str
+    salinity: float
+    salinity_status: PlantMeasurementStatus
+    scientific_name: str
+    temperature: float
+    temperature_status: PlantMeasurementStatus
+
+    @classmethod
+    def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]:
+
+        d |= {"sensor_available": True}
+        d |= {"online": True}
+
+        if d.get("measurements") is not None:
+            d |= {"battery_level": d["measurements"]["battery"]}
+            d |= {"light": d["measurements"]["light"]["values"]["current"]}
+            d |= {"light_status": d["measurements"]["light"].get("status")}
+            d |= {"moisture": d["measurements"]["moisture"]["values"]["current"]}
+            d |= {"moisture_status": d["measurements"]["moisture"].get("status")}
+            d |= {"ph": d["measurements"].get("ph").get("values").get("current")}
+            d |= {"salinity": d["measurements"]["salinity"]["values"]["current"]}
+            d |= {"salinity_status": d["measurements"]["salinity"].get("status")}
+            d |= {"temperature": d["measurements"]["temperature"]["values"]["current"]}
+            d |= {"temperature_status": d["measurements"]["temperature"].get("status")}
+
+        if d.get("sensor") is not None:
+            d |= {"battery_status": d["sensor"]["is_battery_low"]}
+            d |= {"last_updated": d["sensor"]["received_data_at"]}
+            d |= {"sw_version": d["sensor"]["version"]}
+
+        return d