Repository: XKNX/xknx
Pull request #721 (2021-06-17 21:04:55)

Add RawValue device
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
Add RawValue device.
This can be used in a future HA select entity.

Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] I have performed a self-review of my own code
- [x] The documentation has been adjusted accordingly
- [x] The changes generate no new warnings
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/721?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#721](https://codecov.io/gh/XKNX/xknx/pull/721?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (d835ab9) into [main](https://codecov.io/gh/XKNX/xknx/commit/0410fc74d6d09c66b3b5e37994d883db95d2d9c7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (0410fc7) will **increase** coverage by `0.01%`.
> The diff coverage is `96.87%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/721/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/721?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #721      +/-   ##
==========================================
+ Coverage   94.59%   94.61%   +0.01%     
==========================================
  Files         104      106       +2     
  Lines        6573     6631      +58     
==========================================
+ Hits         6218     6274      +56     
- Misses        355      357       +2     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/721?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/devices/numeric\_value.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL251bWVyaWNfdmFsdWUucHk=) | `100.00% <ø> (ø)` | |
| [xknx/remote\_value/remote\_value\_raw.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9yZW1vdGVfdmFsdWUvcmVtb3RlX3ZhbHVlX3Jhdy5weQ==) | `93.75% <93.75%> (ø)` | |
| [xknx/devices/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL19faW5pdF9fLnB5) | `100.00% <100.00%> (ø)` | |
| [xknx/devices/raw\_value.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL3Jhd192YWx1ZS5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/dpt/dpt.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kcHQvZHB0LnB5) | `100.00% <100.00%> (ø)` | |
| [xknx/remote\_value/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9yZW1vdGVfdmFsdWUvX19pbml0X18ucHk=) | `100.00% <100.00%> (ø)` | |
| [xknx/remote\_value/remote\_value.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9yZW1vdGVfdmFsdWUvcmVtb3RlX3ZhbHVlLnB5) | `100.00% <100.00%> (ø)` | |
| [xknx/remote\_value/remote\_value\_climate\_mode.py](https://codecov.io/gh/XKNX/xknx/pull/721/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9yZW1vdGVfdmFsdWUvcmVtb3RlX3ZhbHVlX2NsaW1hdGVfbW9kZS5weQ==) | `97.72% <100.00%> (-0.03%)` | :arrow_down: |


Patch:
diff --git a/changelog.md b/changelog.md
index 40650fca..31f0236c 100644
--- a/changelog.md
+++ b/changelog.md
@@ -4,6 +4,7 @@
 
 ### Devices
 
+- Add RawValue device.
 - Remove unused HA-specific attributes (unique_id, device_class, create_sensors)
 - Climate: add `group_address_active_state`, `group_address_command_value_state` and a `is_active` property.
 - Configurable `sync_state` in all devices.
diff --git a/docs/numeric_value.md b/docs/numeric_value.md
index f0ad62d1..472fc41e 100644
--- a/docs/numeric_value.md
+++ b/docs/numeric_value.md
@@ -34,17 +34,17 @@ value = NumericValue(
 )
 
 # Set a value without sending to the bus
-sensor.sensor_value.value = 23.0
+value.sensor_value.value = 23.0
 
 # Send a new value to the bus
 await value.set(24.0)
 
 # Returns the value of in a human readable way
-sensor.resolve_state()
+value.resolve_state()
 
 # Returns the unit of the value as string
-sensor.unit_of_measurement()
+value.unit_of_measurement()
 
 # Returns the last received telegram or None
-sensor.last_telegram
+value.last_telegram
 ```
diff --git a/docs/raw_value.md b/docs/raw_value.md
new file mode 100644
index 00000000..a0854dd9
--- /dev/null
+++ b/docs/raw_value.md
@@ -0,0 +1,47 @@
+---
+layout: default
+title: RawValue
+parent: Devices
+nav_order: 7
+---
+
+# [](#header-1)RawValue - Send and receive raw values over KNX
+
+## [](#header-2)Overview
+
+RawValue devices send uint values to the KNX bus. Received values update the devices state. Optionally the current state can be read from the KNX bus.
+
+## [](#header-2)Interface
+
+- `xknx` is the XKNX object.
+- `name` is the name of the object.
+- `payload_length` is the appended byte size to a CEMI-Frame. `0` for DPT 1, 2 and 3.
+- `group_address` is the KNX group address of the raw value device. Used for sending.
+- `group_address_state` is the KNX group address of the raw value device.
+- `respond_to_read` if `True` GroupValueRead requests to the `group_address` are answered. Defaults to `False`
+- `sync_state` defines if and how often the value should be actively read from the bus. If `False` no GroupValueRead telegrams will be sent to its group address. Defaults to `True`
+- `device_updated_cb` awaitable callback for each update.
+
+## [](#header-2)Example
+
+```python
+value = RawValue(
+    xknx=xknx,
+    name='Raw',
+    payload_length=2,
+    group_address='6/2/1',
+    respond_to_read=True,
+)
+
+# Set a value without sending to the bus
+value.remote_value.value = 23.0
+
+# Send a new value to the bus
+await value.set(24.0)
+
+# Returns the value of in a human readable way
+value.resolve_state()
+
+# Returns the last received telegram or None
+value.last_telegram
+```
diff --git a/docs/sensor.md b/docs/sensor.md
index 1dca5187..93788d8c 100644
--- a/docs/sensor.md
+++ b/docs/sensor.md
@@ -2,7 +2,7 @@
 layout: default
 title: Sensor
 parent: Devices
-nav_order: 7
+nav_order: 8
 ---
 
 # [](#header-1)Sensor - Monitor values of KNX
diff --git a/docs/switch.md b/docs/switch.md
index 8284cccf..b53a1ede 100644
--- a/docs/switch.md
+++ b/docs/switch.md
@@ -2,7 +2,7 @@
 layout: default
 title: Switches
 parent: Devices
-nav_order: 8
+nav_order: 9
 ---
 
 # [](#header-1)Switches
diff --git a/docs/time.md b/docs/time.md
index 04bea417..0e4f8f6e 100644
--- a/docs/time.md
+++ b/docs/time.md
@@ -2,7 +2,7 @@
 layout: default
 title: Time
 parent: Devices
-nav_order: 9
+nav_order: 10
 ---
 
 # [](#header-1)Time
diff --git a/docs/weather.md b/docs/weather.md
index 16da52cf..59e2d02c 100644
--- a/docs/weather.md
+++ b/docs/weather.md
@@ -2,7 +2,7 @@
 layout: default
 title: Weather
 parent: Devices
-nav_order: 10
+nav_order: 11
 ---
 
 # [](#header-1)Weather device
diff --git a/xknx/devices/__init__.py b/xknx/devices/__init__.py
index 3aa7bb02..75845132 100644
--- a/xknx/devices/__init__.py
+++ b/xknx/devices/__init__.py
@@ -12,6 +12,7 @@ from .fan import Fan
 from .light import Light
 from .notification import Notification
 from .numeric_value import NumericValue
+from .raw_value import RawValue
 from .scene import Scene
 from .sensor import Sensor
 from .switch import Switch
@@ -31,6 +32,7 @@ __all__ = [
     "Light",
     "Notification",
     "NumericValue",
+    "RawValue",
     "Scene",
     "Sensor",
     "Switch",
diff --git a/xknx/devices/numeric_value.py b/xknx/devices/numeric_value.py
index e77d2db4..3767b2cb 100644
--- a/xknx/devices/numeric_value.py
+++ b/xknx/devices/numeric_value.py
@@ -3,7 +3,7 @@ Module for managing a generic numeric value via KNX.
 
 It provides functionality for
 
-* readingthe current state from KNX bus or providing its state to the bus.
+* reading the current state from KNX bus or providing its state to the bus.
 * send local state changes to KNX bus.
 * watching for state updates from KNX bus.
 """
diff --git a/xknx/devices/raw_value.py b/xknx/devices/raw_value.py
new file mode 100644
index 00000000..cd37cd25
--- /dev/null
+++ b/xknx/devices/raw_value.py
@@ -0,0 +1,85 @@
+"""
+Module for managing a raw value via KNX.
+
+It provides functionality for
+
+* reading the current value from KNX bus or providing a value to the bus.
+* send local state changes to KNX bus.
+* watching for state updates from KNX bus.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Iterator
+
+from xknx.remote_value import GroupAddressesType, RemoteValueRaw
+
+from .device import Device, DeviceCallbackType
+
+if TYPE_CHECKING:
+    from xknx.telegram import Telegram
+    from xknx.xknx import XKNX
+
+
+class RawValue(Device):
+    """Class for managing a raw value."""
+
+    def __init__(
+        self,
+        xknx: XKNX,
+        name: str,
+        payload_length: int,
+        group_address: GroupAddressesType | None = None,
+        group_address_state: GroupAddressesType | None = None,
+        respond_to_read: bool = False,
+        sync_state: bool | int | float | str = True,
+        device_updated_cb: DeviceCallbackType | None = None,
+    ):
+        """Initialize Sensor class."""
+        super().__init__(xknx, name, device_updated_cb)
+        self.respond_to_read = respond_to_read
+        self.remote_value = RemoteValueRaw(
+            xknx,
+            payload_length=payload_length,
+            group_address=group_address,
+            group_address_state=group_address_state,
+            sync_state=sync_state,
+            device_name=self.name,
+            after_update_cb=self.after_update,
+        )
+
+    def _iter_remote_values(self) -> Iterator[RemoteValueRaw]:
+        """Iterate the devices RemoteValue classes."""
+        yield self.remote_value
+
+    @property
+    def last_telegram(self) -> Telegram | None:
+        """Return the last telegram received from the RemoteValue."""
+        return self.remote_value.telegram
+
+    async def process_group_write(self, telegram: "Telegram") -> None:
+        """Process incoming and outgoing GROUP WRITE telegram."""
+        await self.remote_value.process(telegram)
+
+    async def process_group_read(self, telegram: "Telegram") -> None:
+        """Process incoming GroupValueResponse telegrams."""
+        if (
+            self.respond_to_read
+            and telegram.destination_address == self.remote_value.group_address
+        ):
+            await self.remote_value.respond()
+
+    async def set(self, value: int) -> None:
+        """Set new value."""
+        await self.remote_value.set(value)
+
+    def resolve_state(self) -> int | None:
+        """Return the current state of the sensor as an unsigned integer."""
+        return self.remote_value.value
+
+    def __str__(self) -> str:
+        """Return object as readable string."""
+        return '<RawValue name="{}" addresses={} value={}/>'.format(
+            self.name,
+            self.remote_value.group_addr_str(),
+            self.resolve_state().__repr__(),
+        )
diff --git a/xknx/dpt/dpt.py b/xknx/dpt/dpt.py
index 48d78c4c..4595afe5 100644
--- a/xknx/dpt/dpt.py
+++ b/xknx/dpt/dpt.py
@@ -7,7 +7,6 @@ from typing import Any, Iterator, Type, TypeVar, cast
 
 from xknx.exceptions import ConversionError
 
-DPTPayloadType = TypeVar("DPTPayloadType", "DPTArray", "DPTBinary")
 T = TypeVar("T", bound=Type["DPTBase"])  # pylint: disable=invalid-name
 
 
@@ -171,7 +170,7 @@ class DPTBinary:
             value = value[0]
         if not isinstance(value, int):
             raise TypeError()
-        if value > DPTBinary.APCI_BITMASK:
+        if value > DPTBinary.APCI_BITMASK or value < 0:
             raise ConversionError("Could not init DPTBinary", value=str(value))
 
         self.value = value
@@ -200,9 +199,7 @@ class DPTArray:
         if isinstance(value, int):
             self.value = (value,)
         elif isinstance(value, (list, bytes)):
-            self.value = tuple(
-                value,
-            )
+            self.value = tuple(value)
         elif isinstance(value, tuple):
             self.value = value
         else:
diff --git a/xknx/remote_value/__init__.py b/xknx/remote_value/__init__.py
index b52a244d..ba8c3a19 100644
--- a/xknx/remote_value/__init__.py
+++ b/xknx/remote_value/__init__.py
@@ -15,6 +15,7 @@ from .remote_value_control import RemoteValueControl
 from .remote_value_datetime import RemoteValueDateTime
 from .remote_value_dpt_2_byte_unsigned import RemoteValueDpt2ByteUnsigned
 from .remote_value_dpt_value_1_ucount import RemoteValueDptValue1Ucount
+from .remote_value_raw import RemoteValueRaw
 from .remote_value_scaling import RemoteValueScaling
 from .remote_value_scene_number import RemoteValueSceneNumber
 from .remote_value_sensor import RemoteValueNumeric, RemoteValueSensor
@@ -41,6 +42,7 @@ __all__ = [
     "RemoteValueDptValue1Ucount",
     "RemoteValueNumeric",
     "RemoteValueOperationMode",
+    "RemoteValueRaw",
     "RemoteValueScaling",
     "RemoteValueSceneNumber",
     "RemoteValueSensor",
diff --git a/xknx/remote_value/remote_value.py b/xknx/remote_value/remote_value.py
index 3878ac03..859a89bf 100644
--- a/xknx/remote_value/remote_value.py
+++ b/xknx/remote_value/remote_value.py
@@ -21,7 +21,7 @@ from typing import (
     Union,
 )
 
-from xknx.dpt.dpt import DPTArray, DPTBinary, DPTPayloadType
+from xknx.dpt.dpt import DPTArray, DPTBinary
 from xknx.exceptions import CouldNotParseTelegram
 from xknx.telegram import GroupAddress, Telegram
 from xknx.telegram.address import (
@@ -38,6 +38,9 @@ if TYPE_CHECKING:
 logger = logging.getLogger("xknx.log")
 
 AsyncCallbackType = Callable[[], Awaitable[None]]
+DPTPayloadType = TypeVar(
+    "DPTPayloadType", DPTArray, DPTBinary, Union[DPTArray, DPTBinary]
+)
 GroupAddressesType = Union["DeviceAddressableType", List["DeviceAddressableType"]]
 ValueType = TypeVar("ValueType")
 
diff --git a/xknx/remote_value/remote_value_climate_mode.py b/xknx/remote_value/remote_value_climate_mode.py
index d3d83716..2e6b6671 100644
--- a/xknx/remote_value/remote_value_climate_mode.py
+++ b/xknx/remote_value/remote_value_climate_mode.py
@@ -16,11 +16,15 @@ from xknx.dpt import (
     DPTHVACContrMode,
     DPTHVACMode,
 )
-from xknx.dpt.dpt import DPTPayloadType
 from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACModeType, HVACOperationMode
 from xknx.exceptions import ConversionError, CouldNotParseTelegram
 
-from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
+from .remote_value import (
+    AsyncCallbackType,
+    DPTPayloadType,
+    GroupAddressesType,
+    RemoteValue,
+)
 
 if TYPE_CHECKING:
     from xknx.xknx import XKNX
diff --git a/xknx/remote_value/remote_value_raw.py b/xknx/remote_value/remote_value_raw.py
new file mode 100644
index 00000000..57145338
--- /dev/null
+++ b/xknx/remote_value/remote_value_raw.py
@@ -0,0 +1,77 @@
+"""
+Module for managing a remote value typically used within a sensor.
+
+The module maps a given value_type to a DPT class and uses this class
+for serialization and deserialization of the KNX value.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Union
+
+from xknx.dpt.dpt import DPTArray, DPTBinary
+from xknx.exceptions import ConversionError
+
+from .remote_value import AsyncCallbackType, GroupAddressesType, RemoteValue
+
+if TYPE_CHECKING:
+    from xknx.xknx import XKNX
+
+
+class RemoteValueRaw(RemoteValue[Union[DPTArray, DPTBinary], int]):
+    """Abstraction for raw values."""
+
+    def __init__(
+        self,
+        xknx: XKNX,
+        payload_length: int,
+        group_address: GroupAddressesType | None = None,
+        group_address_state: GroupAddressesType | None = None,
+        sync_state: bool | int | float | str = True,
+        device_name: str | None = None,
+        feature_name: str = "Raw",
+        after_update_cb: AsyncCallbackType | None = None,
+    ):
+        """Initialize RemoteValueSensor class."""
+        self.payload_length = payload_length
+        super().__init__(
+            xknx,
+            group_address,
+            group_address_state,
+            sync_state=sync_state,
+            device_name=device_name,
+            feature_name=feature_name,
+            after_update_cb=after_update_cb,
+        )
+
+    def payload_valid(
+        self, payload: DPTArray | DPTBinary | None
+    ) -> DPTArray | DPTBinary | None:
+        """Test if telegram payload may be parsed."""
+        if isinstance(payload, DPTBinary) and self.payload_length == 0:
+            return payload
+        if isinstance(payload, DPTArray) and len(payload.value) == self.payload_length:
+            return payload
+        return None
+
+    def to_knx(self, value: int) -> DPTArray | DPTBinary:
+        """Convert value to payload."""
+        if self.payload_length == 0:
+            try:
+                return DPTBinary(value)
+            except TypeError as err:
+                raise ConversionError(
+                    "Could not init DPTBinary", value=str(value)
+                ) from err
+        try:
+            return DPTArray(value.to_bytes(length=self.payload_length, byteorder="big"))
+        except (AttributeError, OverflowError) as err:
+            raise ConversionError("Could not init DPTArray", value=str(value)) from err
+
+    def from_knx(self, payload: DPTArray | DPTBinary) -> int:
+        """Convert current payload to value."""
+        if isinstance(payload, DPTBinary):
+            return payload.value
+        try:
+            return int.from_bytes(payload.value, byteorder="big")
+        except ValueError as err:
+            raise ConversionError("Could not parse payload", payload=payload) from err


Test patch:
diff --git a/test/devices_tests/raw_value_test.py b/test/devices_tests/raw_value_test.py
new file mode 100644
index 00000000..469fdd59
--- /dev/null
+++ b/test/devices_tests/raw_value_test.py
@@ -0,0 +1,178 @@
+"""Unit test for RawValue objects."""
+import pytest
+from xknx import XKNX
+from xknx.devices import RawValue
+from xknx.dpt import DPTArray, DPTBinary
+from xknx.telegram import GroupAddress, Telegram
+from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
+
+
+@pytest.mark.asyncio
+class TestRawValue:
+    """Test class for RawValue objects."""
+
+    @pytest.mark.parametrize(
+        "payload_length,raw_payload,expected_state",
+        [
+            # DPT-14 values are according to ETS group monitor values
+            (
+                0,
+                DPTBinary(0),
+                0,
+            ),
+            (
+                0,
+                DPTBinary(True),
+                1,
+            ),
+            (
+                0,
+                DPTBinary(45),
+                45,
+            ),
+            (
+                1,
+                DPTArray((0x6B,)),
+                107,
+            ),
+            (
+                1,
+                DPTArray((0xFC,)),
+                252,
+            ),
+            (
+                2,
+                DPTArray((0x6C, 0x95)),
+                27797,
+            ),
+        ],
+    )
+    async def test_payloads(
+        self,
+        payload_length,
+        raw_payload,
+        expected_state,
+    ):
+        """Test raw value types."""
+        xknx = XKNX()
+        raw_value = RawValue(
+            xknx,
+            "Test",
+            payload_length=payload_length,
+            group_address="1/2/3",
+        )
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(value=raw_payload),
+        )
+        await raw_value.process(telegram)
+        assert raw_value.resolve_state() == expected_state
+        assert raw_value.last_telegram == telegram
+
+    #
+    # TEST RESPOND
+    #
+    async def test_respond_to_read(self):
+        """Test respond_to_read function."""
+        xknx = XKNX()
+        responding = RawValue(
+            xknx,
+            "TestSensor1",
+            2,
+            group_address="1/1/1",
+            respond_to_read=True,
+        )
+        non_responding = RawValue(
+            xknx,
+            "TestSensor2",
+            2,
+            group_address="1/1/1",
+            respond_to_read=False,
+        )
+        responding_multiple = RawValue(
+            xknx,
+            "TestSensor3",
+            2,
+            group_address=["1/1/1", "3/3/3"],
+            group_address_state="2/2/2",
+            respond_to_read=True,
+        )
+        #  set initial payload of RawValue
+        responding.remote_value.value = 256
+        non_responding.remote_value.value = 256
+        responding_multiple.remote_value.value = 256
+
+        read_telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"), payload=GroupValueRead()
+        )
+        # verify no response when respond is False
+        await non_responding.process(read_telegram)
+        assert xknx.telegrams.qsize() == 0
+
+        # verify response when respond is True
+        await responding.process(read_telegram)
+        assert xknx.telegrams.qsize() == 1
+        response = xknx.telegrams.get_nowait()
+        assert response == Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            payload=GroupValueResponse(DPTArray((0x01, 0x00))),
+        )
+        # verify no response when GroupValueRead request is not for group_address
+        await responding_multiple.process(read_telegram)
+        assert xknx.telegrams.qsize() == 1
+        response = xknx.telegrams.get_nowait()
+        assert response == Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            payload=GroupValueResponse(DPTArray((0x01, 0x00))),
+        )
+        await responding_multiple.process(
+            Telegram(
+                destination_address=GroupAddress("2/2/2"), payload=GroupValueRead()
+            )
+        )
+        await responding_multiple.process(
+            Telegram(
+                destination_address=GroupAddress("3/3/3"), payload=GroupValueRead()
+            )
+        )
+        assert xknx.telegrams.qsize() == 0
+
+    #
+    # TEST SET
+    #
+    async def test_set_0(self):
+        """Test set with raw value."""
+        xknx = XKNX()
+        raw_value = RawValue(xknx, "TestSensor", 0, group_address="1/2/3")
+        await raw_value.set(True)
+        assert xknx.telegrams.qsize() == 1
+
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTBinary(True)),
+        )
+
+    async def test_set_1(self):
+        """Test set with raw value."""
+        xknx = XKNX()
+        raw_value = RawValue(xknx, "TestSensor", 1, group_address="1/2/3")
+        await raw_value.set(75)
+        assert xknx.telegrams.qsize() == 1
+
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTArray((0x4B,))),
+        )
+
+    def test_string(self):
+        """Test RawValue string representation."""
+
+        xknx = XKNX()
+        value = RawValue(xknx, "Raw", 1, group_address="1/2/3")
+        value.remote_value.value = 4
+        assert (
+            str(value)
+            == '<RawValue name="Raw" addresses=<1/2/3, None, [], 4 /> value=4/>'
+        )
diff --git a/test/remote_value_tests/remote_value_raw_test.py b/test/remote_value_tests/remote_value_raw_test.py
new file mode 100644
index 00000000..936b922c
--- /dev/null
+++ b/test/remote_value_tests/remote_value_raw_test.py
@@ -0,0 +1,200 @@
+"""Unit test for RemoteValueRaw objects."""
+import pytest
+from xknx import XKNX
+from xknx.dpt import DPTArray, DPTBinary
+from xknx.exceptions import ConversionError, CouldNotParseTelegram
+from xknx.remote_value import RemoteValueRaw
+from xknx.telegram import GroupAddress, Telegram
+from xknx.telegram.apci import GroupValueWrite
+
+
+@pytest.mark.asyncio
+class TestRemoteValueRaw:
+    """Test class for RemoteValueRaw objects."""
+
+    def test_to_knx(self):
+        """Test to_knx function with normal operation."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0)
+        rv_1 = RemoteValueRaw(xknx, payload_length=1)
+        rv_2 = RemoteValueRaw(xknx, payload_length=2)
+
+        assert rv_0.to_knx(1) == DPTBinary(True)
+        assert rv_0.to_knx(4) == DPTBinary(4)
+        assert rv_1.to_knx(100) == DPTArray((0x64,))
+        assert rv_2.to_knx(100) == DPTArray((0x00, 0x64))
+
+    def test_from_knx(self):
+        """Test from_knx function with normal operation."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0)
+        rv_1 = RemoteValueRaw(xknx, payload_length=1)
+        rv_2 = RemoteValueRaw(xknx, payload_length=2)
+
+        assert rv_0.from_knx(DPTBinary(True)) == 1
+        assert rv_0.from_knx(DPTBinary(0x4)) == 4
+        assert rv_1.from_knx(DPTArray((0x64,))) == 100
+        assert rv_2.from_knx(DPTArray((0x00, 0x64))) == 100
+
+        with pytest.raises(ConversionError):
+            assert rv_1.from_knx(DPTArray((256,)))
+
+    async def test_set(self):
+        """Test setting value."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/2/3")
+        rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/2/3")
+        rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/3")
+
+        await rv_0.set(0)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTBinary(False)),
+        )
+        await rv_0.set(63)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTBinary(0x3F)),
+        )
+
+        await rv_1.set(0)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTArray((0x00,))),
+        )
+        await rv_1.set(63)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTArray((0x3F,))),
+        )
+
+        await rv_2.set(0)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTArray((0x00, 0x00))),
+        )
+        await rv_2.set(63)
+        assert xknx.telegrams.qsize() == 1
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueWrite(DPTArray((0x00, 0x3F))),
+        )
+
+    async def test_process(self):
+        """Test process telegram."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
+        rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
+        rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/0/0"),
+            payload=GroupValueWrite(DPTBinary(0)),
+        )
+        await rv_0.process(telegram)
+        assert rv_0.value == 0
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            payload=GroupValueWrite(DPTArray((0x64,))),
+        )
+        await rv_1.process(telegram)
+        assert rv_1.value == 100
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/2"),
+            payload=GroupValueWrite(DPTArray((0x12, 0x34))),
+        )
+        await rv_2.process(telegram)
+        assert rv_2.value == 4660
+
+    async def test_to_process_error(self):
+        """Test process errornous telegram."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
+        rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
+        rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
+
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/0/0"),
+                payload=GroupValueWrite(DPTArray((0x01,))),
+            )
+            await rv_0.process(telegram)
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/0/0"),
+                payload=GroupValueWrite(DPTArray((0x64, 0x65))),
+            )
+            await rv_0.process(telegram)
+
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/1/1"),
+                payload=GroupValueWrite(DPTBinary(1)),
+            )
+            await rv_1.process(telegram)
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/1/1"),
+                payload=GroupValueWrite(DPTArray((0x64, 0x65))),
+            )
+            await rv_1.process(telegram)
+
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/2/2"),
+                payload=GroupValueWrite(DPTBinary(1)),
+            )
+            await rv_2.process(telegram)
+        with pytest.raises(CouldNotParseTelegram):
+            telegram = Telegram(
+                destination_address=GroupAddress("1/2/2"),
+                payload=GroupValueWrite(DPTArray((0x64,))),
+            )
+            await rv_2.process(telegram)
+
+    def test_to_knx_error(self):
+        """Test to_knx function with wrong parameters."""
+        xknx = XKNX()
+        rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
+        rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
+        rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
+
+        with pytest.raises(ConversionError):
+            rv_0.to_knx(-1)
+        with pytest.raises(ConversionError):
+            rv_0.to_knx(64)
+        with pytest.raises(ConversionError):
+            rv_0.to_knx(5.5)
+        with pytest.raises(ConversionError):
+            rv_0.to_knx("a")
+
+        with pytest.raises(ConversionError):
+            rv_1.to_knx(-1)
+        with pytest.raises(ConversionError):
+            rv_1.to_knx(256)
+        with pytest.raises(ConversionError):
+            rv_1.to_knx(5.5)
+        with pytest.raises(ConversionError):
+            rv_1.to_knx("a")
+
+        with pytest.raises(ConversionError):
+            rv_2.to_knx(-1)
+        with pytest.raises(ConversionError):
+            rv_2.to_knx(65536)
+        with pytest.raises(ConversionError):
+            rv_2.to_knx(5.5)
+        with pytest.raises(ConversionError):
+            rv_2.to_knx("a")