Feature/access doors

PyMyGekko/__init__.py

@@ -5,6 +5,8 @@
 from typing import Union
 
 from aiohttp import ClientSession
+from PyMyGekko.resources.AccessDoors import AccessDoor
+from PyMyGekko.resources.AccessDoors import AccessDoorValueAccessor
 from PyMyGekko.resources.Actions import Action
 from PyMyGekko.resources.Actions import ActionValueAccessor
 from PyMyGekko.resources.AlarmsLogics import AlarmsLogic
@@ -23,8 +25,8 @@
 from PyMyGekko.resources.Meteo import MeteoValueAccessor
 from PyMyGekko.resources.RoomTemps import RoomTemp
 from PyMyGekko.resources.RoomTemps import RoomTempsValueAccessor
-from PyMyGekko.resources.vents import Vent
-from PyMyGekko.resources.vents import VentValueAccessor
+from PyMyGekko.resources.Vents import Vent
+from PyMyGekko.resources.Vents import VentValueAccessor
 from yarl import URL
 
 from .data_provider import DataProvider
@@ -62,6 +64,7 @@ def __init__(
                 self._url, self._authentication_params, self._session
             )
 
+        self._access_doors_value_accessor = AccessDoorValueAccessor(self._data_provider)
         self._actions_value_accessor = ActionValueAccessor(self._data_provider)
         self._alarm_logics_value_accessor = AlarmsLogicValueAccessor(
             self._data_provider
@@ -71,11 +74,11 @@ def __init__(
         self._hot_water_systems_value_accessor = HotWaterSystemValueAccessor(
             self._data_provider
         )
+        self._meteo_value_accessor = MeteoValueAccessor(self._data_provider)
         self._light_value_accessor = LightValueAccessor(self._data_provider)
         self._loads_value_accessor = LoadValueAccessor(self._data_provider)
         self._room_temps_value_accessor = RoomTempsValueAccessor(self._data_provider)
         self._vents_value_accessor = VentValueAccessor(self._data_provider)
-        self._meteo_value_accessor = MeteoValueAccessor(self._data_provider)
 
     async def try_connect(self) -> None:
         """Tries to connect to the MyGekko API using the given credentials"""
@@ -101,6 +104,10 @@ def get_globals_network(self):
 
         return result
 
+    def get_access_doors(self) -> list[AccessDoor]:
+        """Returns the MyGekko access doors"""
+        return self._access_doors_value_accessor.access_doors
+
     def get_actions(self) -> list[Action]:
         """Returns the MyGekko actions"""
         return self._actions_value_accessor.actions
@@ -129,6 +136,10 @@ def get_loads(self) -> list[Load]:
         """Returns the MyGekko load"""
         return self._loads_value_accessor.loads
 
+    def get_meteo(self) -> Meteo:
+        """Returns the MyGekko meteo"""
+        return self._meteo_value_accessor.meteo
+
     def get_room_temps(self) -> list[RoomTemp]:
         """Returns the MyGekko room_temps"""
         return self._room_temps_value_accessor.room_temps
@@ -137,10 +148,6 @@ def get_vents(self) -> list[Vent]:
         """Returns the MyGekko vents"""
         return self._vents_value_accessor.vents
 
-    def get_meteo(self) -> Meteo:
-        """Returns the MyGekko meteo"""
-        return self._meteo_value_accessor.meteo
-
 
 class MyGekkoQueryApiClient(MyGekkoApiClientBase):
     """The api client to access MyGekko via the MyGekko query api"""

PyMyGekko/resources/AccessDoors.py

@@ -0,0 +1,169 @@
+"""MyGekko AccessDoors implementation"""
+from __future__ import annotations
+
+from enum import IntEnum
+
+from PyMyGekko.data_provider import DataProviderBase
+from PyMyGekko.data_provider import EntityValueAccessor
+from PyMyGekko.resources import Entity
+
+
+class AccessDoor(Entity):
+    """Class for MyGekko AccessDoor"""
+
+    def __init__(
+        self, entity_id: str, name: str, value_accessor: AccessDoorValueAccessor
+    ) -> None:
+        super().__init__(entity_id, name, "/accessdoors/")
+        self._value_accessor = value_accessor
+        self._supported_features = self._value_accessor.get_features(self)
+
+    @property
+    def supported_features(self) -> list[AccessDoorFeature]:
+        """Returns the supported features"""
+        return self._supported_features
+
+    @property
+    def access_state(self) -> AccessDoorAccessState | None:
+        """Returns the current access state"""
+        value = self._value_accessor.get_value(self, "accessState")
+        return AccessDoorAccessState(int(value)) if value is not None else None
+
+    @property
+    def access_type(self) -> AccessDoorAccessType | None:
+        """Returns the current access type"""
+        value = self._value_accessor.get_value(self, "accessType")
+        return AccessDoorAccessType(int(value)) if value is not None else None
+
+    async def set_state(self, state: AccessDoorCommand):
+        """Sets the state"""
+        await self._value_accessor.set_state(self, state)
+
+    @property
+    def element_info(self) -> AccessDoorElementInfo | None:
+        """Returns the element info"""
+        value = self._value_accessor.get_value(self, "elementInfo")
+        return AccessDoorElementInfo(float(value)) if value is not None else None
+
+
+class AccessDoorAccessState(IntEnum):
+    """MyGekko AccessDoors Access State"""
+
+    CLOSED = 0
+    OPEN = 1
+
+
+class AccessDoorAccessType(IntEnum):
+    """MyGekko AccessDoors State"""
+
+    DOOR = 0
+    GATE_CONTROL = 1
+    BARRIER_CONTROL = 2
+    GATE_OPERATION = 3
+    MYGEKKO_NET_REMOTE_DOOR = 10
+
+
+class AccessDoorState(IntEnum):
+    """MyGekko AccessDoors State"""
+
+    CLOSE = 0
+    OPEN = 1
+    HOLD_DOWN = 2
+    PARTIALLY_OPEN = 3
+    PARTIALLY_HOLD_OPEN = 4
+
+
+class AccessDoorCommand(IntEnum):
+    """MyGekko AccessDoors Command"""
+
+    STOP = -3
+    CLOSE = -2
+    LOCK = -1
+    OPEN = 1
+    HOLD_OPEN = 2
+
+
+class AccessDoorFeature(IntEnum):
+    """MyGekko AccessDoors Feature"""
+
+    OPEN = 0
+
+
+class AccessDoorElementInfo(IntEnum):
+    """MyGekko AccessDoors Element Info"""
+
+    OK = 0
+    MANUAL_OFF = 1
+    MANUAL_ON = 2
+    LOCKED = 3
+    ALARM = 4
+
+
+class AccessDoorValueAccessor(EntityValueAccessor):
+    """AccessDoor value accessor"""
+
+    def __init__(self, data_provider: DataProviderBase):
+        self._data = {}
+        self._data_provider = data_provider
+        self._data_provider.subscribe(self)
+
+    def update_status(self, status, hardware):
+        if status is not None and "accessdoors" in status:
+            access_doors = status["accessdoors"]
+            for key in access_doors:
+                if key.startswith("item"):
+                    if key not in self._data:
+                        self._data[key] = {}
+
+                    if (
+                        "sumstate" in access_doors[key]
+                        and "value" in access_doors[key]["sumstate"]
+                    ):
+                        (
+                            self._data[key]["accessControllerActionState"],
+                            self._data[key]["elementInfo"],
+                            self._data[key]["accessState"],
+                            self._data[key]["gateRuntimeLevel"],
+                            self._data[key]["accessType"],
+                            *_other,
+                        ) = access_doors[key]["sumstate"]["value"].split(
+                            ";",
+                        )
+
+    def update_resources(self, resources):
+        if resources is not None and "accessdoors" in resources:
+            access_doors = resources["accessdoors"]
+            for key in access_doors:
+                if key.startswith("item"):
+                    if key not in self._data:
+                        self._data[key] = {}
+                    self._data[key]["name"] = access_doors[key]["name"]
+
+    @property
+    def access_doors(self):
+        """Returns the access doors read from MyGekko"""
+        result: list[AccessDoor] = []
+        for key, data in self._data.items():
+            result.append(AccessDoor(key, data["name"], self))
+
+        return result
+
+    def get_features(self, door: AccessDoor) -> list[AccessDoorFeature]:
+        """Returns the supported features"""
+        result = list()
+
+        if door and door.entity_id:
+            if door.entity_id in self._data:
+                data = self._data[door.entity_id]
+                if (
+                    "accessControllerActionState" in data
+                    and data["accessControllerActionState"]
+                ):
+                    result.append(AccessDoorFeature.OPEN)
+
+        return result
+
+    async def set_state(self, door: AccessDoor, state: AccessDoorCommand) -> None:
+        """Sets the state"""
+        if door and door.entity_id:
+            await self._data_provider.write_data(door.resource_path, str(state))

PyMyGekko/resources/EnergyCosts.py

@@ -106,7 +106,10 @@ def _decode_values(self, value: str, hardware: str) -> any:
                     self._transform_value(value_descriptions[index], value_parts)
                 )
             else:
-                _LOGGER.error("OutOfBounds access for value %s", value)
+                _LOGGER.info(
+                    "OutOfBounds access for value %s. Not all energy_costs value are read currently.",
+                    value,
+                )
 
         return values
 

README.md

@@ -64,6 +64,20 @@ hatch env create
 hatch run pytest
 ```
 
+### Build
+
+```
+hatch build
+```
+
+### Release
+
+- Increase version, then
+  ´´´
+  hatch build
+  hatch publish -u \_\_token\_\_ -a <PyPiToken>
+  ´´´
+
 ---
 
 [buymecoffee]: https://www.buymeacoffee.com/stephanu

tests/access_doors/data/api_var_response_879015.json

@@ -0,0 +1,31 @@
+{
+  "accessdoors": {
+    "item0": {
+      "name": "Türöffner",
+      "page": "Türöffner",
+      "sumstate": {
+        "description": "Summary of the states and their formats",
+        "format": "accessControllerActionState enum[0=close,1=open,2=holdOpen,3=partiallyOpen,4=partiallyHoldOpen](); elementInfo enum[0=ok,1=manualOff,2=manualOn,3=locked,4=alarm](); accessState enum[0=closed,1=open](); gateRuntimeLevel int[0,100](%); accessType enum[0=door control,1=gate control,2=barrier control,3=gate operation,10=myGEKKO Net remote door](); ",
+        "type": "STRING",
+        "permission": "READ"
+      },
+      "scmd": {
+        "description": "Summary of the commands and their formats",
+        "format": "-3|-2|-1|1|2 (Stop|Close|Lock|Open|Hold open)",
+        "type": "STRING",
+        "permission": "WRITE",
+        "index": 550070
+      }
+    },
+    "group0": {
+      "name": "Grp 1",
+      "sumstate": {
+        "description": "Summary of the group states and their formats",
+        "format": "State[0=Closed|1=Open]",
+        "type": "AO",
+        "permission": "READ",
+        "index": 5500000
+      }
+    }
+  }
+}

tests/access_doors/data/api_var_status_response_879015.json

@@ -0,0 +1,30 @@
+{
+  "globals": {
+    "network": {
+      "gekkoname": {
+        "value": "myGEKKO"
+      },
+      "language": {
+        "value": "0"
+      },
+      "version": {
+        "value": "879015"
+      },
+      "hardware": {
+        "value": "Slide 2 (AC0DFE3012E6)"
+      }
+    }
+  },
+  "accessdoors": {
+    "item0": {
+      "sumstate": {
+        "value": "0;0;0;0;0;"
+      }
+    },
+    "group0": {
+      "sumstate": {
+        "value": "0"
+      }
+    }
+  }
+}

tests/access_doors/test_pymygekko_access_doors_879015.py

@@ -0,0 +1,59 @@
+import logging
+
+import pytest
+from aiohttp import ClientSession
+from aiohttp import web
+from PyMyGekko import MyGekkoApiClientBase
+from PyMyGekko.resources.AccessDoors import AccessDoorElementInfo
+from PyMyGekko.resources.AccessDoors import AccessDoorFeature
+from PyMyGekko.resources.AccessDoors import AccessDoorState
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+
+async def var_response(request):
+    varResponseFile = open("tests/access_doors/data/api_var_response_879015.json")
+    return web.Response(status=200, body=varResponseFile.read())
+
+
+async def var_status_response(request):
+    statusResponseFile = open(
+        "tests/access_doors/data/api_var_status_response_879015.json"
+    )
+    return web.Response(status=200, body=statusResponseFile.read())
+
+
+@pytest.fixture
+def mock_server(aiohttp_server):
+    app = web.Application()
+    app.router.add_get("/api/v1/var", var_response)
+    app.router.add_get("/api/v1/var/status", var_status_response)
+    return aiohttp_server(app)
+
+
+@pytest.mark.asyncio
+async def test_get_access_doors(mock_server):
+    _LOGGER.setLevel(logging.DEBUG)
+
+    server = await mock_server
+    async with ClientSession() as session:
+        api = MyGekkoApiClientBase(
+            {},
+            session,
+            scheme=server.scheme,
+            host=server.host,
+            port=server.port,
+        )
+
+        await api.read_data()
+        doors = api.get_access_doors()
+
+        assert doors is not None
+        assert len(doors) == 1
+
+        assert doors[0].entity_id == "item0"
+        assert doors[0].name == "Türöffner"
+        assert doors[0].access_state == AccessDoorState.CLOSE
+        assert doors[0].element_info == AccessDoorElementInfo.OK
+        assert len(doors[0].supported_features) == 1
+        assert AccessDoorFeature.OPEN in doors[0].supported_features