Merge pull request #71 from mago0211/add_limitations

Add function to get limitation for windows with rain sensor

README.md

@@ -63,13 +63,18 @@ async def main(loop):
     await pyvlx.nodes['Bath'].close()
     await pyvlx.nodes['Bath'].set_position(Position(position_percent=45))
 
+    # Read limits of windows
+    # limit = await pyvlx.nodes['Bath'].get_limitation()
+    # limit.min_value
+    # limit.max_value
+
     # Changing of on-off switches:
     # await pyvlx.nodes['CoffeeMaker'].set_on()
     # await pyvlx.nodes['CoffeeMaker'].set_off()
 
     # You can easily rename nodes:
     # await pyvlx.nodes["Window 10"].rename("Window 11")
-
+        
     await pyvlx.disconnect()
 
 if __name__ == '__main__':

pyvlx/api/frame_creation.py

@@ -14,13 +14,15 @@
     FrameGetAllNodesInformationConfirmation,
     FrameGetAllNodesInformationFinishedNotification,
     FrameGetAllNodesInformationNotification,
-    FrameGetAllNodesInformationRequest, FrameGetLocalTimeConfirmation,
-    FrameGetLocalTimeRequest, FrameGetNetworkSetupConfirmation,
-    FrameGetNetworkSetupRequest, FrameGetNodeInformationConfirmation,
-    FrameGetNodeInformationNotification, FrameGetNodeInformationRequest,
-    FrameGetProtocolVersionConfirmation, FrameGetProtocolVersionRequest,
-    FrameGetSceneListConfirmation, FrameGetSceneListNotification,
-    FrameGetSceneListRequest, FrameGetStateConfirmation, FrameGetStateRequest,
+    FrameGetAllNodesInformationRequest, FrameGetLimitationStatus,
+    FrameGetLimitationStatusConfirmation, FrameGetLimitationStatusNotification,
+    FrameGetLocalTimeConfirmation, FrameGetLocalTimeRequest,
+    FrameGetNetworkSetupConfirmation, FrameGetNetworkSetupRequest,
+    FrameGetNodeInformationConfirmation, FrameGetNodeInformationNotification,
+    FrameGetNodeInformationRequest, FrameGetProtocolVersionConfirmation,
+    FrameGetProtocolVersionRequest, FrameGetSceneListConfirmation,
+    FrameGetSceneListNotification, FrameGetSceneListRequest,
+    FrameGetStateConfirmation, FrameGetStateRequest,
     FrameGetVersionConfirmation, FrameGetVersionRequest,
     FrameHouseStatusMonitorDisableConfirmation,
     FrameHouseStatusMonitorDisableRequest,
@@ -151,6 +153,13 @@ def create_frame(command):
     if command == Command.GW_GET_STATE_CFM:
         return FrameGetStateConfirmation()
 
+    if command == Command.GW_GET_LIMITATION_STATUS_REQ:
+        return FrameGetLimitationStatus()
+    if command == Command.GW_GET_LIMITATION_STATUS_CFM:
+        return FrameGetLimitationStatusConfirmation()
+    if command == Command.GW_LIMITATION_STATUS_NTF:
+        return FrameGetLimitationStatusNotification()
+
     if command == Command.GW_GET_NETWORK_SETUP_REQ:
         return FrameGetNetworkSetupRequest()
     if command == Command.GW_GET_NETWORK_SETUP_CFM:

pyvlx/api/frames/__init__.py

@@ -22,6 +22,9 @@
     FrameGetAllNodesInformationFinishedNotification,
     FrameGetAllNodesInformationNotification,
     FrameGetAllNodesInformationRequest)
+from .frame_get_limitation import (
+    FrameGetLimitationStatus, FrameGetLimitationStatusConfirmation,
+    FrameGetLimitationStatusNotification)
 from .frame_get_local_time import (
     FrameGetLocalTimeConfirmation, FrameGetLocalTimeRequest)
 from .frame_get_network_setup import (

pyvlx/api/frames/frame_get_limitation.py

@@ -0,0 +1,114 @@
+
+"""Module for get local time classes."""
+from pyvlx.const import Command, LimitationType, Originator, Priority
+
+from .frame import FrameBase
+
+
+class FrameGetLimitationStatus(FrameBase):
+    """Frame for requesting limitation status."""
+
+    PAYLOAD_LEN = 25
+
+    def __init__(self, node_ids=None, session_id=None, limitation_type=LimitationType.MIN_LIMITATION):
+        """Init Frame."""
+        super().__init__(Command.GW_GET_LIMITATION_STATUS_REQ)
+        self.session_id = session_id
+        self.originator = Originator.USER
+        self.priority = Priority.USER_LEVEL_2
+        self.node_ids = node_ids
+
+        self.parameter_id = 0  # Main Parameter
+        self.limitations_type = limitation_type
+
+    def get_payload(self):
+        """Return Payload."""
+        ret = bytes([self.session_id >> 8 & 255, self.session_id & 255])
+        ret += bytes([len(self.node_ids)])  # index array count
+        ret += bytes(self.node_ids) + bytes(20 - len(self.node_ids))
+        ret += bytes([self.parameter_id])
+        ret += bytes([self.limitations_type.value])
+        return ret
+
+    def __str__(self):
+        """Return human readable string."""
+        return f'<{type(self).__name__} node_ids="{self.node_ids}" ' \
+               f'session_id="{self.session_id}" originator="{self.originator}" />'
+
+
+class FrameGetLimitationStatusConfirmation(FrameBase):
+    """Frame for response for get limitation requests."""
+
+    PAYLOAD_LEN = 3
+
+    def __init__(self, session_id=None, data=None):
+        """Init Frame."""
+        super().__init__(Command.GW_GET_LIMITATION_STATUS_CFM)
+        self.session_id = session_id
+        self.data = data
+
+    def get_payload(self):
+        """Return Payload."""
+        ret = bytes([self.session_id >> 8 & 255, self.session_id & 255])
+        ret += bytes([self.data])
+        return ret
+
+    def from_payload(self, payload):
+        """Init frame from binary data."""
+        self.session_id = payload[0] * 256 + payload[1]
+        self.data = payload[2]
+
+    def __str__(self):
+        """Return human readable string."""
+        return '<{} session_id="{}" status="{}"/>'.format(
+            type(self).__name__, self.session_id, self.data
+        )
+
+
+class FrameGetLimitationStatusNotification(FrameBase):
+    """Frame for notification of note information request."""
+
+    PAYLOAD_LEN = 10
+
+    def __init__(self):
+        """Init Frame."""
+        super().__init__(Command.GW_LIMITATION_STATUS_NTF)
+        self.session_id = None
+        self.node_id = 0
+        self.parameter_id = 0
+        self.min_value = None
+        self.max_value = None
+        self.limit_originator = None
+        self.limit_time = None
+
+    def get_payload(self):
+        """Return Payload."""
+        payload = bytes([self.session_id >> 8 & 255, self.session_id & 255])
+        payload += bytes([self.node_id])
+        payload += bytes([self.parameter_id])
+        payload += bytes([self.min_value >> 8 & 255, self.min_value & 255])
+        payload += bytes([self.max_value >> 8 & 255, self.max_value & 255])
+        payload += bytes([self.limit_originator])
+        payload += bytes([self.limit_time])
+        return payload
+
+    def from_payload(self, payload):
+        """Init frame from binary data."""
+        self.session_id = payload[0] * 256 + payload[1]
+        self.node_id = payload[2]
+        self.parameter_id = payload[3]
+        self.min_value = payload[4:5]
+        self.max_value = payload[6:7]
+        self.limit_originator = Originator(payload[8])
+        self.limit_time = payload[9]
+
+    def __str__(self):
+        """Return human readable string."""
+        return (
+            '<{} node_id="{}" session_id="{}" min_value="{}" '
+            'max_value="{}" originator="{}" limit_time="{}"/>'.format(
+                type(self).__name__, self.node_id, self.session_id,
+                self.min_value, self.max_value, self.limit_originator,
+                self.limit_time
+            )
+        )

pyvlx/api/get_limitation.py

@@ -0,0 +1,55 @@
+"""Module for retrieving limitation value from API."""
+
+from ..const import LimitationType
+from ..parameter import Position
+from .api_event import ApiEvent
+from .frames import (
+    FrameGetLimitationStatus, FrameGetLimitationStatusConfirmation,
+    FrameGetLimitationStatusNotification)
+from .session_id import get_new_session_id
+
+
+class GetLimitation(ApiEvent):
+    """Class for retrieving gateway state from API."""
+
+    def __init__(self, pyvlx, node_id, limitation_type=LimitationType.MIN_LIMITATION):
+        """Initialize SceneList class."""
+        super().__init__(pyvlx=pyvlx)
+        self.node_id = node_id
+        self.limitation_type = limitation_type
+        self.success = False
+        self.notification_frame = None
+        self.session_id = None
+        self.min_value_raw = None
+        self.max_value_raw = None
+        self.originator = None
+        self.limit_time = None
+
+    @property
+    def max_value(self):
+        return Position.to_percent(self.max_value_raw)
+
+    @property
+    def min_value(self):
+        return Position.to_percent(self.min_value_raw)
+
+    async def handle_frame(self, frame):
+        """Handle incoming API frame, return True if this was the expected frame."""
+        if isinstance(frame, FrameGetLimitationStatusConfirmation):
+            return False  # Wait for Notification Frame
+        if isinstance(frame, FrameGetLimitationStatusNotification):
+            if frame.session_id == self.session_id:
+                self.success = True
+                self.min_value_raw = frame.min_value
+                self.max_value_raw = frame.max_value
+                self.originator = frame.limit_originator
+                self.limit_time = frame.limit_time
+                self.notification_frame = frame
+                return True
+        return False
+
+    def request_frame(self):
+        """Construct initiating frame."""
+        self.session_id = get_new_session_id()
+        return FrameGetLimitationStatus(node_ids=[self.node_id], session_id=self.session_id,
+                                        limitation_type=self.limitation_type)

pyvlx/opening_device.py

@@ -1,5 +1,6 @@
 """Module for window openers."""
 from .api.command_send import CommandSend
+from .api.get_limitation import GetLimitation
 from .exception import PyVLXException
 from .node import Node
 from .parameter import CurrentPosition, Parameter, Position, TargetPosition
@@ -137,6 +138,13 @@ def __str__(self):
             )
         )
 
+    async def get_limitation(self):
+        get_limitation = GetLimitation(pyvlx=self.pyvlx, node_id=self.node_id)
+        await get_limitation.do_api_call()
+        if not get_limitation.success:
+            raise PyVLXException("Unable to send command")
+        return get_limitation
+
 
 class Blind(OpeningDevice):
     """Blind objects."""

pyvlx/pyvlx.py

@@ -7,7 +7,8 @@
 """
 import asyncio
 
-from .api import house_status_monitor_disable, house_status_monitor_enable
+from .api import (
+    get_limitation, house_status_monitor_disable, house_status_monitor_enable)
 from .config import Config
 from .connection import Connection
 from .heartbeat import Heartbeat
@@ -79,3 +80,7 @@ async def load_nodes(self, node_id=None):
     async def load_scenes(self):
         """Load scenes from KLF 200."""
         await self.scenes.load()
+
+    async def get_limitation(self, node_id):
+        limit = get_limitation.GetLimitation(self, [node_id])
+        await limit.do_api_call()

test/frame_get_limitation_status_cfm_test.py

@@ -0,0 +1,38 @@
+"""Unit tests for FrameGetLimitationStatusConfirmation."""
+import unittest
+
+from pyvlx.api.frame_creation import frame_from_raw
+from pyvlx.api.frames.frame_get_limitation import (
+    FrameGetLimitationStatusConfirmation)
+
+
+class TestFrameGetLimitationStatusConfirmation(unittest.TestCase):
+    """Test class for FrameGetLimitationStatusConfirmation."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_bytes(self):
+        """Test FrameGetLimitationStatusConfirmation bytes."""
+        frame = FrameGetLimitationStatusConfirmation(session_id=1, data=1)
+        self.assertEqual(bytes(frame), b'\x00\x06\x03\x13\x00\x01\x01\x16')
+
+        frame = FrameGetLimitationStatusConfirmation(session_id=2, data=0)
+        self.assertEqual(bytes(frame), b'\x00\x06\x03\x13\x00\x02\x00\x14')
+
+    def test_frame_from_raw(self):
+        """Test parse FrameGetLimitationStatusConfirmation from raw."""
+        frame = frame_from_raw(b'\x00\x06\x03\x13\x00\x01\x01\x16')
+        self.assertTrue(isinstance(frame, FrameGetLimitationStatusConfirmation))
+        self.assertEqual(frame.session_id, 1)
+        self.assertEqual(frame.data, 1)
+
+        frame = frame_from_raw(b'\x00\x06\x03\x13\x00\x02\x00\x14')
+        self.assertTrue(isinstance(frame, FrameGetLimitationStatusConfirmation))
+        self.assertEqual(frame.session_id, 2)
+        self.assertEqual(frame.data, 0)
+
+    def test_str(self):
+        """Test string representation of FrameGetLimitationStatusConfirmation."""
+        frame = FrameGetLimitationStatusConfirmation(session_id=1)
+        self.assertEqual(str(frame),
+                         '<FrameGetLimitationStatusConfirmation session_id="1" status="None"/>')

test/frame_get_limitation_status_ntf_test.py

@@ -0,0 +1,46 @@
+"""Unit tests for FrameGetLimitationStatusNotification."""
+import unittest
+
+from pyvlx.api.frame_creation import frame_from_raw
+from pyvlx.api.frames.frame_get_limitation import (
+    FrameGetLimitationStatusNotification)
+from pyvlx.const import Originator
+
+
+class TestFrameGetLimitationStatusNotification(unittest.TestCase):
+    """Test class for TestFrameGetLimitationStatusNotification."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_bytes(self):
+        """Test FrameGetLimitationStatusNotification bytes."""
+
+        frame = FrameGetLimitationStatusNotification()
+        frame.session_id = 1
+        frame.node_id = 1
+        frame.parameter_id = 0
+        frame.min_value = 47668
+        frame.max_value = 63487
+        frame.limit_originator = Originator.USER.value
+        frame.limit_time = 0
+        self.assertEqual(bytes(frame), b'\x00\r\x03\x14\x00\x01\x01\x00\xba4\xf7\xff\x01\x00\x9d')
+
+    def test_frame_from_raw(self):
+        """Test parse FrameGetLimitationStatusNotification from raw."""
+        frame = frame_from_raw(b'\x00\r\x03\x14\x00\x01\x01\x00\xba4\xf7\xff\x01\x00\x9d')
+        self.assertTrue(isinstance(frame, FrameGetLimitationStatusNotification))
+        self.assertEqual(frame.limit_originator, Originator.USER)
+        self.assertEqual(frame.node_id, 1)
+        self.assertEqual(frame.parameter_id, 0)
+        self.assertEqual(frame.session_id, 1)
+        self.assertEqual(frame.max_value, b'\xf7')
+        self.assertEqual(frame.min_value, b'\xba')
+        self.assertEqual(frame.limit_time, 0)
+
+    def test_str(self):
+        """Test string representation of FrameGetLimitationStatusNotification."""
+        frame = FrameGetLimitationStatusNotification()
+        self.assertEqual(str(frame),
+                         '<FrameGetLimitationStatusNotification node_id="0" '
+                         'session_id="None" min_value="None" max_value="None" '
+                         'originator="None" limit_time="None"/>')

test/frame_get_limitation_status_req_test.py

@@ -0,0 +1,33 @@
+"""Unit tests for FrameGetLimitationStatus."""
+import unittest
+
+from pyvlx.api.frame_creation import frame_from_raw
+from pyvlx.api.frames.frame_get_limitation import FrameGetLimitationStatus
+from pyvlx.const import LimitationType
+
+
+class TestFrameGetLimitationStatus(unittest.TestCase):
+    """Test class for FrameGetLimitationStatus."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_bytes(self):
+        """Test FrameGetLimitationStatus bytes."""
+        frame = FrameGetLimitationStatus(node_ids=[1], session_id=1, limitation_type=LimitationType.MIN_LIMITATION)
+        self.assertEqual(bytes(frame), b'\x00\x1c\x03\x12\x00\x01\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+                                       b'\x00\x00\x0c')
+
+        frame = FrameGetLimitationStatus(node_ids=[1, 2], session_id=2, limitation_type=LimitationType.MAX_LIMITATION)
+        self.assertEqual(bytes(frame), b'\x00\x1c\x03\x12\x00\x02\x02\x01\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+                                       b'\x00\x01\x0f')
+
+    def test_frame_from_raw(self):
+        """Test parse FrameGetLimitationStatus from raw."""
+        frame = frame_from_raw(b'\x00\x1c\x03\x12\x00\x01\x01\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+                               b'\x0c')
+        self.assertTrue(isinstance(frame, FrameGetLimitationStatus))
+
+    def test_str(self):
+        """Test string representation of FrameGetLimitationStatus."""
+        frame = FrameGetLimitationStatus(node_ids=[1], session_id=1, limitation_type=LimitationType.MIN_LIMITATION)
+        self.assertEqual(str(frame), '<FrameGetLimitationStatus node_ids="[1]" session_id="1" originator="Originator.USER" />')