-
-
Notifications
You must be signed in to change notification settings - Fork 101
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added DPT-3s and corresponding sensor types (#167)
* Added DPTControl classes and RemoteValueControl * Update DPT mapping in RemoteValueControl * Accept some suggested changes * Fix tuple type hint for python < 3.9 * Fix keyword argument * Update xknx/remote_value/remote_value_control.py Co-authored-by: Matthias Alphart <farmio@alphart.net> * Change value_type of DPTControlStepCode * Port features from RemoteValueSensor to RemoteValueControl * Add value_type branch to sensor device * Remove asyncio.Task from remote_value_control_test * Add GroupValueWrite to Telegram payloads in test * Update changelog.md
- Loading branch information
Showing
8 changed files
with
701 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,249 @@ | ||
"""Unit test for DPTControlStepCode objects.""" | ||
import unittest | ||
|
||
from xknx.dpt import ( | ||
DPTControlStartStop, | ||
DPTControlStartStopBlinds, | ||
DPTControlStartStopDimming, | ||
DPTControlStepCode, | ||
DPTControlStepwise, | ||
) | ||
from xknx.exceptions import ConversionError | ||
|
||
|
||
class TestDPTControlStepCode(unittest.TestCase): | ||
"""Test class for DPTControlStepCode objects.""" | ||
|
||
def test_to_knx(self): | ||
"""Test serializing values to DPTControlStepCode.""" | ||
for rawref in range(16): | ||
control = 1 if rawref >> 3 else 0 | ||
raw = DPTControlStepCode.to_knx( | ||
{"control": control, "step_code": rawref & 0x07} | ||
) | ||
self.assertEqual(raw, rawref) | ||
|
||
def test_to_knx_inverted(self): | ||
"""Test serializing values to DPTControlStepCode in inverted mode.""" | ||
for rawref in range(16): | ||
control = 0 if rawref >> 3 else 1 | ||
raw = DPTControlStepCode.to_knx( | ||
{"control": control, "step_code": rawref & 0x07}, invert=True | ||
) | ||
self.assertEqual(raw, rawref) | ||
|
||
def test_to_knx_wrong_type(self): | ||
"""Test serializing wrong type to DPTControlStepCode.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx("") | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx(0) | ||
|
||
def test_to_knx_wrong_keys(self): | ||
"""Test serializing map with missing keys to DPTControlStepCode.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": 0}) | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"step_code": 0}) | ||
|
||
def test_to_knx_wrong_value_types(self): | ||
"""Test serializing map with keys of invalid type to DPTControlStepCode.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": ""}) | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"step_code": ""}) | ||
|
||
def test_to_knx_wrong_values(self): | ||
"""Test serializing map with keys of invalid values to DPTControlStepCode.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": -1, "step_code": 0}) | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": 2, "step_code": 0}) | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": 0, "step_code": -1}) | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.to_knx({"control": 0, "step_code": 0x8}) | ||
|
||
def test_from_knx(self): | ||
"""Test parsing DPTControlStepCode types from KNX.""" | ||
for raw in range(16): | ||
control = 1 if raw >> 3 else 0 | ||
valueref = {"control": control, "step_code": raw & 0x07} | ||
value = DPTControlStepCode.from_knx(raw) | ||
self.assertEqual(value, valueref) | ||
|
||
def test_from_knx_inverted(self): | ||
"""Test parsing DPTControlStepCode types from KNX.""" | ||
for raw in range(16): | ||
control = 0 if raw >> 3 else 1 | ||
valueref = {"control": control, "step_code": raw & 0x07} | ||
value = DPTControlStepCode.from_knx(raw, invert=True) | ||
self.assertEqual(value, valueref) | ||
|
||
def test_from_knx_wrong_value(self): | ||
"""Test parsing invalid DPTControlStepCode type from KNX.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepCode.from_knx(0x1F) | ||
|
||
def test_unit(self): | ||
"""Test unit_of_measurement function.""" | ||
self.assertEqual(DPTControlStepCode.unit, "") | ||
|
||
|
||
class TestDPTControlStepwise(unittest.TestCase): | ||
"""Test class for DPTControlStepwise objects.""" | ||
|
||
def test_to_knx(self): | ||
"""Test serializing values to DPTControlStepwise.""" | ||
self.assertEqual(DPTControlStepwise.to_knx(1), 0xF) | ||
self.assertEqual(DPTControlStepwise.to_knx(3), 0xE) | ||
self.assertEqual(DPTControlStepwise.to_knx(6), 0xD) | ||
self.assertEqual(DPTControlStepwise.to_knx(12), 0xC) | ||
self.assertEqual(DPTControlStepwise.to_knx(25), 0xB) | ||
self.assertEqual(DPTControlStepwise.to_knx(50), 0xA) | ||
self.assertEqual(DPTControlStepwise.to_knx(100), 0x9) | ||
self.assertEqual(DPTControlStepwise.to_knx(-1), 0x7) | ||
self.assertEqual(DPTControlStepwise.to_knx(-3), 0x6) | ||
self.assertEqual(DPTControlStepwise.to_knx(-6), 0x5) | ||
self.assertEqual(DPTControlStepwise.to_knx(-12), 0x4) | ||
self.assertEqual(DPTControlStepwise.to_knx(-25), 0x3) | ||
self.assertEqual(DPTControlStepwise.to_knx(-50), 0x2) | ||
self.assertEqual(DPTControlStepwise.to_knx(-100), 0x1) | ||
self.assertEqual(DPTControlStepwise.to_knx(0), 0x0) | ||
|
||
def test_to_knx_wrong_type(self): | ||
"""Test serializing wrong type to DPTControlStepwise.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepwise.to_knx("") | ||
|
||
def test_from_knx(self): | ||
"""Test parsing DPTControlStepwise types from KNX.""" | ||
self.assertEqual(DPTControlStepwise.from_knx(0xF), 1) | ||
self.assertEqual(DPTControlStepwise.from_knx(0xE), 3) | ||
self.assertEqual(DPTControlStepwise.from_knx(0xD), 6) | ||
self.assertEqual(DPTControlStepwise.from_knx(0xC), 12) | ||
self.assertEqual(DPTControlStepwise.from_knx(0xB), 25) | ||
self.assertEqual(DPTControlStepwise.from_knx(0xA), 50) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x9), 100) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x8), 0) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x7), -1) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x6), -3) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x5), -6) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x4), -12) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x3), -25) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x2), -50) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x1), -100) | ||
self.assertEqual(DPTControlStepwise.from_knx(0x0), 0) | ||
|
||
def test_from_knx_wrong_value(self): | ||
"""Test parsing invalid DPTControlStepwise type from KNX.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStepwise.from_knx(0x1F) | ||
|
||
def test_unit(self): | ||
"""Test unit_of_measurement function.""" | ||
self.assertEqual(DPTControlStepwise.unit, "%") | ||
|
||
|
||
class TestDPTControlStartStop(unittest.TestCase): | ||
"""Test class for DPTControlStartStop objects.""" | ||
|
||
def test_mode_to_knx(self): | ||
"""Test serializing dimming commands to KNX.""" | ||
self.assertEqual( | ||
DPTControlStartStopDimming.to_knx( | ||
DPTControlStartStopDimming.Direction.INCREASE | ||
), | ||
9, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopDimming.to_knx( | ||
DPTControlStartStopDimming.Direction.DECREASE | ||
), | ||
1, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopDimming.to_knx( | ||
DPTControlStartStopDimming.Direction.STOP | ||
), | ||
0, | ||
) | ||
|
||
def test_mode_to_knx_wrong_value(self): | ||
"""Test serializing invalid data type to KNX.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStartStopDimming.to_knx(1) | ||
|
||
def test_mode_from_knx(self): | ||
"""Test parsing dimming commands from KNX.""" | ||
for i in range(16): | ||
if i > 8: | ||
expected_direction = DPTControlStartStopDimming.Direction.INCREASE | ||
elif i in (0, 8): | ||
expected_direction = DPTControlStartStopDimming.Direction.STOP | ||
elif i < 8: | ||
expected_direction = DPTControlStartStopDimming.Direction.DECREASE | ||
self.assertEqual(DPTControlStartStopDimming.from_knx(i), expected_direction) | ||
|
||
def test_mode_from_knx_wrong_value(self): | ||
"""Test serializing invalid data type to KNX.""" | ||
with self.assertRaises(ConversionError): | ||
DPTControlStartStopDimming.from_knx((1, 2)) | ||
|
||
def test_direction_names(self): | ||
"""Test names of Direction Enum.""" | ||
self.assertEqual(str(DPTControlStartStop.Direction.INCREASE), "Increase") | ||
self.assertEqual(str(DPTControlStartStop.Direction.DECREASE), "Decrease") | ||
self.assertEqual(str(DPTControlStartStop.Direction.STOP), "Stop") | ||
|
||
|
||
class TestDPTControlStartStopDimming(TestDPTControlStartStop): | ||
"""Test class for DPTControlStartStopDimming objects.""" | ||
|
||
def test_direction_names(self): | ||
"""Test names of Direction Enum.""" | ||
self.assertEqual(str(DPTControlStartStopDimming.Direction.INCREASE), "Increase") | ||
self.assertEqual(str(DPTControlStartStopDimming.Direction.DECREASE), "Decrease") | ||
self.assertEqual(str(DPTControlStartStopDimming.Direction.STOP), "Stop") | ||
|
||
def test_direction_values(self): | ||
"""Test values of Direction Enum.""" | ||
# pylint: disable=no-member | ||
self.assertEqual( | ||
DPTControlStartStopDimming.Direction.DECREASE.value, | ||
DPTControlStartStop.Direction.DECREASE.value, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopDimming.Direction.INCREASE.value, | ||
DPTControlStartStop.Direction.INCREASE.value, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopDimming.Direction.STOP.value, | ||
DPTControlStartStop.Direction.STOP.value, | ||
) | ||
|
||
|
||
class TestDPTControlStartStopBlinds(unittest.TestCase): | ||
"""Test class for DPTControlStartStopBlinds objects.""" | ||
|
||
def test_direction_names(self): | ||
"""Test names of Direction Enum.""" | ||
self.assertEqual(str(DPTControlStartStopBlinds.Direction.DOWN), "Down") | ||
self.assertEqual(str(DPTControlStartStopBlinds.Direction.UP), "Up") | ||
self.assertEqual(str(DPTControlStartStopBlinds.Direction.STOP), "Stop") | ||
|
||
def test_direction_values(self): | ||
"""Test values of Direction Enum.""" | ||
# pylint: disable=no-member | ||
self.assertEqual( | ||
DPTControlStartStopBlinds.Direction.UP.value, | ||
DPTControlStartStop.Direction.DECREASE.value, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopBlinds.Direction.DOWN.value, | ||
DPTControlStartStop.Direction.INCREASE.value, | ||
) | ||
self.assertEqual( | ||
DPTControlStartStopBlinds.Direction.STOP.value, | ||
DPTControlStartStop.Direction.STOP.value, | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,86 @@ | ||
"""Unit test for RemoteValueControl objects.""" | ||
import asyncio | ||
import unittest | ||
|
||
from xknx import XKNX | ||
from xknx.dpt import DPTArray, DPTBinary | ||
from xknx.exceptions import ConversionError, CouldNotParseTelegram | ||
from xknx.remote_value import RemoteValueControl | ||
from xknx.telegram import GroupAddress, Telegram | ||
from xknx.telegram.apci import GroupValueWrite | ||
|
||
|
||
class TestRemoteValueControl(unittest.TestCase): | ||
"""Test class for RemoteValueControl objects.""" | ||
|
||
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_wrong_value_type(self): | ||
"""Test initializing with wrong value_type.""" | ||
xknx = XKNX() | ||
with self.assertRaises(ConversionError): | ||
RemoteValueControl(xknx, value_type="wrong_value_type") | ||
|
||
def test_valid_payload(self): | ||
"""Test valid_payload method.""" | ||
self.assertTrue(DPTBinary(0)) | ||
self.assertTrue(DPTArray([0])) | ||
|
||
def test_set(self): | ||
"""Test setting value.""" | ||
xknx = XKNX() | ||
remote_value = RemoteValueControl( | ||
xknx, group_address=GroupAddress("1/2/3"), value_type="stepwise" | ||
) | ||
self.loop.run_until_complete(asyncio.Task(remote_value.set(25))) | ||
self.assertEqual(xknx.telegrams.qsize(), 1) | ||
telegram = xknx.telegrams.get_nowait() | ||
self.assertEqual( | ||
telegram, | ||
Telegram( | ||
destination_address=GroupAddress("1/2/3"), | ||
payload=GroupValueWrite(DPTBinary(0xB)), | ||
), | ||
) | ||
|
||
def test_process(self): | ||
"""Test process telegram.""" | ||
xknx = XKNX() | ||
remote_value = RemoteValueControl( | ||
xknx, group_address=GroupAddress("1/2/3"), value_type="stepwise" | ||
) | ||
telegram = Telegram( | ||
destination_address=GroupAddress("1/2/3"), | ||
payload=GroupValueWrite(DPTBinary(0xB)), | ||
) | ||
self.assertEqual(remote_value.value, None) | ||
self.loop.run_until_complete(remote_value.process(telegram)) | ||
self.assertEqual(remote_value.value, 25) | ||
|
||
def test_to_process_error(self): | ||
"""Test process errornous telegram.""" | ||
xknx = XKNX() | ||
remote_value = RemoteValueControl( | ||
xknx, group_address=GroupAddress("1/2/3"), value_type="stepwise" | ||
) | ||
with self.assertRaises(CouldNotParseTelegram): | ||
telegram = Telegram( | ||
destination_address=GroupAddress("1/2/3"), | ||
payload=GroupValueWrite(DPTArray(0x01)), | ||
) | ||
self.loop.run_until_complete(remote_value.process(telegram)) | ||
with self.assertRaises(ConversionError): | ||
telegram = Telegram( | ||
destination_address=GroupAddress("1/2/3"), | ||
payload=GroupValueWrite(DPTBinary(0x10)), | ||
) | ||
self.loop.run_until_complete(remote_value.process(telegram)) | ||
# pylint: disable=pointless-statement | ||
remote_value.value |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.