Merge 20c941b into 8784f51

changelog.md

@@ -3,6 +3,12 @@
 ## Unreleased changes
 
 - Add support for cover lock
+- Add more information to sensors and binary sensors in the HA integration
+- Store last Telegram and decoded value in RemoteValue
+
+### Breaking Changes
+
+- Remove configuration handling from core library (use https://xknx.io/config-converter)
 - Drop support for python 3.7
 
 ### Internals

docs/binary_sensor.md

@@ -33,19 +33,8 @@ binarysensor = BinarySensor(xknx, 'TestInput', group_address_state='1/2/3', devi
 ## [](#header-2)Example
 
 ```python
-outlet = Switch(xknx, 'TestOutlet', group_address='1/2/3')
-
 binarysensor = BinarySensor(xknx, 'TestInput', group_address_state='2/3/4')
-action_on = Action(
-    xknx,
-    hook='on',
-    target='TestOutlet',
-    method='on')
-binarysensor.actions.append(action_on)
-action_off = Action(
-    xknx,
-    hook='off',
-    target='TestOutlet',
-    method='off')
-binarysensor.actions.append(action_off)
+
+# Returns the last received Telegram or None
+binarysensor.last_telegram
 ```

docs/sensor.md

@@ -41,6 +41,9 @@ sensor.resolve_state()
 
 # Returns the unit of the value as string
 sensor.unit_of_measurement()
+
+# Returns the last received telegram or None
+sensor.last_telegram
 ```
 
 

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

@@ -10,7 +10,7 @@
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
-from .const import ATTR_COUNTER, DOMAIN
+from .const import ATTR_COUNTER, ATTR_LAST_KNX_UPDATE, ATTR_SOURCE, DOMAIN
 from .knx_entity import KnxEntity
 
 
@@ -51,9 +51,14 @@ def is_on(self) -> bool:
     @property
     def extra_state_attributes(self) -> dict[str, Any] | None:
         """Return device specific state attributes."""
+        attr = {}
+
         if self._device.counter is not None:
-            return {ATTR_COUNTER: self._device.counter}
-        return None
+            attr[ATTR_COUNTER] = self._device.counter
+        if self._device.last_telegram is not None:
+            attr[ATTR_SOURCE] = str(self._device.last_telegram.source_address)
+            attr[ATTR_LAST_KNX_UPDATE] = str(self._device.last_telegram.timestamp)
+        return attr
 
     @property
     def force_update(self) -> bool:

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

@@ -26,6 +26,8 @@
 CONF_RESET_AFTER = "reset_after"
 
 ATTR_COUNTER = "counter"
+ATTR_SOURCE = "source"
+ATTR_LAST_KNX_UPDATE = "last_knx_update"
 ATTR_MOVEMENT_LOCKED = "movement_locked"
 
 

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

@@ -1,7 +1,7 @@
 """Support for KNX/IP sensors."""
 from __future__ import annotations
 
-from typing import Callable, Iterable
+from typing import Any, Callable, Iterable
 
 from xknx.devices import Sensor as XknxSensor
 
@@ -10,7 +10,7 @@
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
 
-from .const import DOMAIN
+from .const import ATTR_LAST_KNX_UPDATE, ATTR_SOURCE, DOMAIN
 from .knx_entity import KnxEntity
 
 
@@ -54,6 +54,16 @@ def device_class(self) -> str | None:
             return device_class
         return None
 
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        """Return device specific state attributes."""
+        attr = {}
+
+        if self._device.last_telegram is not None:
+            attr[ATTR_SOURCE] = str(self._device.last_telegram.source_address)
+            attr[ATTR_LAST_KNX_UPDATE] = str(self._device.last_telegram.timestamp)
+        return attr
+
     @property
     def force_update(self) -> bool:
         """

test/devices_tests/binary_sensor_test.py

@@ -47,6 +47,7 @@ async def test_process(self):
             payload=GroupValueWrite(DPTBinary(0)),
         )
         await binaryinput2.process(telegram_off2)
+        assert binaryinput2.last_telegram == telegram_off2
         assert binaryinput2.state is False
 
     async def test_process_invert(self):

test/devices_tests/expose_sensor_test.py

@@ -16,35 +16,52 @@ class TestExposeSensor:
     #
     # STR FUNCTIONS
     #
-    def test_str_binary(self):
+    async def test_str_binary(self):
         """Test resolve state with binary sensor."""
         xknx = XKNX()
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", group_address="1/2/3", value_type="binary"
         )
-        expose_sensor.sensor_value.payload = DPTBinary(1)
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(value=DPTBinary(1)),
+            )
+        )
 
         assert expose_sensor.resolve_state() is True
         assert expose_sensor.unit_of_measurement() is None
 
-    def test_str_percent(self):
+    async def test_str_percent(self):
         """Test resolve state with percent sensor."""
         xknx = XKNX()
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", group_address="1/2/3", value_type="percent"
         )
-        expose_sensor.sensor_value.payload = DPTArray((0x40,))
+
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(DPTArray((0x40,))),
+            )
+        )
 
         assert expose_sensor.resolve_state() == 25
         assert expose_sensor.unit_of_measurement() == "%"
 
-    def test_str_temperature(self):
+    async def test_str_temperature(self):
         """Test resolve state with temperature sensor."""
         xknx = XKNX()
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", group_address="1/2/3", value_type="temperature"
         )
-        expose_sensor.sensor_value.payload = DPTArray((0x0C, 0x1A))
+
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(DPTArray((0x0C, 0x1A))),
+            )
+        )
 
         assert expose_sensor.resolve_state() == 21.0
         assert expose_sensor.unit_of_measurement() == "°C"
@@ -104,7 +121,13 @@ async def test_process_binary(self):
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", value_type="binary", group_address="1/2/3"
         )
-        expose_sensor.sensor_value.payload = DPTBinary(1)
+
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(DPTBinary(True)),
+            )
+        )
 
         telegram = Telegram(GroupAddress("1/2/3"), payload=GroupValueRead())
         await expose_sensor.process(telegram)
@@ -121,7 +144,13 @@ async def test_process_percent(self):
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", value_type="percent", group_address="1/2/3"
         )
-        expose_sensor.sensor_value.payload = DPTArray((0x40,))
+
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(DPTArray((0x40,))),
+            )
+        )
 
         telegram = Telegram(GroupAddress("1/2/3"), payload=GroupValueRead())
         await expose_sensor.process(telegram)
@@ -138,7 +167,13 @@ async def test_process_temperature(self):
         expose_sensor = ExposeSensor(
             xknx, "TestSensor", value_type="temperature", group_address="1/2/3"
         )
-        expose_sensor.sensor_value.payload = DPTArray((0x0C, 0x1A))
+
+        await expose_sensor.process(
+            Telegram(
+                destination_address=GroupAddress("1/2/3"),
+                payload=GroupValueWrite(DPTArray((0x0C, 0x1A))),
+            )
+        )
 
         telegram = Telegram(GroupAddress("1/2/3"), payload=GroupValueRead())
         await expose_sensor.process(telegram)