diff --git a/changelog.md b/changelog.md
index da63d2e15..e61724dd3 100644
--- a/changelog.md
+++ b/changelog.md
@@ -2,9 +2,13 @@
## Unreleased changes
+### HA integration
+
+- knx_event: renamed `address` to `destination` and added `source`, `telegramtype`, `direction` attributes.
+
### Internals
-- Telegram: `group_address` renamed to `destination_address`, to prepare support for other APCI services.
+- Telegram: `group_address` renamed to `destination_address`, to prepare support for other APCI services and add `source_address`
- Farewell Travis CI; Welcome Github Actions!
## 0.15.6 Bugfix for StateUpater 2020-11-26
diff --git a/examples/example_tunnel.py b/examples/example_tunnel.py
index 090df9db6..e64611252 100644
--- a/examples/example_tunnel.py
+++ b/examples/example_tunnel.py
@@ -37,9 +37,13 @@ async def main():
await tunnel.connect_udp()
await tunnel.connect()
- await tunnel.send_telegram(Telegram(GroupAddress("1/0/15"), payload=DPTBinary(1)))
+ await tunnel.send_telegram(
+ Telegram(destination_address=GroupAddress("1/0/15"), payload=DPTBinary(1))
+ )
await asyncio.sleep(2)
- await tunnel.send_telegram(Telegram(GroupAddress("1/0/15"), payload=DPTBinary(0)))
+ await tunnel.send_telegram(
+ Telegram(destination_address=GroupAddress("1/0/15"), payload=DPTBinary(0))
+ )
await asyncio.sleep(2)
await tunnel.connectionstate()
diff --git a/home-assistant-plugin/custom_components/xknx/__init__.py b/home-assistant-plugin/custom_components/xknx/__init__.py
index 5a788491e..4b82e1d31 100644
--- a/home-assistant-plugin/custom_components/xknx/__init__.py
+++ b/home-assistant-plugin/custom_components/xknx/__init__.py
@@ -325,8 +325,11 @@ async def telegram_received_cb(self, telegram):
self.hass.bus.async_fire(
"knx_event",
{
- "address": str(telegram.destination_address),
"data": telegram.payload.value,
+ "destination": str(telegram.destination_address),
+ "direction": telegram.direction.value,
+ "source": str(telegram.source_address),
+ "telegramtype": telegram.telegramtype.value,
},
)
diff --git a/test/devices_tests/binary_sensor_test.py b/test/devices_tests/binary_sensor_test.py
index 4eb719320..33649836f 100644
--- a/test/devices_tests/binary_sensor_test.py
+++ b/test/devices_tests/binary_sensor_test.py
@@ -196,7 +196,9 @@ def test_process_wrong_payload(self):
"""Test process wrong telegram (wrong payload type)."""
xknx = XKNX()
binary_sensor = BinarySensor(xknx, "Warning", group_address_state="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x1, 0x2, 0x3)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x1, 0x2, 0x3))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(binary_sensor.process(telegram))
diff --git a/test/devices_tests/climate_test.py b/test/devices_tests/climate_test.py
index d9fb3e308..7d9587b52 100644
--- a/test/devices_tests/climate_test.py
+++ b/test/devices_tests/climate_test.py
@@ -232,21 +232,24 @@ async def async_after_update_callback(device):
climate.register_device_updated_cb(async_after_update_callback)
telegram = Telegram(
- GroupAddress("1/2/1"), payload=DPTArray(DPTTemperature.to_knx(23))
+ destination_address=GroupAddress("1/2/1"),
+ payload=DPTArray(DPTTemperature.to_knx(23)),
)
self.loop.run_until_complete(climate.process(telegram))
after_update_callback.assert_called_with(climate)
after_update_callback.reset_mock()
telegram = Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPTTemperature.to_knx(23))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPTTemperature.to_knx(23)),
)
self.loop.run_until_complete(climate.process(telegram))
after_update_callback.assert_called_with(climate)
after_update_callback.reset_mock()
telegram = Telegram(
- GroupAddress("1/2/3"), payload=DPTArray(DPTValue1Count.to_knx(-4))
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray(DPTValue1Count.to_knx(-4)),
)
self.loop.run_until_complete(climate.process(telegram))
after_update_callback.assert_called_with(climate)
@@ -272,7 +275,9 @@ async def async_after_update_callback(device):
# Note: the climate object processes the telegram, but the cb
# is called with the climate_mode object.
- telegram = Telegram(GroupAddress("1/2/4"), payload=DPTArray(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/4"), payload=DPTArray(1)
+ )
self.loop.run_until_complete(climate.process(telegram))
after_update_callback.assert_called_with(climate_mode)
after_update_callback.reset_mock()
@@ -297,7 +302,7 @@ def test_set_operation_mode(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/4"),
+ destination_address=GroupAddress("1/2/4"),
payload=DPTArray(DPTHVACMode.to_knx(operation_mode)),
),
)
@@ -319,7 +324,7 @@ def test_set_controller_operation_mode(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/4"),
+ destination_address=GroupAddress("1/2/4"),
payload=DPTArray(DPTHVACContrMode.to_knx(controller_mode)),
),
)
@@ -361,7 +366,7 @@ def test_set_operation_mode_with_controller_status(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/4"),
+ destination_address=GroupAddress("1/2/4"),
payload=DPTArray(DPTControllerStatus.to_knx(operation_mode)),
),
)
@@ -388,10 +393,16 @@ def test_set_operation_mode_with_separate_addresses(self):
telegrams.append(xknx.telegrams.get_nowait())
test_telegrams = [
- Telegram(GroupAddress("1/2/4"), payload=DPTArray(1)),
- Telegram(GroupAddress("1/2/5"), payload=DPTBinary(False)),
- Telegram(GroupAddress("1/2/6"), payload=DPTBinary(False)),
- Telegram(GroupAddress("1/2/7"), payload=DPTBinary(True)),
+ Telegram(destination_address=GroupAddress("1/2/4"), payload=DPTArray(1)),
+ Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTBinary(False)
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/6"), payload=DPTBinary(False)
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/7"), payload=DPTBinary(True)
+ ),
]
self.assertSetEqual(set(telegrams), set(test_telegrams))
@@ -412,7 +423,10 @@ def test_set_heat_cool_binary(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/14"), payload=DPTBinary(True))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/14"), payload=DPTBinary(True)
+ ),
)
self.loop.run_until_complete(
climate_mode.set_controller_mode(HVACControllerMode.COOL)
@@ -420,7 +434,10 @@ def test_set_heat_cool_binary(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/14"), payload=DPTBinary(False))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/14"), payload=DPTBinary(False)
+ ),
)
#
@@ -516,7 +533,7 @@ def test_target_temperature_up(self):
self.assertEqual(
_telegram,
# DEFAULT_TEMPERATURE_STEP is 0.1 -> payload = setpoint_shift * 10
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(30)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(30)),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -526,7 +543,8 @@ def test_target_temperature_up(self):
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(23.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(23.00)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -538,14 +556,15 @@ def test_target_temperature_up(self):
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(40)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(40)),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(24.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(24.00)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -557,14 +576,15 @@ def test_target_temperature_up(self):
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(35)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(35)),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(23.50))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(23.50)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -599,7 +619,7 @@ def test_target_temperature_down(self):
self.assertEqual(
_telegram,
# DEFAULT_TEMPERATURE_STEP is 0.1 -> payload = setpoint_shift * 10
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(10)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(10)),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -609,7 +629,8 @@ def test_target_temperature_down(self):
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(23.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(23.00)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -621,14 +642,15 @@ def test_target_temperature_down(self):
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(0xF1)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(0xF1)),
) # -15
self.loop.run_until_complete(xknx.devices.process(_telegram))
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(20.50))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(20.50)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -640,14 +662,15 @@ def test_target_temperature_down(self):
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(0xE2)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(0xE2)),
) # -30
self.loop.run_until_complete(xknx.devices.process(_telegram))
_telegram = xknx.telegrams.get_nowait()
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(19.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(19.00)),
),
)
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -686,7 +709,7 @@ def test_target_temperature_modified_step(self):
self.assertEqual(
_telegram,
# temperature_step is 0.5 -> payload = setpoint_shift * 2
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(6)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(6)),
)
self.loop.run_until_complete(climate.target_temperature.set(23.00))
@@ -696,7 +719,8 @@ def test_target_temperature_modified_step(self):
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(23.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(23.00)),
),
)
self.assertEqual(climate.base_temperature, 20.00)
@@ -706,14 +730,15 @@ def test_target_temperature_modified_step(self):
self.loop.run_until_complete(xknx.devices.process(_telegram))
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(8)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(8)),
)
_telegram = xknx.telegrams.get_nowait()
self.loop.run_until_complete(xknx.devices.process(_telegram))
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(24.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(24.00)),
),
)
self.assertEqual(climate.target_temperature.value, 24.00)
@@ -745,7 +770,8 @@ def test_base_temperature(self):
self.assertEqual(
_telegram,
Telegram(
- GroupAddress("1/2/2"), payload=DPTArray(DPT2ByteFloat().to_knx(21.00))
+ destination_address=GroupAddress("1/2/2"),
+ payload=DPTArray(DPT2ByteFloat().to_knx(21.00)),
),
)
self.assertFalse(climate.initialized_for_setpoint_shift_calculations)
@@ -759,7 +785,7 @@ def test_base_temperature(self):
self.assertEqual(
_telegram,
# DEFAULT_TEMPERATURE_STEP is 0.1 -> payload = setpoint_shift * 10
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(10)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(10)),
)
self.assertTrue(climate.initialized_for_setpoint_shift_calculations)
self.assertEqual(climate.base_temperature, 20.00)
@@ -773,7 +799,7 @@ def test_base_temperature(self):
self.assertEqual(
_telegram,
# DEFAULT_TEMPERATURE_STEP is 0.1 -> payload = setpoint_shift * 10
- Telegram(GroupAddress("1/2/3"), payload=DPTArray(20)),
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(20)),
)
_telegram = xknx.telegrams.get_nowait()
self.loop.run_until_complete(xknx.devices.process(_telegram))
@@ -799,7 +825,7 @@ def test_target_temperature_step_mode_9002(self):
self.loop.run_until_complete(
climate.target_temperature.process(
Telegram(
- GroupAddress("1/2/2"),
+ destination_address=GroupAddress("1/2/2"),
payload=DPTArray(DPT2ByteFloat().to_knx(20.00)),
)
)
@@ -813,7 +839,10 @@ def test_target_temperature_step_mode_9002(self):
self.assertEqual(climate.base_temperature, 20.00)
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x00, 0x00))),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x00, 0x00)),
+ ),
) # 0
# - 0.6 °C = 19.4
self.loop.run_until_complete(climate.set_target_temperature(19.40))
@@ -822,13 +851,16 @@ def test_target_temperature_step_mode_9002(self):
self.loop.run_until_complete(xknx.devices.process(_telegram))
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x87, 0xC4))),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x87, 0xC4)),
+ ),
) # -0.6
# simulate incoming new target temperature for next calculation
self.loop.run_until_complete(
climate.target_temperature.process(
Telegram(
- GroupAddress("1/2/2"),
+ destination_address=GroupAddress("1/2/2"),
payload=DPTArray(DPT2ByteFloat().to_knx(19.40)),
)
)
@@ -840,7 +872,10 @@ def test_target_temperature_step_mode_9002(self):
self.loop.run_until_complete(xknx.devices.process(_telegram))
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x01, 0x5E))),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x01, 0x5E)),
+ ),
) # +3.5
#
@@ -892,7 +927,11 @@ def test_sync(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_operation_mode(self):
@@ -908,7 +947,11 @@ def test_sync_operation_mode(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_controller_status(self):
@@ -924,7 +967,11 @@ def test_sync_controller_status(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/24"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/24"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_controller_mode(self):
@@ -940,7 +987,11 @@ def test_sync_controller_mode(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/14"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/14"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_operation_mode_state(self):
@@ -965,9 +1016,18 @@ def test_sync_operation_mode_state(self):
self.assertSetEqual(
set(telegrams),
{
- Telegram(GroupAddress("1/2/5"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/6"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/14"), TelegramType.GROUP_READ),
+ Telegram(
+ destination_address=GroupAddress("1/2/5"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/6"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/14"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
},
)
@@ -984,7 +1044,11 @@ def test_sync_heat_cool(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/15"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/15"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -995,7 +1059,7 @@ def test_process_temperature(self):
xknx = XKNX()
climate = Climate(xknx, "TestClimate", group_address_temperature="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.payload = DPTArray(DPTTemperature().to_knx(21.34))
self.loop.run_until_complete(climate.process(telegram))
self.assertEqual(climate.temperature.value, 21.34)
@@ -1010,14 +1074,14 @@ def test_process_operation_mode(self):
group_address_controller_status="1/2/3",
)
for operation_mode in DPT_20102_MODES:
- telegram = Telegram(GroupAddress("1/2/5"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/5"))
telegram.payload = DPTArray(DPTHVACMode.to_knx(operation_mode))
self.loop.run_until_complete(climate_mode.process(telegram))
self.assertEqual(climate_mode.operation_mode, operation_mode)
for operation_mode in DPT_20102_MODES:
if operation_mode == HVACOperationMode.AUTO:
continue
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.payload = DPTArray(DPTControllerStatus.to_knx(operation_mode))
self.loop.run_until_complete(climate_mode.process(telegram))
self.assertEqual(climate_mode.operation_mode, operation_mode)
@@ -1029,7 +1093,7 @@ def test_process_controller_mode(self):
xknx, "TestClimate", group_address_controller_mode="1/2/5"
)
for _, controller_mode in DPTHVACContrMode.SUPPORTED_MODES.items():
- telegram = Telegram(GroupAddress("1/2/5"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/5"))
telegram.payload = DPTArray(DPTHVACContrMode.to_knx(controller_mode))
self.loop.run_until_complete(climate_mode.process(telegram))
self.assertEqual(climate_mode.controller_mode, controller_mode)
@@ -1043,7 +1107,9 @@ def test_process_controller_status_wrong_payload(self):
group_address_operation_mode="1/2/5",
group_address_controller_status="1/2/3",
)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(climate_mode.process(telegram))
@@ -1056,7 +1122,9 @@ def test_process_controller_status_payload_invalid_length(self):
group_address_operation_mode="1/2/5",
group_address_controller_status="1/2/3",
)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(climate_mode.process(telegram))
@@ -1069,7 +1137,9 @@ def test_process_operation_mode_wrong_payload(self):
group_address_operation_mode="1/2/5",
group_address_controller_status="1/2/3",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(climate_mode.process(telegram))
@@ -1082,7 +1152,9 @@ def test_process_operation_mode_payload_invalid_length(self):
group_address_operation_mode="1/2/5",
group_address_controller_status="1/2/3",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(climate_mode.process(telegram))
@@ -1100,7 +1172,7 @@ async def async_after_update_callback(device):
climate.register_device_updated_cb(async_after_update_callback)
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.payload = DPTArray(DPTTemperature().to_knx(21.34))
self.loop.run_until_complete(climate.process(telegram))
after_update_callback.assert_called_with(climate)
@@ -1115,12 +1187,12 @@ def test_process_heat_cool(self):
group_address_heat_cool_state="1/2/15",
)
- telegram = Telegram(GroupAddress("1/2/14"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/14"))
telegram.payload = DPTBinary(False)
self.loop.run_until_complete(climate_mode.process(telegram))
self.assertEqual(climate_mode.controller_mode, HVACControllerMode.COOL)
- telegram = Telegram(GroupAddress("1/2/14"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/14"))
telegram.payload = DPTBinary(True)
self.loop.run_until_complete(climate_mode.process(telegram))
self.assertEqual(climate_mode.controller_mode, HVACControllerMode.HEAT)
@@ -1277,7 +1349,7 @@ def test_process_power_status(self):
"""Test process / reading telegrams from telegram queue. Test if DPT20.105 controller mode is set correctly."""
xknx = XKNX()
climate = Climate(xknx, "TestClimate", group_address_on_off="1/2/2")
- telegram = Telegram(GroupAddress("1/2/2"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/2"))
telegram.payload = DPTBinary(1)
self.loop.run_until_complete(climate.process(telegram))
self.assertEqual(climate.is_on, True)
@@ -1285,7 +1357,7 @@ def test_process_power_status(self):
climate_inv = Climate(
xknx, "TestClimate", group_address_on_off="1/2/2", on_off_invert=True
)
- telegram = Telegram(GroupAddress("1/2/2"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/2"))
telegram.payload = DPTBinary(1)
self.loop.run_until_complete(climate_inv.process(telegram))
self.assertEqual(climate_inv.is_on, False)
@@ -1300,7 +1372,9 @@ def test_power_on_off(self):
self.assertEqual(climate.is_on, True)
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/2"), payload=DPTBinary(True)),
+ Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(True)
+ ),
)
self.loop.run_until_complete(climate.turn_off())
_telegram = xknx.telegrams.get_nowait()
@@ -1308,7 +1382,9 @@ def test_power_on_off(self):
self.assertEqual(climate.is_on, False)
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/2"), payload=DPTBinary(False)),
+ Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(False)
+ ),
)
climate_inv = Climate(
@@ -1320,7 +1396,9 @@ def test_power_on_off(self):
self.assertEqual(climate_inv.is_on, True)
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/2"), payload=DPTBinary(False)),
+ Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(False)
+ ),
)
self.loop.run_until_complete(climate_inv.turn_off())
_telegram = xknx.telegrams.get_nowait()
@@ -1328,5 +1406,7 @@ def test_power_on_off(self):
self.assertEqual(climate_inv.is_on, False)
self.assertEqual(
_telegram,
- Telegram(GroupAddress("1/2/2"), payload=DPTBinary(True)),
+ Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(True)
+ ),
)
diff --git a/test/devices_tests/cover_test.py b/test/devices_tests/cover_test.py
index ddef1cd78..c70353379 100644
--- a/test/devices_tests/cover_test.py
+++ b/test/devices_tests/cover_test.py
@@ -127,7 +127,11 @@ def test_sync(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_state(self):
@@ -145,7 +149,11 @@ def test_sync_state(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_angle(self):
@@ -163,11 +171,19 @@ def test_sync_angle(self):
self.assertEqual(xknx.telegrams.qsize(), 2)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
telegram2 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram2, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram2,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_angle_state(self):
@@ -185,7 +201,11 @@ def test_sync_angle_state(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -207,7 +227,8 @@ def test_set_up(self):
telegram = xknx.telegrams.get_nowait()
# DPT 1.008 - 0:up 1:down
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(0)),
)
#
@@ -228,7 +249,8 @@ def test_set_short_down(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(1)),
)
#
@@ -250,7 +272,8 @@ def test_set_down_inverted(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(0)),
)
#
@@ -272,7 +295,8 @@ def test_set_short_up(self):
telegram = xknx.telegrams.get_nowait()
# DPT 1.008 - 0:up 1:down
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/2"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/2"), payload=DPTBinary(0)),
)
#
@@ -295,7 +319,8 @@ def test_set_up_inverted(self):
telegram = xknx.telegrams.get_nowait()
# DPT 1.008 - 0:up 1:down
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/2"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/2"), payload=DPTBinary(1)),
)
#
@@ -317,7 +342,8 @@ def test_set_down(self):
telegram = xknx.telegrams.get_nowait()
# DPT 1.008 - 0:up 1:down
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/2"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/2"), payload=DPTBinary(1)),
)
#
@@ -338,7 +364,8 @@ def test_stop(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/2"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/2"), payload=DPTBinary(1)),
)
cover_manual_stop = Cover(
@@ -354,7 +381,8 @@ def test_stop(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/0"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/0"), payload=DPTBinary(1)),
)
#
@@ -375,7 +403,8 @@ def test_position(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray(0x80))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(0x80)),
)
def test_position_without_position_address_up(self):
@@ -394,7 +423,8 @@ def test_position_without_position_address_up(self):
telegram = xknx.telegrams.get_nowait()
# DPT 1.008 - 0:up 1:down
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(0)),
)
self.assertEqual(cover.travelcalculator.travel_to_position, 50)
self.assertTrue(cover.is_opening())
@@ -418,7 +448,8 @@ def test_position_without_position_address_down(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(1)),
)
self.assertEqual(cover.travelcalculator.travel_to_position, 80)
self.assertTrue(cover.is_closing())
@@ -448,7 +479,8 @@ def test_position_without_position_address_uninitialized_up(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(0)),
)
def test_position_without_position_address_uninitialized_down(self):
@@ -473,7 +505,8 @@ def test_position_without_position_address_uninitialized_down(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTBinary(1)),
)
def test_angle(self):
@@ -493,7 +526,10 @@ def test_angle(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/4/18"), payload=DPTArray(0x80))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/4/18"), payload=DPTArray(0x80)
+ ),
)
def test_angle_not_supported(self):
@@ -527,24 +563,32 @@ def test_process_position(self):
group_address_position_state="1/2/4",
)
# initial position process - position is unknown so this is the new state - not moving
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray(213))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray(213)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertEqual(cover.current_position(), 84)
self.assertFalse(cover.is_traveling())
# state telegram updates current position - we are not moving so this is new state - not moving
- telegram = Telegram(GroupAddress("1/2/4"), payload=DPTArray(42))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/4"), payload=DPTArray(42)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertEqual(cover.current_position(), 16)
self.assertFalse(cover.is_traveling())
self.assertEqual(cover.travelcalculator.travel_to_position, 16)
# new position - movement starts
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray(255))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray(255)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertEqual(cover.current_position(), 16)
self.assertTrue(cover.is_closing())
self.assertEqual(cover.travelcalculator.travel_to_position, 100)
# new state while moving - movement goes on; travelcalculator updated
- telegram = Telegram(GroupAddress("1/2/4"), payload=DPTArray(213))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/4"), payload=DPTArray(213)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertEqual(cover.current_position(), 84)
self.assertTrue(cover.is_closing())
@@ -561,7 +605,9 @@ def test_process_angle(self):
group_address_angle="1/2/3",
group_address_angle_state="1/2/4",
)
- telegram = Telegram(GroupAddress("1/2/4"), payload=DPTArray(42))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/4"), payload=DPTArray(42)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertEqual(cover.current_angle(), 16)
@@ -573,7 +619,9 @@ def test_process_up(self):
)
cover.travelcalculator.set_position(50)
self.assertFalse(cover.is_traveling())
- telegram = Telegram(GroupAddress("1/2/1"), payload=DPTBinary(0))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/1"), payload=DPTBinary(0)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertTrue(cover.is_opening())
@@ -585,7 +633,9 @@ def test_process_down(self):
)
cover.travelcalculator.set_position(50)
self.assertFalse(cover.is_traveling())
- telegram = Telegram(GroupAddress("1/2/1"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/1"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertTrue(cover.is_closing())
@@ -601,7 +651,9 @@ def test_process_stop(self):
cover.travelcalculator.set_position(50)
self.loop.run_until_complete(cover.set_down())
self.assertTrue(cover.is_traveling())
- telegram = Telegram(GroupAddress("1/2/2"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertFalse(cover.is_traveling())
@@ -617,7 +669,9 @@ def test_process_short_stop(self):
cover.travelcalculator.set_position(50)
self.loop.run_until_complete(cover.set_down())
self.assertTrue(cover.is_traveling())
- telegram = Telegram(GroupAddress("1/2/2"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/2"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(cover.process(telegram))
self.assertFalse(cover.is_traveling())
@@ -653,12 +707,16 @@ async def async_after_update_callback(device):
("1/2/7", DPTArray(51), "angle state"),
]:
with self.subTest(address=address, feature=feature):
- telegram = Telegram(GroupAddress(address), payload=payload)
+ telegram = Telegram(
+ destination_address=GroupAddress(address), payload=payload
+ )
self.loop.run_until_complete(cover.process(telegram))
after_update_callback.assert_called_with(cover)
after_update_callback.reset_mock()
# Stop only when cover is travelling
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(cover.process(telegram))
after_update_callback.assert_not_called()
self.loop.run_until_complete(cover.set_down())
diff --git a/test/devices_tests/device_test.py b/test/devices_tests/device_test.py
index 276030488..c925e0b43 100644
--- a/test/devices_tests/device_test.py
+++ b/test/devices_tests/device_test.py
@@ -102,7 +102,7 @@ def test_process(self):
fut.set_result(None)
mock_group_read.return_value = fut
telegram = Telegram(
- GroupAddress("1/2/1"),
+ destination_address=GroupAddress("1/2/1"),
payload=DPTArray((0x01, 0x02)),
telegramtype=TelegramType.GROUP_READ,
)
@@ -114,7 +114,7 @@ def test_process(self):
fut.set_result(None)
mock_group_write.return_value = fut
telegram = Telegram(
- GroupAddress("1/2/1"),
+ destination_address=GroupAddress("1/2/1"),
payload=DPTArray((0x01, 0x02)),
telegramtype=TelegramType.GROUP_WRITE,
)
@@ -126,7 +126,7 @@ def test_process(self):
fut.set_result(None)
mock_group_response.return_value = fut
telegram = Telegram(
- GroupAddress("1/2/1"),
+ destination_address=GroupAddress("1/2/1"),
payload=DPTArray((0x01, 0x02)),
telegramtype=TelegramType.GROUP_RESPONSE,
)
diff --git a/test/devices_tests/expose_sensor_test.py b/test/devices_tests/expose_sensor_test.py
index 2ebe79c6f..c7fbafe17 100644
--- a/test/devices_tests/expose_sensor_test.py
+++ b/test/devices_tests/expose_sensor_test.py
@@ -72,7 +72,9 @@ def test_set_binary(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"), TelegramType.GROUP_WRITE, payload=DPTBinary(0)
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_WRITE,
+ payload=DPTBinary(0),
),
)
@@ -89,8 +91,8 @@ def test_set_percent(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_WRITE,
payload=DPTArray((0xBF,)),
),
)
@@ -107,8 +109,8 @@ def test_set_temperature(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_WRITE,
payload=DPTArray((0x0C, 0x1A)),
),
)
@@ -124,7 +126,7 @@ def test_process_binary(self):
)
expose_sensor.sensor_value.payload = DPTBinary(1)
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.telegramtype = TelegramType.GROUP_READ
self.loop.run_until_complete(expose_sensor.process(telegram))
self.assertEqual(xknx.telegrams.qsize(), 1)
@@ -132,8 +134,8 @@ def test_process_binary(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
- TelegramType.GROUP_RESPONSE,
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_RESPONSE,
payload=DPTBinary(True),
),
)
@@ -146,7 +148,7 @@ def test_process_percent(self):
)
expose_sensor.sensor_value.payload = DPTArray((0x40,))
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.telegramtype = TelegramType.GROUP_READ
self.loop.run_until_complete(expose_sensor.process(telegram))
self.assertEqual(xknx.telegrams.qsize(), 1)
@@ -154,8 +156,8 @@ def test_process_percent(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
- TelegramType.GROUP_RESPONSE,
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_RESPONSE,
payload=DPTArray((0x40,)),
),
)
@@ -168,7 +170,7 @@ def test_process_temperature(self):
)
expose_sensor.sensor_value.payload = DPTArray((0x0C, 0x1A))
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.telegramtype = TelegramType.GROUP_READ
self.loop.run_until_complete(expose_sensor.process(telegram))
self.assertEqual(xknx.telegrams.qsize(), 1)
@@ -176,8 +178,8 @@ def test_process_temperature(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
- TelegramType.GROUP_RESPONSE,
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_RESPONSE,
payload=DPTArray((0x0C, 0x1A)),
),
)
diff --git a/test/devices_tests/fan_test.py b/test/devices_tests/fan_test.py
index 10440d743..f9715e6ba 100644
--- a/test/devices_tests/fan_test.py
+++ b/test/devices_tests/fan_test.py
@@ -37,7 +37,11 @@ def test_sync(self):
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -58,7 +62,11 @@ def test_sync_state_address(self):
telegram1 = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram1, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram1,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -74,7 +82,8 @@ def test_set_speed(self):
telegram = xknx.telegrams.get_nowait()
# 140 is 55% as byte (0...255)
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray(140))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTArray(140)),
)
#
@@ -87,7 +96,9 @@ def test_process_speed(self):
self.assertEqual(fan.current_speed, None)
# 140 is 55% as byte (0...255)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray(140))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray(140)
+ )
self.loop.run_until_complete(fan.process(telegram))
self.assertEqual(fan.current_speed, 55)
@@ -95,7 +106,9 @@ def test_process_speed_wrong_payload(self): # pylint: disable=invalid-name
"""Test process wrong telegrams. (wrong payload type)."""
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(fan.process(telegram))
@@ -104,7 +117,9 @@ def test_process_fan_payload_invalid_length(self):
# pylint: disable=invalid-name
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(fan.process(telegram))
diff --git a/test/devices_tests/light_test.py b/test/devices_tests/light_test.py
index 3666e01e3..0d6e8b104 100644
--- a/test/devices_tests/light_test.py
+++ b/test/devices_tests/light_test.py
@@ -155,12 +155,30 @@ def test_sync(self):
telegrams.append(xknx.telegrams.get_nowait())
test_telegrams = [
- Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/5"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/6"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/7"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/8"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/9"), TelegramType.GROUP_READ),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/5"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/6"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/7"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/8"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/9"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
]
self.assertEqual(len(set(telegrams)), 6)
@@ -197,12 +215,30 @@ def test_sync_state_address(self):
telegrams.append(xknx.telegrams.get_nowait())
test_telegrams = [
- Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/6"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/8"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/10"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/12"), TelegramType.GROUP_READ),
- Telegram(GroupAddress("1/2/14"), TelegramType.GROUP_READ),
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/6"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/8"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/10"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/12"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
+ Telegram(
+ destination_address=GroupAddress("1/2/14"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
]
self.assertEqual(len(set(telegrams)), 6)
@@ -224,7 +260,8 @@ def test_set_on(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
#
@@ -243,7 +280,8 @@ def test_set_off(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
#
@@ -262,7 +300,8 @@ def test_set_brightness(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/5"), payload=DPTArray(23))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/5"), payload=DPTArray(23)),
)
def test_set_brightness_not_dimmable(self):
@@ -293,7 +332,11 @@ def test_set_color(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24, 25)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/5"),
+ payload=DPTArray((23, 24, 25)),
+ ),
)
self.loop.run_until_complete(xknx.devices.process(telegram))
self.assertEqual(light.current_color, ((23, 24, 25), None))
@@ -328,7 +371,10 @@ def test_set_color_rgbw(self):
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
telegram,
- Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24, 25, 26, 0, 15))),
+ Telegram(
+ destination_address=GroupAddress("1/2/5"),
+ payload=DPTArray((23, 24, 25, 26, 0, 15)),
+ ),
)
self.loop.run_until_complete(xknx.devices.process(telegram))
self.assertEqual(light.current_color, ([23, 24, 25], 26))
@@ -367,7 +413,8 @@ def test_set_tw(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/5"), payload=DPTArray(23))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/5"), payload=DPTArray(23)),
)
def test_set_tw_unsupported(self):
@@ -400,7 +447,7 @@ def test_set_color_temp(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/5"),
+ destination_address=GroupAddress("1/2/5"),
payload=DPTArray(
(
0x0F,
@@ -436,11 +483,15 @@ def test_process_switch(self):
)
self.assertEqual(light.state, None)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.state, True)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)
+ )
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.state, False)
@@ -463,7 +514,9 @@ async def async_after_update_callback(device):
light.register_device_updated_cb(async_after_update_callback)
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
self.loop.run_until_complete(light.process(telegram))
after_update_callback.assert_called_with(light)
@@ -479,7 +532,9 @@ def test_process_dimm(self):
)
self.assertEqual(light.current_brightness, None)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray(23))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray(23)
+ )
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.current_brightness, 23)
@@ -492,7 +547,9 @@ def test_process_dimm_wrong_payload(self):
group_address_switch="1/2/3",
group_address_brightness="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
@@ -506,7 +563,9 @@ def test_process_dimm_payload_invalid_length(self):
group_address_switch="1/2/3",
group_address_brightness="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
@@ -520,7 +579,9 @@ def test_process_color(self):
group_address_color="1/2/5",
)
self.assertEqual(light.current_color, (None, None))
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24, 25)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray((23, 24, 25))
+ )
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.current_color, ((23, 24, 25), None))
@@ -536,7 +597,8 @@ def test_process_color_rgbw(self):
)
self.assertEqual(light.current_color, (None, None))
telegram = Telegram(
- GroupAddress("1/2/5"), payload=DPTArray((23, 24, 25, 26, 0, 15))
+ destination_address=GroupAddress("1/2/5"),
+ payload=DPTArray((23, 24, 25, 26, 0, 15)),
)
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.current_color, ([23, 24, 25], 26))
@@ -552,7 +614,9 @@ def test_process_tunable_white(self):
)
self.assertEqual(light.current_tunable_white, None)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray(23))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray(23)
+ )
self.loop.run_until_complete(light.process(telegram))
self.assertEqual(light.current_tunable_white, 23)
@@ -565,7 +629,9 @@ def test_process_tunable_white_wrong_payload(self):
group_address_switch="1/2/3",
group_address_tunable_white="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
@@ -579,7 +645,9 @@ def test_process_tunable_white_payload_invalid_length(self):
group_address_switch="1/2/3",
group_address_tunable_white="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
@@ -595,7 +663,7 @@ def test_process_color_temperature(self):
self.assertEqual(light.current_color_temperature, None)
telegram = Telegram(
- GroupAddress("1/2/5"),
+ destination_address=GroupAddress("1/2/5"),
payload=DPTArray(
(
0x0F,
@@ -615,7 +683,9 @@ def test_process_color_temperature_wrong_payload(self):
group_address_switch="1/2/3",
group_address_color_temperature="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
@@ -629,7 +699,9 @@ def test_process_color_temperature_payload_invalid_length(self):
group_address_switch="1/2/3",
group_address_color_temperature="1/2/5",
)
- telegram = Telegram(GroupAddress("1/2/5"), payload=DPTArray(23))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/5"), payload=DPTArray(23)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(light.process(telegram))
diff --git a/test/devices_tests/notification_test.py b/test/devices_tests/notification_test.py
index 174f12b28..31e5a22c0 100644
--- a/test/devices_tests/notification_test.py
+++ b/test/devices_tests/notification_test.py
@@ -35,7 +35,11 @@ def test_sync_state(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -46,13 +50,15 @@ def test_process(self):
xknx = XKNX()
notification = Notification(xknx, "Warning", group_address="1/2/3")
telegram_set = Telegram(
- GroupAddress("1/2/3"), payload=DPTArray(DPTString().to_knx("Ein Prosit!"))
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray(DPTString().to_knx("Ein Prosit!")),
)
self.loop.run_until_complete(notification.process(telegram_set))
self.assertEqual(notification.message, "Ein Prosit!")
telegram_unset = Telegram(
- GroupAddress("1/2/3"), payload=DPTArray(DPTString().to_knx(""))
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray(DPTString().to_knx("")),
)
self.loop.run_until_complete(notification.process(telegram_unset))
self.assertEqual(notification.message, "")
@@ -70,7 +76,8 @@ async def async_after_update_callback(device):
notification.register_device_updated_cb(async_after_update_callback)
telegram_set = Telegram(
- GroupAddress("1/2/3"), payload=DPTArray(DPTString().to_knx("Ein Prosit!"))
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray(DPTString().to_knx("Ein Prosit!")),
)
self.loop.run_until_complete(notification.process(telegram_set))
after_update_callback.assert_called_with(notification)
@@ -80,7 +87,9 @@ def test_process_payload_invalid_length(self):
# pylint: disable=invalid-name
xknx = XKNX()
notification = Notification(xknx, "Warning", group_address="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray((23, 24)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((23, 24))
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(notification.process(telegram))
@@ -88,7 +97,9 @@ def test_process_wrong_payload(self):
"""Test process wrong telegram (wrong payload type)."""
xknx = XKNX()
notification = Notification(xknx, "Warning", group_address="1/2/3")
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(notification.process(telegram))
@@ -105,7 +116,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray(DPTString().to_knx("Ein Prosit!")),
),
)
@@ -117,7 +128,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray(DPTString().to_knx("This is too lo")),
),
)
diff --git a/test/devices_tests/scene_test.py b/test/devices_tests/scene_test.py
index 2daf43ce3..671b94003 100644
--- a/test/devices_tests/scene_test.py
+++ b/test/devices_tests/scene_test.py
@@ -45,7 +45,8 @@ def test_run(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTArray(0x16))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTArray(0x16)),
)
def test_do(self):
@@ -56,7 +57,8 @@ def test_do(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/1"), payload=DPTArray(0x16))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/1"), payload=DPTArray(0x16)),
)
def test_wrong_do(self):
diff --git a/test/devices_tests/sensor_expose_loop_test.py b/test/devices_tests/sensor_expose_loop_test.py
index 0716b081b..db73175b9 100644
--- a/test/devices_tests/sensor_expose_loop_test.py
+++ b/test/devices_tests/sensor_expose_loop_test.py
@@ -1459,8 +1459,8 @@ def test_array_sensor_loop(self):
)
incoming_telegram = Telegram(
- GroupAddress("1/1/1"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("1/1/1"),
+ telegramtype=TelegramType.GROUP_WRITE,
direction=TelegramDirection.INCOMING,
payload=test_payload,
)
@@ -1481,8 +1481,8 @@ def test_array_sensor_loop(self):
self.assertEqual(
outgoing_telegram,
Telegram(
- GroupAddress("2/2/2"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("2/2/2"),
+ telegramtype=TelegramType.GROUP_WRITE,
direction=TelegramDirection.OUTGOING,
payload=test_payload,
),
@@ -1509,8 +1509,8 @@ def test_binary_sensor_loop(self):
)
incoming_telegram = Telegram(
- GroupAddress("1/1/1"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("1/1/1"),
+ telegramtype=TelegramType.GROUP_WRITE,
direction=TelegramDirection.INCOMING,
payload=test_payload,
)
@@ -1525,8 +1525,8 @@ def test_binary_sensor_loop(self):
self.assertEqual(
outgoing_telegram,
Telegram(
- GroupAddress("2/2/2"),
- TelegramType.GROUP_WRITE,
+ destination_address=GroupAddress("2/2/2"),
+ telegramtype=TelegramType.GROUP_WRITE,
direction=TelegramDirection.OUTGOING,
payload=test_payload,
),
diff --git a/test/devices_tests/sensor_test.py b/test/devices_tests/sensor_test.py
index 5394a1d8c..7157ce386 100644
--- a/test/devices_tests/sensor_test.py
+++ b/test/devices_tests/sensor_test.py
@@ -2640,7 +2640,11 @@ def test_sync(self):
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -2665,7 +2669,7 @@ def test_process(self):
xknx, "TestSensor", value_type="temperature", group_address_state="1/2/3"
)
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.payload = DPTArray((0x06, 0xA0))
self.loop.run_until_complete(sensor.process(telegram))
self.assertEqual(sensor.sensor_value.payload, DPTArray((0x06, 0xA0)))
@@ -2687,7 +2691,7 @@ async def async_after_update_callback(device):
sensor.register_device_updated_cb(async_after_update_callback)
- telegram = Telegram(GroupAddress("1/2/3"))
+ telegram = Telegram(destination_address=GroupAddress("1/2/3"))
telegram.payload = DPTArray((0x01, 0x02))
self.loop.run_until_complete(sensor.process(telegram))
after_update_callback.assert_called_with(sensor)
diff --git a/test/devices_tests/switch_test.py b/test/devices_tests/switch_test.py
index 08525bb7d..2ed54db4d 100644
--- a/test/devices_tests/switch_test.py
+++ b/test/devices_tests/switch_test.py
@@ -42,7 +42,11 @@ def test_sync(self):
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), TelegramType.GROUP_READ)
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
def test_sync_state_address(self):
@@ -57,7 +61,11 @@ def test_sync_state_address(self):
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/4"), TelegramType.GROUP_READ)
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/4"),
+ telegramtype=TelegramType.GROUP_READ,
+ ),
)
#
@@ -247,7 +255,8 @@ def test_set_on(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
#
@@ -261,7 +270,8 @@ def test_set_off(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
#
@@ -276,14 +286,16 @@ def test_set_invert(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
self.loop.run_until_complete(switch.set_off())
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
#
diff --git a/test/io_tests/tunnelling_test.py b/test/io_tests/tunnelling_test.py
index 6f29ad1c1..265e5ef91 100644
--- a/test/io_tests/tunnelling_test.py
+++ b/test/io_tests/tunnelling_test.py
@@ -28,7 +28,9 @@ def test_tunnelling(self):
xknx = XKNX()
communication_channel_id = 23
udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x1, 0x2, 0x3)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x1, 0x2, 0x3))
+ )
sequence_counter = 42
src_address = PhysicalAddress("2.2.2")
tunnelling = Tunnelling(
diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py
index c5f43e5f5..6ee944a7d 100644
--- a/test/knxip_tests/routing_indication_test.py
+++ b/test/knxip_tests/routing_indication_test.py
@@ -105,7 +105,14 @@ def test_EndTOEnd_group_write_binary_on(self):
knxipframe = KNXIPFrame(xknx)
knxipframe.from_knx(raw)
telegram = knxipframe.body.cemi.telegram
- self.assertEqual(telegram, Telegram(GroupAddress("329"), payload=DPTBinary(1)))
+ self.assertEqual(
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("329"),
+ payload=DPTBinary(1),
+ source_address=PhysicalAddress("15.15.249"),
+ ),
+ )
knxipframe2 = KNXIPFrame(xknx)
knxipframe2.init(KNXIPServiceType.ROUTING_INDICATION)
@@ -126,7 +133,14 @@ def test_EndTOEnd_group_write_binary_off(self):
knxipframe = KNXIPFrame(xknx)
knxipframe.from_knx(raw)
telegram = knxipframe.body.cemi.telegram
- self.assertEqual(telegram, Telegram(GroupAddress("329"), payload=DPTBinary(0)))
+ self.assertEqual(
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("329"),
+ payload=DPTBinary(0),
+ source_address=PhysicalAddress("15.15.249"),
+ ),
+ )
knxipframe2 = KNXIPFrame(xknx)
knxipframe2.init(KNXIPServiceType.ROUTING_INDICATION)
@@ -148,7 +162,12 @@ def test_EndTOEnd_group_write_1byte(self):
knxipframe.from_knx(raw)
telegram = knxipframe.body.cemi.telegram
self.assertEqual(
- telegram, Telegram(GroupAddress("331"), payload=DPTArray(0x65))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("331"),
+ payload=DPTArray(0x65),
+ source_address=PhysicalAddress("15.15.249"),
+ ),
)
knxipframe2 = KNXIPFrame(xknx)
@@ -173,7 +192,9 @@ def test_EndTOEnd_group_write_2bytes(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("2049"), payload=DPTArray(DPTTemperature().to_knx(19.85))
+ destination_address=GroupAddress("2049"),
+ payload=DPTArray(DPTTemperature().to_knx(19.85)),
+ source_address=PhysicalAddress("1.4.2"),
),
)
@@ -197,7 +218,12 @@ def test_EndTOEnd_group_read(self):
knxipframe.from_knx(raw)
telegram = knxipframe.body.cemi.telegram
self.assertEqual(
- telegram, Telegram(GroupAddress("440"), TelegramType.GROUP_READ)
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("440"),
+ telegramtype=TelegramType.GROUP_READ,
+ source_address=PhysicalAddress("15.15.249"),
+ ),
)
knxipframe2 = KNXIPFrame(xknx)
@@ -222,7 +248,10 @@ def test_EndTOEnd_group_response(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("392"), TelegramType.GROUP_RESPONSE, payload=DPTBinary(1)
+ destination_address=GroupAddress("392"),
+ telegramtype=TelegramType.GROUP_RESPONSE,
+ payload=DPTBinary(1),
+ source_address=PhysicalAddress("1.3.1"),
),
)
@@ -242,6 +271,7 @@ def test_maximum_apci(self):
telegram = Telegram(
destination_address=GroupAddress(337),
payload=DPTBinary(DPTBinary.APCI_MAX_VALUE),
+ source_address=PhysicalAddress("1.3.1"),
)
xknx = XKNX()
knxipframe = KNXIPFrame(xknx)
diff --git a/test/knxip_tests/tunnelling_request_test.py b/test/knxip_tests/tunnelling_request_test.py
index d334b1cd3..d13eca7ad 100644
--- a/test/knxip_tests/tunnelling_request_test.py
+++ b/test/knxip_tests/tunnelling_request_test.py
@@ -59,13 +59,13 @@ def test_connect_request(self):
self.assertEqual(
knxipframe.body.cemi.telegram,
- Telegram(GroupAddress("9/0/8"), payload=DPTBinary(1)),
+ Telegram(destination_address=GroupAddress("9/0/8"), payload=DPTBinary(1)),
)
knxipframe2 = KNXIPFrame(xknx)
knxipframe2.init(KNXIPServiceType.TUNNELLING_REQUEST)
knxipframe2.body.cemi.telegram = Telegram(
- GroupAddress("9/0/8"), payload=DPTBinary(1)
+ destination_address=GroupAddress("9/0/8"), payload=DPTBinary(1)
)
knxipframe2.body.sequence_counter = 23
knxipframe2.normalize()
diff --git a/test/remote_value_tests/remote_value_1count_test.py b/test/remote_value_tests/remote_value_1count_test.py
index 29a734ba7..bcf5b4112 100644
--- a/test/remote_value_tests/remote_value_1count_test.py
+++ b/test/remote_value_tests/remote_value_1count_test.py
@@ -41,7 +41,10 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x64,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x64,))
+ ),
)
self.loop.run_until_complete(
remote_value.set(
@@ -51,7 +54,10 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x65,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x65,))
+ ),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_climate_mode_test.py b/test/remote_value_tests/remote_value_climate_mode_test.py
index c30c4a80d..2f745e363 100644
--- a/test/remote_value_tests/remote_value_climate_mode_test.py
+++ b/test/remote_value_tests/remote_value_climate_mode_test.py
@@ -176,7 +176,10 @@ def test_set_operation_mode(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x03,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x03,))
+ ),
)
self.loop.run_until_complete(
remote_value.set(HVACOperationMode.FROST_PROTECTION)
@@ -184,7 +187,10 @@ def test_set_operation_mode(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x04,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x04,))
+ ),
)
def test_set_binary(self):
@@ -199,7 +205,10 @@ def test_set_binary(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(True))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(True)
+ ),
)
self.loop.run_until_complete(
remote_value.set(HVACOperationMode.FROST_PROTECTION)
@@ -207,7 +216,10 @@ def test_set_binary(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(False))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(False)
+ ),
)
def test_process_operation_mode(self):
diff --git a/test/remote_value_tests/remote_value_color_rgb_test.py b/test/remote_value_tests/remote_value_color_rgb_test.py
index b75b3c19d..e6c33dd02 100644
--- a/test/remote_value_tests/remote_value_color_rgb_test.py
+++ b/test/remote_value_tests/remote_value_color_rgb_test.py
@@ -64,14 +64,20 @@ def test_set(self):
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x64, 0x65, 0x66))),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x64, 0x65, 0x66)),
+ ),
)
self.loop.run_until_complete(remote_value.set((100, 101, 104)))
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
telegram,
- Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x64, 0x65, 0x68))),
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x64, 0x65, 0x68)),
+ ),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_color_rgbw_test.py b/test/remote_value_tests/remote_value_color_rgbw_test.py
index 79f966c1e..5b929b23b 100644
--- a/test/remote_value_tests/remote_value_color_rgbw_test.py
+++ b/test/remote_value_tests/remote_value_color_rgbw_test.py
@@ -93,7 +93,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray((0x64, 0x65, 0x66, 0x67, 0x00, 0x0F)),
),
)
@@ -103,7 +103,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray((0x64, 0x65, 0x68, 0x69, 0x00, 0x0F)),
),
)
diff --git a/test/remote_value_tests/remote_value_dpt_2_byte_unsigned_test.py b/test/remote_value_tests/remote_value_dpt_2_byte_unsigned_test.py
index 0ac5b7463..e34c8dcc0 100644
--- a/test/remote_value_tests/remote_value_dpt_2_byte_unsigned_test.py
+++ b/test/remote_value_tests/remote_value_dpt_2_byte_unsigned_test.py
@@ -52,7 +52,11 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x0A, 0x0B)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"),
+ payload=DPTArray((0x0A, 0x0B)),
+ ),
)
self.loop.run_until_complete(remote_value.set(5500))
self.assertEqual(xknx.telegrams.qsize(), 1)
@@ -60,7 +64,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray(
(
0x15,
diff --git a/test/remote_value_tests/remote_value_dpt_value_1_ucount_test.py b/test/remote_value_tests/remote_value_dpt_value_1_ucount_test.py
index 09b821517..5f3ae1eb0 100644
--- a/test/remote_value_tests/remote_value_dpt_value_1_ucount_test.py
+++ b/test/remote_value_tests/remote_value_dpt_value_1_ucount_test.py
@@ -52,13 +52,19 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x0A,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x0A,))
+ ),
)
self.loop.run_until_complete(remote_value.set(11))
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x0B,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x0B,))
+ ),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_scene_number_test.py b/test/remote_value_tests/remote_value_scene_number_test.py
index 7bd34074f..a356bd7c0 100644
--- a/test/remote_value_tests/remote_value_scene_number_test.py
+++ b/test/remote_value_tests/remote_value_scene_number_test.py
@@ -50,13 +50,19 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x0A,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x0A,))
+ ),
)
self.loop.run_until_complete(remote_value.set(12))
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTArray((0x0B,)))
+ telegram,
+ Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTArray((0x0B,))
+ ),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_step_test.py b/test/remote_value_tests/remote_value_step_test.py
index 71bf7207b..5a326954e 100644
--- a/test/remote_value_tests/remote_value_step_test.py
+++ b/test/remote_value_tests/remote_value_step_test.py
@@ -80,13 +80,15 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
self.loop.run_until_complete(remote_value.increase())
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_string_test.py b/test/remote_value_tests/remote_value_string_test.py
index 305cec59b..78ac49491 100644
--- a/test/remote_value_tests/remote_value_string_test.py
+++ b/test/remote_value_tests/remote_value_string_test.py
@@ -92,7 +92,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray((97, 115, 100, 102, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)),
),
)
@@ -102,7 +102,7 @@ def test_set(self):
self.assertEqual(
telegram,
Telegram(
- GroupAddress("1/2/3"),
+ destination_address=GroupAddress("1/2/3"),
payload=DPTArray((65, 83, 68, 70, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)),
),
)
diff --git a/test/remote_value_tests/remote_value_switch_test.py b/test/remote_value_tests/remote_value_switch_test.py
index ae16d07b9..325dc0426 100644
--- a/test/remote_value_tests/remote_value_switch_test.py
+++ b/test/remote_value_tests/remote_value_switch_test.py
@@ -64,13 +64,15 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
self.loop.run_until_complete(remote_value.off())
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
def test_process(self):
diff --git a/test/remote_value_tests/remote_value_test.py b/test/remote_value_tests/remote_value_test.py
index cd87890a4..44c528a0c 100644
--- a/test/remote_value_tests/remote_value_test.py
+++ b/test/remote_value_tests/remote_value_test.py
@@ -94,7 +94,10 @@ def test_process_invalid_payload(self):
patch_valid.return_value = False
patch_has_group_address.return_value = True
- telegram = Telegram(GroupAddress("1/2/1"), payload=DPTArray((0x01, 0x02)))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/1"),
+ payload=DPTArray((0x01, 0x02)),
+ )
with self.assertRaises(CouldNotParseTelegram):
self.loop.run_until_complete(remote_value.process(telegram))
@@ -109,7 +112,9 @@ def test_read_state(self):
patch_valid.return_value = True
fut = asyncio.Future()
- telegram = Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram = Telegram(
+ destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)
+ )
fut.set_result(telegram)
patch_read.return_value = fut
@@ -152,7 +157,9 @@ def test_process_listening_address(self):
with patch("xknx.remote_value.RemoteValue.payload_valid") as patch_valid:
patch_valid.return_value = True
test_payload = DPTArray((0x01, 0x02))
- telegram = Telegram(GroupAddress("1/1/1"), payload=test_payload)
+ telegram = Telegram(
+ destination_address=GroupAddress("1/1/1"), payload=test_payload
+ )
self.assertTrue(
self.loop.run_until_complete(
asyncio.Task(remote_value.process(telegram))
diff --git a/test/remote_value_tests/remote_value_updown_test.py b/test/remote_value_tests/remote_value_updown_test.py
index 9898b5c8e..303c9fb45 100644
--- a/test/remote_value_tests/remote_value_updown_test.py
+++ b/test/remote_value_tests/remote_value_updown_test.py
@@ -80,13 +80,15 @@ def test_set(self):
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(1))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(1)),
)
self.loop.run_until_complete(remote_value.up())
self.assertEqual(xknx.telegrams.qsize(), 1)
telegram = xknx.telegrams.get_nowait()
self.assertEqual(
- telegram, Telegram(GroupAddress("1/2/3"), payload=DPTBinary(0))
+ telegram,
+ Telegram(destination_address=GroupAddress("1/2/3"), payload=DPTBinary(0)),
)
def test_process(self):
diff --git a/test/str_test.py b/test/str_test.py
index 539876cc5..43e9f339f 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -473,8 +473,8 @@ def test_telegram(self):
)
self.assertEqual(
str(telegram),
- '" telegramtype="TelegramType.GROUP_WRITE" direction='
- '"TelegramDirection.OUTGOING" />',
+ '" />',
)
def test_dib_generic(self):
diff --git a/xknx/core/value_reader.py b/xknx/core/value_reader.py
index 3ef0f7787..6c3ea6f62 100644
--- a/xknx/core/value_reader.py
+++ b/xknx/core/value_reader.py
@@ -48,7 +48,9 @@ async def read(self):
async def send_group_read(self):
"""Send group read."""
- telegram = Telegram(self.group_address, TelegramType.GROUP_READ)
+ telegram = Telegram(
+ destination_address=self.group_address, telegramtype=TelegramType.GROUP_READ
+ )
await self.xknx.telegrams.put(telegram)
async def telegram_received(self, telegram):
diff --git a/xknx/knxip/cemi_frame.py b/xknx/knxip/cemi_frame.py
index 44b7dedb7..5f4ac8be1 100644
--- a/xknx/knxip/cemi_frame.py
+++ b/xknx/knxip/cemi_frame.py
@@ -75,8 +75,9 @@ def resolve_telegram_type(cmd):
return Telegram(
destination_address=self.dst_addr,
- payload=self.payload,
telegramtype=resolve_telegram_type(self.cmd),
+ payload=self.payload,
+ source_address=self.src_addr,
)
@telegram.setter
diff --git a/xknx/telegram/telegram.py b/xknx/telegram/telegram.py
index 492db070b..b5fe023cc 100644
--- a/xknx/telegram/telegram.py
+++ b/xknx/telegram/telegram.py
@@ -23,16 +23,16 @@
class TelegramDirection(Enum):
"""Enum class for the communication direction of a telegram (from KNX bus or to KNX bus)."""
- INCOMING = 1
- OUTGOING = 2
+ INCOMING = "Incoming"
+ OUTGOING = "Outgoing"
class TelegramType(Enum):
"""Enum class for type of telegram."""
- GROUP_READ = 1
- GROUP_WRITE = 2
- GROUP_RESPONSE = 3
+ GROUP_READ = "GroupValueRead"
+ GROUP_WRITE = "GroupValueWrite"
+ GROUP_RESPONSE = "GroupValueResponse"
class Telegram:
@@ -46,22 +46,25 @@ def __init__(
telegramtype: TelegramType = TelegramType.GROUP_WRITE,
direction: TelegramDirection = TelegramDirection.OUTGOING,
payload: Any = None,
+ source_address: PhysicalAddress = PhysicalAddress(None),
) -> None:
"""Initialize Telegram class."""
- self.direction = direction
- self.telegramtype = telegramtype
self.destination_address = destination_address
+ self.telegramtype = telegramtype
+ self.direction = direction
self.payload = payload
+ self.source_address = source_address
def __str__(self) -> str:
"""Return object as readable string."""
return (
- ''.format(
+ ''.format(
+ self.direction.value,
+ self.telegramtype.value,
+ self.source_address.__repr__(),
self.destination_address.__repr__(),
self.payload,
- self.telegramtype,
- self.direction,
)
)