Move casting to event listener and add OverkizStateType (#305)

* Add casting to event listener

* Remove duplicate test assert

* Bugfixes

* Make event casting simpler

Author: iMicknl

pyoverkiz/models.py

@@ -19,6 +19,7 @@
     UIWidget,
     UpdateBoxStatus,
 )
+from pyoverkiz.types import DATA_TYPE_TO_PYTHON, StateType
 
 # pylint: disable=unused-argument, too-many-instance-attributes, too-many-locals
 
@@ -269,20 +270,37 @@ def __len__(self) -> int:
 class State:
     name: str
     type: DataType
-    value: None | int | float | str | bool
+    value: StateType
 
     def __init__(
         self,
         name: str,
         type: int,
-        value: None | int | float | str | bool = None,
+        value: StateType = None,
         **_: Any,
     ):
         self.name = name
         self.value = value
         self.type = DataType(type)
 
 
+@define(init=False, kw_only=True)
+class EventState(State):
+    def __init__(
+        self,
+        name: str,
+        type: int,
+        value: str | None = None,
+        **_: Any,
+    ):
+        super().__init__(name, type, value, **_)
+
+        # Overkiz returns all state values as a string
+        # We cast them here based on the data type provided by Overkiz
+        if self.type in DATA_TYPE_TO_PYTHON:
+            self.value = DATA_TYPE_TO_PYTHON[self.type](self.value)
+
+
 @define(init=False)
 class States:
     _states: list[State]
@@ -351,7 +369,7 @@ class Event:
     gateway_id: str | None = field(repr=obfuscate_id, default=None)
     exec_id: str | None = None
     device_url: str | None = field(repr=obfuscate_id, default=None)
-    device_states: list[State]
+    device_states: list[EventState]
     old_state: ExecutionState | None = None
     new_state: ExecutionState | None = None
 
@@ -387,7 +405,7 @@ def __init__(
         self.exec_id = exec_id
         self.device_url = device_url
         self.device_states = (
-            [State(**s) for s in device_states] if device_states else []
+            [EventState(**s) for s in device_states] if device_states else []
         )
         self.old_state = ExecutionState(old_state) if old_state else None
         self.new_state = ExecutionState(new_state) if new_state else None

pyoverkiz/types.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Callable, Dict, List, Union
+
+from pyoverkiz.enums import DataType
+
+StateType = Union[str, int, float, bool, Dict[Any, Any], List[Any], None]
+
+
+DATA_TYPE_TO_PYTHON: dict[DataType, Callable[[Any], StateType]] = {
+    DataType.INTEGER: int,
+    DataType.DATE: int,
+    DataType.FLOAT: float,
+    DataType.BOOLEAN: bool,
+    DataType.JSON_ARRAY: json.loads,
+    DataType.JSON_OBJECT: json.loads,
+}

tests/test_client.py

@@ -8,6 +8,7 @@
 
 from pyoverkiz.client import OverkizClient
 from pyoverkiz.const import SUPPORTED_SERVERS
+from pyoverkiz.enums import DataType
 
 CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
 
@@ -39,6 +40,56 @@ async def test_fetch_events_basic(self, client):
             events = await client.fetch_events()
             assert len(events) == 16
 
+    @pytest.mark.asyncio
+    async def test_fetch_events_simple_cast(self, client):
+        with open(
+            os.path.join(CURRENT_DIR, "events.json"), encoding="utf-8"
+        ) as raw_events:
+            resp = MockResponse(raw_events.read())
+
+        with patch.object(aiohttp.ClientSession, "post", return_value=resp):
+            events = await client.fetch_events()
+
+            # check if str to integer cast was succesfull
+            int_state_event = events[2].device_states[0]
+            assert int_state_event.value == 23247220
+            assert isinstance(int_state_event.value, int)
+            assert int_state_event.type == DataType.INTEGER
+
+    @pytest.mark.asyncio
+    async def test_fetch_events_casting(self, client):
+        with open(
+            os.path.join(CURRENT_DIR, "events.json"), encoding="utf-8"
+        ) as raw_events:
+            resp = MockResponse(raw_events.read())
+
+        with patch.object(aiohttp.ClientSession, "post", return_value=resp):
+            events = await client.fetch_events()
+
+            for event in events:
+                for state in event.device_states:
+
+                    if state.type == 0:
+                        assert state.value is None
+
+                    if state.type == 1:
+                        assert isinstance(state.value, int)
+
+                    if state.type == 2:
+                        assert isinstance(state.value, float)
+
+                    if state.type == 3:
+                        assert isinstance(state.value, str)
+
+                    if state.type == 6:
+                        assert isinstance(state.value, bool)
+
+                    if state.type == 10:
+                        assert isinstance(state.value, list)
+
+                    if state.type == 11:
+                        assert isinstance(state.value, dict)
+
     @pytest.mark.parametrize(
         "fixture_name, device_count",
         [