Merge pull request #148 from doudz/dev

v0.34.0

MANIFEST.in

@@ -2,7 +2,10 @@
 include *.md
 include LICENSE
 recursive-include zigate *.json
+recursive-include zigate *.py
+recursive-include zigate *.tpl
 prune .editorconfig
 prune *.rb
 prune build_upload.sh
 prune tests
+prune zigate2zigpy.py

README.md

@@ -324,4 +324,3 @@ Si vous souhaitez apporter une contribution à ce projet, nous vous suggérons d
 * <https://github.com/firstcontributions/first-contributions/blob/master/github-desktop-tutorial.md>
 
 Certains développeurs pourraient également être intéressés par des dons sous forme de matériel, tels que des modules ou des dispositifs Zigbee, et même si ces dons sont le plus souvent donnés sans aucune condition, cela pourrait dans de nombreux cas motiver les développeurs et indirectement améliorer le développement de ce projet.
-

build_upload.sh

@@ -1,4 +1,4 @@
 rm -R build
 rm -R dist
-python3 setup.py sdist bdist_wheel
+python3 setup.py sdist
 twine upload dist/*

setup.py

@@ -29,6 +29,58 @@
 with open(path.join(here, 'README.md'), encoding='utf-8') as f:
     long_description = f.read()
 
+
+# extracted from https://raspberrypi.stackexchange.com/questions/5100/detect-that-a-python-program-is-running-on-the-pi
+def is_raspberry_pi(raise_on_errors=False):
+    """Checks if Raspberry PI.
+
+    :return:
+    """
+    try:
+        with open('/proc/cpuinfo', 'r') as cpuinfo:
+            found = False
+            for line in cpuinfo:
+                if line.startswith('Hardware'):
+                    found = True
+                    label, value = line.strip().split(':', 1)
+                    value = value.strip()
+                    if value not in (
+                        'BCM2708',
+                        'BCM2709',
+                        'BCM2835',
+                        'BCM2836'
+                    ):
+                        if raise_on_errors:
+                            raise ValueError(
+                                'This system does not appear to be a '
+                                'Raspberry Pi.'
+                            )
+                        else:
+                            return False
+            if not found:
+                if raise_on_errors:
+                    raise ValueError(
+                        'Unable to determine if this system is a Raspberry Pi.'
+                    )
+                else:
+                    return False
+    except IOError:
+        if raise_on_errors:
+            raise ValueError('Unable to open `/proc/cpuinfo`.')
+        else:
+            return False
+
+    return True
+
+
+requires = ['pyserial>=3.2',
+            'pydispatcher>=2.0.5',
+            'bottle',
+            'requests'
+            ]
+if is_raspberry_pi():
+    requires.append('RPi.GPIO')
+
 # Setup part
 setup(
     name='zigate',
@@ -52,15 +104,10 @@
     packages=['zigate'],
     include_package_data=True,
 
-    install_requires=[
-        'pyserial',
-        'pydispatcher',
-        'bottle'
-    ],
+    install_requires=requires,
     extras_require={
         'dev': ['tox'],
-        'mqtt': ['paho-mqtt'],
-        'gpio': ['RPi.GPIO']
+        'mqtt': ['paho-mqtt']
     },
     python_requires='>=3',
 

tests/test_clusters.py

@@ -259,6 +259,29 @@ def test_cluster_C000F(self):
                                'name': 'present_value', 'value': True, 'type': bool}}
                          )
 
+    def test_cluster_C0006(self):
+        data = {"attributes": [{"attribute": 0,
+                                "data": True
+                                }],
+                "cluster": 0x0006
+                }
+        c = clusters.C0006.from_json(data)
+        self.assertEqual(c.attributes,
+                         {0: {'attribute': 0, 'data': True,
+                              'name': 'onoff', 'value': True, 'type': bool}}
+                         )
+
+        data = {"attributes": [{"attribute": 0,
+                                "data": 0x80
+                                }],
+                "cluster": 0x0006
+                }
+        c = clusters.C0006.from_json(data)
+        self.assertEqual(c.attributes,
+                         {0: {'attribute': 0, 'data': 0x80,
+                              'name': 'multiclick', 'value': 0x80, 'type': int}}
+                         )
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_core.py

@@ -701,7 +701,7 @@ def test_handle_response_8024(self):
         def setup_connection():
             self.zigate.connection = transport.FakeTransport()
             self.zigate.connection.add_auto_response(0x0009, 0x8009,
-                                                     unhexlify(b'00000123456789abcdef123400000000000000000b'))
+                                                     unhexlify(b'1234fedcba9876543210123400000000000000000b'))
             self.zigate.connection.add_auto_response(0x0024, 0x8024,
                                                      unhexlify(b'001234fedcba98765432100b'))
         self.zigate.setup_connection = setup_connection
@@ -756,6 +756,12 @@ def test_refresh(self):
                          b'02123401010006000000000200000002'
                          )
 
+    def test_write_attribute(self):
+        self.zigate.write_attribute_request('abcd', 1, 0xfc01, [(0, 0x09, b'\x01\x01')])
+        self.assertEqual(hexlify(self.zigate.connection.get_last_cmd()),
+                         b'02abcd0101fc0100000000010000090101'
+                         )
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_devices.py

@@ -4,6 +4,7 @@
 '''
 
 import unittest
+from unittest import mock
 from zigate import core
 import json
 import os
@@ -160,6 +161,13 @@ def test_template(self):
                                {'endpoint': 3, 'cluster': 18, 'attribute': 85, 'name': 'multiclick3',
                                 'value': '', 'expire': 2, 'type': str}]
                               )
+        device = core.Device({'addr': '1234', 'ieee': '0123456789abcdef', 'manufacturer_code': '1110'}, mock.Mock())
+#         device.set_attribute(1, 0, {'attribute': 5, 'lqi': 255, 'data': 'unsupported'})
+        ep = device.get_endpoint(1)
+        ep['profile'] = 260
+        ep['device'] = 512
+        self.assertTrue(device.load_template())
+        self.assertEqual(device.get_type(), 'profalux bso cover')
 
     def test_inverse_bool(self):
         device = core.Device({'addr': '1234', 'ieee': '0123456789abcdef'}, self.zigate)

tests/test_responses.py

@@ -468,8 +468,7 @@ def test_response_804A(self):
                                           ('status', 0),
                                           ('total_transmission', 2),
                                           ('transmission_failures', 1),
-                                          ('scanned_channels1', 0),
-                                          ('scanned_channels2', 1),
+                                          ('scanned_channels', 1),
                                           ('channel_count', 2),
                                           ('lqi', 255),
                                           ('channels', [OrderedDict([('channel', 1)]),
@@ -484,8 +483,7 @@ def test_response_804A(self):
                                           ('status', 0),
                                           ('total_transmission', 2),
                                           ('transmission_failures', 1),
-                                          ('scanned_channels1', 0),
-                                          ('scanned_channels2', 1),
+                                          ('scanned_channels', 1),
                                           ('channel_count', 2),
                                           ('lqi', 255),
                                           ('channels', [OrderedDict([('channel', 1)]),
@@ -516,6 +514,27 @@ def test_response_8030(self):
                                           ])
                              )
 
+    def test_response_004d(self):
+        msg_data = unhexlify(b'abcd0123456789abcdef01')  # fw < 3.1b
+        r = responses.R004D(msg_data, 255)
+        self.assertDictEqual(r.data,
+                             OrderedDict([('addr', 'abcd'),
+                                          ('ieee', '0123456789abcdef'),
+                                          ('mac_capability', '00000001'),
+                                          ('lqi', 255),
+                                          ])
+                             )
+        msg_data = unhexlify(b'abcd0123456789abcdef0101')  # fw >= 3.1b
+        r = responses.R004D(msg_data, 255)
+        self.assertDictEqual(r.data,
+                             OrderedDict([('addr', 'abcd'),
+                                          ('ieee', '0123456789abcdef'),
+                                          ('mac_capability', '00000001'),
+                                          ('rejoin_status', True),
+                                          ('lqi', 255),
+                                          ])
+                             )
+
 
 if __name__ == '__main__':
     unittest.main()

zigate/adminpanel.py → zigate/adminpanel/__init__.py

@@ -6,7 +6,7 @@
 #
 
 import threading
-from bottle import Bottle, route, run, template  # noqa
+from bottle import Bottle, route, run, view, template  # noqa
 
 ADMINPANEL_PORT = 9998
 
@@ -16,8 +16,20 @@ def start_adminpanel(zigate_instance, port=ADMINPANEL_PORT, daemon=True, quiet=T
     app.zigate = zigate_instance
 
     @app.route('/')
+    @view('index')
     def index():
-        return '<html><h1>ZiGate</h1><b>Hello</b>!</html>'
+        from zigate import version
+        return {'port': zigate_instance._port,
+                'libversion': version.__version__,
+                'version': zigate_instance.get_version_text(),
+                'connected': zigate_instance.connection.is_connected(),
+                'devices': zigate_instance.devices
+                }
+
+    @app.route('/networkmap')
+    @view('networkmap')
+    def networkmap():
+        return
 
     kwargs = {'host': '0.0.0.0', 'port': port,
               'quiet': quiet}
@@ -31,4 +43,6 @@ def index():
 
 
 if __name__ == '__main__':
-    start_adminpanel(None, daemon=False, quiet=False)
+    from unittest import mock
+    zigate_instance = mock.MagicMock()
+    start_adminpanel(zigate_instance, daemon=False, quiet=False)

zigate/adminpanel/views/base.tpl

@@ -0,0 +1,12 @@
+<html>
+	<head>
+		<title>ZiGate Admin Panel</title>
+	</head>
+	<body>
+		<h1>ZiGate</h1>
+		<a href="/">Index</a>
+		<a href="/networkmap">Network Map</a>
+		<h2>{{get('subtitle', 'Index')}}</h2>
+		{{!base}}
+	</body>
+</html>

zigate/adminpanel/views/index.tpl

@@ -0,0 +1,14 @@
+% rebase('base.tpl')
+<ul>
+	<li>Lib version : {{libversion}}</li>
+	<li>Port : {{port}}</li>
+	<li>Connected : {{connected}}</li>
+	<li>Firmware version : {{version}}</li>
+</ul>
+
+<h2>Devices :</h2>
+<ul>
+  % for device in devices:
+    <li>{{device}}</li>
+  % end
+</ul>