Merge pull request #189 from farmio/expose-binary-value

new Sensor types, ExposeSensor fix

home-assistant-plugin/custom_components/xknx/__init__.py

@@ -335,4 +335,13 @@ async def _async_entity_changed(self, entity_id, old_state, new_state):
         """Handle entity change."""
         if new_state is None:
             return
-        await self.device.set(float(new_state.state))
+        if new_state.state == "unknown":
+            return
+
+        if self.type == 'binary':
+            if new_state.state == "on":
+                await self.device.set(True)
+            elif new_state.state == "off":
+                await self.device.set(False)
+        else:
+            await self.device.set(new_state.state)

home-assistant-plugin/custom_components/xknx/sensor.py

@@ -95,6 +95,11 @@ def unit_of_measurement(self):
         """Return the unit this state is expressed in."""
         return self.device.unit_of_measurement()
 
+    @property
+    def device_class(self):
+        """Return the device class of the sensor."""
+        return self.device.ha_device_class()
+
     @property
     def device_state_attributes(self):
         """Return the state attributes."""

test/dpt_scaling_test.py

@@ -0,0 +1,97 @@
+"""Unit test for KNX DPT 5.001 and 5.003 value."""
+import unittest
+
+from xknx.exceptions import ConversionError
+from xknx.knx import DPTAngle, DPTScaling
+
+
+class TestDPTScaling(unittest.TestCase):
+    """Test class for KNX scaling value."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_value_30_pct(self):
+        """Test parsing and streaming of DPTScaling 30%."""
+        self.assertEqual(DPTScaling().to_knx(30), (0x4C,))
+        self.assertEqual(DPTScaling().from_knx((0x4C,)), 30)
+
+    def test_value_99_pct(self):
+        """Test parsing and streaming of DPTScaling 99%."""
+        self.assertEqual(DPTScaling().to_knx(99), (0xFC,))
+        self.assertEqual(DPTScaling().from_knx((0xFC,)), 99)
+
+    def test_value_max(self):
+        """Test parsing and streaming of DPTScaling 100%."""
+        self.assertEqual(DPTScaling().to_knx(100), (0xFF,))
+        self.assertEqual(DPTScaling().from_knx((0xFF,)), 100)
+
+    def test_value_min(self):
+        """Test parsing and streaming of DPTScaling 0."""
+        self.assertEqual(DPTScaling().to_knx(0), (0x00,))
+        self.assertEqual(DPTScaling().from_knx((0x00,)), 0)
+
+    def test_to_knx_min_exceeded(self):
+        """Test parsing of DPTScaling with wrong value (underflow)."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().to_knx(-1)
+
+    def test_to_knx_max_exceeded(self):
+        """Test parsing of DPTScaling with wrong value (overflow)."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().to_knx(101)
+
+    def test_to_knx_wrong_parameter(self):
+        """Test parsing of DPTScaling with wrong value (string)."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().to_knx("fnord")
+
+    def test_from_knx_wrong_parameter(self):
+        """Test parsing of DPTScaling with wrong value (3 byte array)."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().from_knx((0x01, 0x02, 0x03))
+
+    def test_from_knx_wrong_value(self):
+        """Test parsing of DPTScaling with value which exceeds limits."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().from_knx((0x256,))
+
+    def test_from_knx_wrong_parameter2(self):
+        """Test parsing of DPTScaling with wrong value (array containing string)."""
+        with self.assertRaises(ConversionError):
+            DPTScaling().from_knx(("0x23"))
+
+
+class TestDPTAngle(unittest.TestCase):
+    """Test class for KNX scaling value."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_value_30_deg(self):
+        """Test parsing and streaming of DPTAngle 30°."""
+        self.assertEqual(DPTAngle().to_knx(30), (0x15,))
+        self.assertEqual(DPTAngle().from_knx((0x15,)), 30)
+
+    def test_value_270_deg(self):
+        """Test parsing and streaming of DPTAngle 270°."""
+        self.assertEqual(DPTAngle().to_knx(270), (0xBF,))
+        self.assertEqual(DPTAngle().from_knx((0xBF,)), 270)
+
+    def test_value_max(self):
+        """Test parsing and streaming of DPTAngle 360°."""
+        self.assertEqual(DPTAngle().to_knx(360), (0xFF,))
+        self.assertEqual(DPTAngle().from_knx((0xFF,)), 360)
+
+    def test_value_min(self):
+        """Test parsing and streaming of DPTAngle 0°."""
+        self.assertEqual(DPTAngle().to_knx(0), (0x00,))
+        self.assertEqual(DPTAngle().from_knx((0x00,)), 0)
+
+    def test_to_knx_min_exceeded(self):
+        """Test parsing of DPTAngle with wrong value (underflow)."""
+        with self.assertRaises(ConversionError):
+            DPTAngle().to_knx(-1)
+
+    def test_to_knx_max_exceeded(self):
+        """Test parsing of DPTAngle with wrong value (overflow)."""
+        with self.assertRaises(ConversionError):
+            DPTAngle().to_knx(361)

test/dpt_string_test.py

@@ -46,11 +46,6 @@ def test_value_special_chars(self):
         self.assertEqual(DPTString.to_knx(string), raw)
         self.assertEqual(DPTString.from_knx(raw), string)
 
-    def test_to_knx_wrong_parameter(self):
-        """Test serializing DPTString to KNX with wrong value (int)."""
-        with self.assertRaises(ConversionError):
-            DPTString().to_knx(123)
-
     def test_to_knx_too_long(self):
         """Test serializing DPTString to KNX with wrong value (to long)."""
         with self.assertRaises(ConversionError):

test/dpt_value_1_ucount_test.py

@@ -2,11 +2,11 @@
 import unittest
 
 from xknx.exceptions import ConversionError
-from xknx.knx import DPTValue1Ucount
+from xknx.knx import DPTSceneNumber, DPTValue1Ucount
 
 
 class TestDPTValue1Ucount(unittest.TestCase):
-    """Test class for KNX scaling value."""
+    """Test class for KNX 8-bit unsigned value."""
 
     # pylint: disable=too-many-public-methods,invalid-name
 
@@ -16,9 +16,9 @@ def test_value_50(self):
         self.assertEqual(DPTValue1Ucount().from_knx((0x32,)), 50)
 
     def test_value_max(self):
-        """Test parsing and streaming of DPTValue1Ucount 63."""
-        self.assertEqual(DPTValue1Ucount().to_knx(63), (0x3F,))
-        self.assertEqual(DPTValue1Ucount().from_knx((0x3F,)), 63)
+        """Test parsing and streaming of DPTValue1Ucount 255."""
+        self.assertEqual(DPTValue1Ucount().to_knx(255), (0xFF,))
+        self.assertEqual(DPTValue1Ucount().from_knx((0xFF,)), 255)
 
     def test_value_min(self):
         """Test parsing and streaming of DPTValue1Ucount 0."""
@@ -54,3 +54,24 @@ def test_from_knx_wrong_parameter2(self):
         """Test parsing of DPTValue1Ucount with wrong value (array containing string)."""
         with self.assertRaises(ConversionError):
             DPTValue1Ucount().from_knx(("0x23"))
+
+
+class TestDPTSceneNumber(unittest.TestCase):
+    """Test class for KNX scene number value."""
+
+    # pylint: disable=too-many-public-methods,invalid-name
+
+    def test_value_50(self):
+        """Test parsing and streaming of DPTSceneNumber 50."""
+        self.assertEqual(DPTSceneNumber().to_knx(50), (0x32,))
+        self.assertEqual(DPTSceneNumber().from_knx((0x32,)), 50)
+
+    def test_value_max(self):
+        """Test parsing and streaming of DPTSceneNumber 63."""
+        self.assertEqual(DPTSceneNumber().to_knx(63), (0x3F,))
+        self.assertEqual(DPTSceneNumber().from_knx((0x3F,)), 63)
+
+    def test_to_knx_max_exceeded(self):
+        """Test parsing of DPTSceneNumber with wrong value (overflow)."""
+        with self.assertRaises(ConversionError):
+            DPTSceneNumber().to_knx(64)

test/expose_sensor_test.py

@@ -5,7 +5,7 @@
 
 from xknx import XKNX
 from xknx.devices import ExposeSensor
-from xknx.knx import DPTArray, GroupAddress, Telegram, TelegramType
+from xknx.knx import DPTArray, DPTBinary, GroupAddress, Telegram, TelegramType
 
 
 class TestExposeSensor(unittest.TestCase):
@@ -23,6 +23,19 @@ def tearDown(self):
     #
     # STR FUNCTIONS
     #
+    def test_str_binary(self):
+        """Test resolve state with binary sensor."""
+        xknx = XKNX(loop=self.loop)
+        expose_sensor = ExposeSensor(
+            xknx,
+            'TestSensor',
+            group_address='1/2/3',
+            value_type="binary")
+        expose_sensor.sensor_value.payload = DPTBinary(1)
+
+        self.assertEqual(expose_sensor.resolve_state(), True)
+        self.assertEqual(expose_sensor.unit_of_measurement(), None)
+
     def test_str_percent(self):
         """Test resolve state with percent sensor."""
         xknx = XKNX(loop=self.loop)
@@ -33,7 +46,7 @@ def test_str_percent(self):
             value_type="percent")
         expose_sensor.sensor_value.payload = DPTArray((0x40,))
 
-        self.assertEqual(expose_sensor.resolve_state(), 75)
+        self.assertEqual(expose_sensor.resolve_state(), 25)
         self.assertEqual(expose_sensor.unit_of_measurement(), "%")
 
     def test_str_temperature(self):
@@ -52,6 +65,24 @@ def test_str_temperature(self):
     #
     # TEST SET
     #
+    def test_set_binary(self):
+        """Test set with binary sensor."""
+        xknx = XKNX(loop=self.loop)
+        expose_sensor = ExposeSensor(
+            xknx,
+            'TestSensor',
+            group_address='1/2/3',
+            value_type="binary")
+        self.loop.run_until_complete(asyncio.Task(expose_sensor.set(False)))
+        self.assertEqual(xknx.telegrams.qsize(), 1)
+        telegram = xknx.telegrams.get_nowait()
+        self.assertEqual(
+            telegram,
+            Telegram(
+                GroupAddress('1/2/3'),
+                TelegramType.GROUP_WRITE,
+                payload=DPTBinary(0)))
+
     def test_set_percent(self):
         """Test set with percent sensor."""
         xknx = XKNX(loop=self.loop)
@@ -69,10 +100,10 @@ def test_set_percent(self):
             Telegram(
                 GroupAddress('1/2/3'),
                 TelegramType.GROUP_WRITE,
-                payload=DPTArray((0x40,))))
+                payload=DPTArray((0xBF,))))
 
     def test_set_temperature(self):
-        """Test set with temperatur sensor."""
+        """Test set with temperature sensor."""
         xknx = XKNX(loop=self.loop)
         expose_sensor = ExposeSensor(
             xknx,
@@ -92,6 +123,28 @@ def test_set_temperature(self):
     #
     # TEST PROCESS (GROUP READ)
     #
+    def test_process_binary(self):
+        """Test reading binary expose sensor from bus."""
+        xknx = XKNX(loop=self.loop)
+        expose_sensor = ExposeSensor(
+            xknx,
+            'TestSensor',
+            value_type='binary',
+            group_address='1/2/3')
+        expose_sensor.sensor_value.payload = DPTArray(1)
+
+        telegram = Telegram(GroupAddress('1/2/3'))
+        telegram.telegramtype = TelegramType.GROUP_READ
+        self.loop.run_until_complete(asyncio.Task(expose_sensor.process(telegram)))
+        self.assertEqual(xknx.telegrams.qsize(), 1)
+        telegram = xknx.telegrams.get_nowait()
+        self.assertEqual(
+            telegram,
+            Telegram(
+                GroupAddress('1/2/3'),
+                TelegramType.GROUP_RESPONSE,
+                payload=DPTArray(True)))
+
     def test_process_percent(self):
         """Test reading percent expose sensor from bus."""
         xknx = XKNX(loop=self.loop)

test/remote_value_dpt_2_byte_unsigned_test.py

@@ -8,7 +8,7 @@
 from xknx.knx import DPTArray, DPTBinary, GroupAddress, Telegram
 
 
-class TestRemoteValueDptValue1Ucount(unittest.TestCase):
+class TestRemoteValueDptValue2Ucount(unittest.TestCase):
     """Test class for RemoteValueDpt2ByteUnsigned objects."""
 
     def setUp(self):
@@ -33,13 +33,13 @@ def test_from_knx(self):
         self.assertEqual(remote_value.from_knx(DPTArray((0x0A, 0x0B))), 2571)
 
     def test_to_knx_error(self):
-        """Test to_knx function with wrong parametern."""
+        """Test to_knx function with wrong parameters."""
         xknx = XKNX(loop=self.loop)
         remote_value = RemoteValueDpt2ByteUnsigned(xknx)
         with self.assertRaises(ConversionError):
             remote_value.to_knx(65536)
         with self.assertRaises(ConversionError):
-            remote_value.to_knx("256")
+            remote_value.to_knx("a")
 
     def test_set(self):
         """Test setting value."""

test/remote_value_switch_test.py

@@ -24,29 +24,29 @@ def test_to_knx(self):
         """Test to_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
         remote_value = RemoteValueSwitch(xknx)
-        self.assertEqual(remote_value.to_knx(RemoteValueSwitch.Value.ON), DPTBinary(1))
-        self.assertEqual(remote_value.to_knx(RemoteValueSwitch.Value.OFF), DPTBinary(0))
+        self.assertEqual(remote_value.to_knx(True), DPTBinary(True))
+        self.assertEqual(remote_value.to_knx(False), DPTBinary(False))
 
     def test_from_knx(self):
         """Test from_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
         remote_value = RemoteValueSwitch(xknx)
-        self.assertEqual(remote_value.from_knx(DPTBinary(1)), RemoteValueSwitch.Value.ON)
-        self.assertEqual(remote_value.from_knx(DPTBinary(0)), RemoteValueSwitch.Value.OFF)
+        self.assertEqual(remote_value.from_knx(DPTBinary(True)), True)
+        self.assertEqual(remote_value.from_knx(DPTBinary(0)), False)
 
     def test_to_knx_invert(self):
         """Test to_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
         remote_value = RemoteValueSwitch(xknx, invert=True)
-        self.assertEqual(remote_value.to_knx(RemoteValueSwitch.Value.ON), DPTBinary(0))
-        self.assertEqual(remote_value.to_knx(RemoteValueSwitch.Value.OFF), DPTBinary(1))
+        self.assertEqual(remote_value.to_knx(True), DPTBinary(0))
+        self.assertEqual(remote_value.to_knx(False), DPTBinary(1))
 
     def test_from_knx_invert(self):
         """Test from_knx function with normal operation."""
         xknx = XKNX(loop=self.loop)
         remote_value = RemoteValueSwitch(xknx, invert=True)
-        self.assertEqual(remote_value.from_knx(DPTBinary(1)), RemoteValueSwitch.Value.OFF)
-        self.assertEqual(remote_value.from_knx(DPTBinary(0)), RemoteValueSwitch.Value.ON)
+        self.assertEqual(remote_value.from_knx(DPTBinary(1)), False)
+        self.assertEqual(remote_value.from_knx(DPTBinary(0)), True)
 
     def test_to_knx_error(self):
         """Test to_knx function with wrong parametern."""
@@ -90,7 +90,7 @@ def test_process(self):
         self.assertEqual(remote_value.value, None)
         self.loop.run_until_complete(asyncio.Task(remote_value.process(telegram)))
         self.assertIsNotNone(remote_value.payload)
-        self.assertEqual(remote_value.value, RemoteValueSwitch.Value.ON)
+        self.assertEqual(remote_value.value, True)
 
     def test_process_off(self):
         """Test process OFF telegram."""
@@ -104,7 +104,7 @@ def test_process_off(self):
         self.assertEqual(remote_value.value, None)
         self.loop.run_until_complete(asyncio.Task(remote_value.process(telegram)))
         self.assertIsNotNone(remote_value.payload)
-        self.assertEqual(remote_value.value, RemoteValueSwitch.Value.OFF)
+        self.assertEqual(remote_value.value, False)
 
     def test_to_process_error(self):
         """Test process errornous telegram."""