(re-)added pyvlx object. (re-)implemented old scene logic

examples/demo.py

@@ -1,53 +1,31 @@
 """Just a demo of the new PyVLX module."""
 import asyncio
-from pyvlx.config import Config
-from pyvlx.connection import Connection
-from pyvlx.login import Login
-from pyvlx.scene_list import SceneList
-from pyvlx.get_node_information import GetNodeInformation
-from pyvlx.get_all_nodes_information import GetAllNodesInformation
 
+from pyvlx import PyVLX
 
-async def frame_received_callback(frame):
-    """Log frame for better debugging."""
-    # print("Frame: ", frame)
-    # pylint: disable=unused-argument
-    pass
-
-
-async def demo(config, loop):
+async def main(loop):
     """Demonstrate functionality of PyVLX."""
-    connection = Connection(loop=loop, config=config)
-    connection.register_frame_received_cb(frame_received_callback)
-    await connection.connect()
-
-    login = Login(connection=connection, password=config.password)
-    await login.do_api_call()
+    pyvlx = PyVLX('pyvlx.yaml', log_frames=True, loop=loop)
+    # Alternative:
+    # pyvlx = PyVLX(host="192.168.2.127", password="velux123", loop=loop)
 
-    if not login.success:
-        print("Unable to login. Giving up""")
-        return
+    await pyvlx.connect()
+    await pyvlx.load_scenes()
 
-    scene_list = SceneList(connection=connection)
-    await scene_list.do_api_call()
-    print(scene_list.scenes)
+    await pyvlx.scenes["All Windows Closed"].run()
 
-    get_node_information = GetNodeInformation(connection=connection, node_id=6)
-    await get_node_information.do_api_call()
-    print(get_node_information.node)
+    #get_node_information = GetNodeInformation(connection=connection, node_id=6)
+    #await get_node_information.do_api_call()
+    #print(get_node_information.node)
 
-    print("---------------------------------------")
+    #print("---------------------------------------")
 
-    get_all_nodes_information = GetAllNodesInformation(connection=connection)
-    await get_all_nodes_information.do_api_call()
-    print(get_all_nodes_information.nodes)
+    #from pyvlx.get_node_information import GetNodeInformation
+    #from pyvlx.get_all_nodes_information import GetAllNodesInformation
+    # get_all_nodes_information = GetAllNodesInformation(connection=pyvlx.connection)
+    # await get_all_nodes_information.do_api_call()
+    # print(get_all_nodes_information.nodes)
 
-    # from pyvlx.activate_scene import ActivateScene
-    # activate_scene = ActivateScene(connection=connection, scene_id=2)
-    # await activate_scene.do_api_call()
-
-    # activate_scene = ActivateScene(connection=connection, scene_id=0)
-    # await activate_scene.do_api_call()
 
     # BACKUP:
     #
@@ -65,14 +43,8 @@ async def demo(config, loop):
     # connection.write(FrameCommandSendRequest(node_ids=[4], position=100, session_id=9))
 
 
-async def main(loop):
-    """Do main."""
-    config = Config(host="192.168.2.102", port=51200, password='velux123')
-    await demo(loop=loop, config=config)
-
-
-# pylint: disable=invalid-name
 if __name__ == '__main__':
+    # pylint: disable=invalid-name
     LOOP = asyncio.get_event_loop()
     LOOP.run_until_complete(main(LOOP))
     # LOOP.run_forever()

pyvlx.yaml

@@ -1,4 +1,4 @@
 
 config:
-    host: "192.168.2.127"
+    host: "192.168.2.102"
     password: "velux123"

pyvlx/__init__.py

@@ -1 +1,5 @@
 """Module for accessing KLF 200 gateway with python."""
+
+# flake8: noqa
+from .pyvlx import PyVLX
+from .exception import PyVLXException

pyvlx/config.py

@@ -1,13 +1,45 @@
 """Module for configuration."""
 
+import yaml
 
+from .exception import PyVLXException
+
+
+# pylint: disable=too-few-public-methods, too-many-arguments
 class Config:
-    """Class for Configuration."""
+    """Object for configuration."""
 
-    # pylint: disable=too-few-public-methods
+    DEFAULT_PORT = 51200
 
-    def __init__(self, host, password, port=51200):
-        """Initialize configuration."""
-        self.port = port
+    def __init__(self, pyvlx, path=None, host=None, password=None, port=None):
+        """Initialize Config class."""
+        self.pyvlx = pyvlx
         self.host = host
         self.password = password
+        self.port = port or self.DEFAULT_PORT
+        if path is not None:
+            self.read_config(path)
+
+    def read_config(self, path):
+        """Read configuration file."""
+        self.pyvlx.logger.info('Reading config file: ', path)
+        try:
+            with open(path, 'r') as filehandle:
+                doc = yaml.load(filehandle)
+                self.test_configuration(doc, path)
+                self.host = doc['config']['host']
+                self.password = doc['config']['password']
+                if 'port' in doc['config']:
+                    self.port = doc['config']['port']
+        except FileNotFoundError as ex:
+            raise PyVLXException('file does not exist: {0}'.format(ex))
+
+    @staticmethod
+    def test_configuration(doc, path):
+        """Test if configuration file is sane."""
+        if 'config' not in doc:
+            raise PyVLXException('no element config found in: {0}'.format(path))
+        if 'host' not in doc['config']:
+            raise PyVLXException('no element host found in: {0}'.format(path))
+        if 'password' not in doc['config']:
+            raise PyVLXException('no element password found in: {0}'.format(path))

pyvlx/node_helper.py

@@ -1,10 +1,10 @@
+"""Helper module for Node objects."""
 from .const import NodeTypeWithSubtype
 
 
 def convert_frame_to_node(frame):
-
+    """Convert FrameGet[All]Node[s]InformationNotification into Node object."""
     if frame.node_type == NodeTypeWithSubtype.WINDOW_OPENER_WITH_RAIN_SENSOR:
         return "WINDOW"
-
     print("{} not implemented", format(frame.node_type))
     return None

pyvlx/pyvlx.py

@@ -0,0 +1,57 @@
+"""
+Module for PyVLX object.
+
+PyVLX is an asynchronous library for connecting to
+a VELUX KLF 200 device for controlling window openers
+and roller shutters.
+"""
+import logging
+import asyncio
+from .config import Config
+from .connection import Connection
+from .login import Login
+from .exception import PyVLXException
+# from .devices import Devices
+from .scene_list import SceneList
+
+
+class PyVLX:
+    """Class for PyVLX."""
+
+    # pylint: disable=too-many-arguments
+
+    def __init__(self, path=None, host=None, password=None, log_frames=False, loop=None):
+        """Initialize PyVLX class."""
+        self.loop = loop or asyncio.get_event_loop()
+        self.logger = logging.getLogger('pyvlx.log')
+        self.config = Config(self, path, host, password)
+        self.connection = Connection(loop=self.loop, config=self.config)
+        if log_frames:
+            self.connection.register_frame_received_cb(self.log_frame)
+        # self.devices = Devices(self)
+        self.scenes = SceneList(self)
+
+    async def connect(self):
+        """Connect to KLF 200."""
+        await self.connection.connect()
+        login = Login(connection=self.connection, password=self.config.password)
+        await login.do_api_call()
+        if not login.success:
+            raise PyVLXException("Unable to login")
+
+    async def disconnect(self):
+        """Disconnect from KLF 200."""
+        self.connection.disconnect()
+
+    async def load_devices(self):
+        """Load devices from KLF 200."""
+        # await self.devices.load()
+        pass
+
+    async def load_scenes(self):
+        """Load scenes from KLF 200."""
+        await self.scenes.load()
+
+    async def log_frame(self, frame):
+        """Log frame to logger."""
+        self.logger.warning("%s", frame)

pyvlx/scene.py

@@ -0,0 +1,29 @@
+"""Module for scene."""
+from pyvlx.activate_scene import ActivateScene
+
+
+class Scene:
+    """Object for scene."""
+
+    def __init__(self, pyvlx, scene_id, name):
+        """Initialize Scene object."""
+        self.pyvlx = pyvlx
+        self.scene_id = scene_id
+        self.name = name
+
+    async def run(self):
+        """Run scene."""
+        activate_scene = ActivateScene(connection=self.pyvlx.connection, scene_id=self.scene_id)
+        await activate_scene.do_api_call()
+
+    def __str__(self):
+        """Return object as readable string."""
+        return '<Scene name="{0}" ' \
+            'id="{1}" />' \
+            .format(
+                self.name,
+                self.scene_id)
+
+    def __eq__(self, other):
+        """Equal operator."""
+        return self.__dict__ == other.__dict__

pyvlx/scene_list.py

@@ -1,17 +1,19 @@
 """Module for retrieving scene list from API."""
-from pyvlx.frame_get_scene_list import FrameGetSceneListRequest, FrameGetSceneListConfirmation, FrameGetSceneListNotification
-from pyvlx.api_event import ApiEvent
+from .scene import Scene
+from .frame_get_scene_list import FrameGetSceneListRequest, FrameGetSceneListConfirmation, FrameGetSceneListNotification
+from .api_event import ApiEvent
 
 
 class SceneList(ApiEvent):
     """Class for retrieving scene list from API."""
 
-    def __init__(self, connection):
+    def __init__(self, pyvlx):
         """Initialize SceneList class."""
-        super().__init__(connection)
+        super().__init__(pyvlx.connection)
+        self.pyvlx = pyvlx
         self.success = False
         self.count_scenes = None
-        self.scenes = []
+        self.__scenes = []
 
     async def handle_frame(self, frame):
         """Handle incoming API frame, return True if this was the expected frame."""
@@ -20,16 +22,48 @@ async def handle_frame(self, frame):
             # We are still waiting for FrameGetSceneListNotification(s)
             return False
         if isinstance(frame, FrameGetSceneListNotification):
-            self.scenes.extend(frame.scenes)
+            self.add_scenes(frame.scenes)
             if frame.remaining_scenes != 0:
                 # We are still waiting for FrameGetSceneListConfirmation(s)
                 return False
-            if self.count_scenes != len(self.scenes):
-                print("Warning: number of received scenes does not match expected number")
+            if self.count_scenes != len(self.__scenes):
+                self.pyvlx.logger.warning("Warning: number of received scenes does not match expected number")
             self.success = True
             return True
         return False
 
+    def add_scenes(self, scenes):
+        """Adding scenes to internal scene array."""
+        for scene in scenes:
+            self.add(Scene(pyvlx=self.pyvlx, scene_id=scene[0], name=scene[1]))
+
     def request_frame(self):
         """Construct initiating frame."""
         return FrameGetSceneListRequest()
+
+    def __iter__(self):
+        """Iterator."""
+        yield from self.__scenes
+
+    def __getitem__(self, key):
+        """Return scene by name or by index."""
+        for scene in self.__scenes:
+            if scene.name == key:
+                return scene
+        if isinstance(key, int):
+            return self.__scenes[key]
+        raise KeyError
+
+    def __len__(self):
+        """Return number of scenes."""
+        return len(self.__scenes)
+
+    def add(self, scene):
+        """Add scene."""
+        if not isinstance(scene, Scene):
+            raise TypeError()
+        self.__scenes.append(scene)
+
+    async def load(self):
+        """Load scenes from KLF 200."""
+        await self.do_api_call()

old_api/test/config_test.py → test/config_test.py

@@ -1,34 +1,30 @@
 """Unit test for configuration."""
 import unittest
-import asyncio
-
 from pyvlx import PyVLX, PyVLXException
 
 
 # pylint: disable=too-many-public-methods,invalid-name
 class TestConfig(unittest.TestCase):
     """Test class for configuration."""
 
-    def setUp(self):
-        """Set up test class."""
-        self.loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self.loop)
-
-    def tearDown(self):
-        """Tear down test class."""
-        self.loop.close()
-
     def test_config_file(self):
         """Test host/password configuration via config.yaml."""
         pyvlx = PyVLX("test/data/test_config.yaml")
         self.assertEqual(pyvlx.config.password, "velux123")
         self.assertEqual(pyvlx.config.host, "192.168.2.127")
+        self.assertEqual(pyvlx.config.port, 51200)
+
+    def test_config_file_with_port(self):
+        """Test host/password configuration via config.yaml."""
+        pyvlx = PyVLX("test/data/test_config_with_port.yaml")
+        self.assertEqual(pyvlx.config.port, 1234)
 
     def test_config_explicit(self):
         """Test host/password configuration via parameter."""
         pyvlx = PyVLX(host="192.168.2.127", password="velux123")
         self.assertEqual(pyvlx.config.password, "velux123")
         self.assertEqual(pyvlx.config.host, "192.168.2.127")
+        self.assertEqual(pyvlx.config.port, 51200)
 
     def test_config_wrong1(self):
         """Test configuration with missing password."""
@@ -49,7 +45,3 @@ def test_config_non_existent(self):
         """Test non existing configuration path."""
         with self.assertRaises(PyVLXException):
             PyVLX("test/data/test_config_non_existent.yaml")
-
-
-SUITE = unittest.TestLoader().loadTestsFromTestCase(TestConfig)
-unittest.TextTestRunner(verbosity=2).run(SUITE)

test/data/test_config_with_port.yaml

@@ -0,0 +1,5 @@
+
+config:
+    host: "192.168.2.127"
+    password: "velux123"
+    port: 1234