Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Convert dataclasses to pydantic models #245

Draft
wants to merge 8 commits into
base: projects/zha-web-socket-server
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ dependencies = [
"zha-quirks==0.0.123",
"pyserial==3.5",
"pyserial-asyncio-fast",
"pydantic==2.9.2"
]

[tool.setuptools.packages.find]
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
{"ieee":"00:0d:6f:00:0f:3a:e3:69","nwk":"0x970A","manufacturer":"CentraLite","model":"3320-L","name":"CentraLite 3320-L","quirk_applied":true,"quirk_class":"zhaquirks.centralite.ias.CentraLiteIASSensor","quirk_id":null,"manufacturer_code":49887,"power_source":"Battery or Unknown","lqi":null,"rssi":null,"available":true,"device_type":"EndDevice","signature":{"node_descriptor":{"logical_type":2,"complex_descriptor_available":0,"user_descriptor_available":0,"reserved":0,"aps_flags":0,"frequency_band":8,"mac_capability_flags":128,"manufacturer_code":49887,"maximum_buffer_size":82,"maximum_incoming_transfer_size":82,"server_mask":0,"maximum_outgoing_transfer_size":82,"descriptor_capability_field":0},"endpoints":{"1":{"profile_id":"0x0104","device_type":"0x0402","input_clusters":["0x0000","0x0001","0x0003","0x0020","0x0402","0x0500","0x0b05"],"output_clusters":["0x0019"]},"2":{"profile_id":"0xc2df","device_type":"0x000c","input_clusters":["0x0000","0x0003","0x0b05","0xfc0f"],"output_clusters":["0x0003"]}},"manufacturer":"CentraLite","model":"3320-L"},"active_coordinator":false,"entities":{"binary_sensor,00:0d:6f:00:0f:3a:e3:69-1-1280":{"platform":"binary_sensor","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-1280","class_name":"IASZone","translation_key":null,"device_class":"opening","state_class":null,"entity_category":null,"entity_registry_enabled_default":true,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"IASZoneClusterHandler","generic_id":"cluster_handler_0x0500","endpoint_id":1,"cluster":{"id":1280,"name":"IAS Zone","type":"server","commands":[{"id":0,"name":"enroll_response","schema":{"command":"enroll_response","fields":[{"name":"enroll_response_code","type":"EnrollResponse","optional":false},{"name":"zone_id","type":"uint8_t","optional":false}]},"direction":1,"is_manufacturer_specific":null},{"id":1,"name":"init_normal_op_mode","schema":{"command":"init_normal_op_mode","fields":[]},"direction":0,"is_manufacturer_specific":null},{"id":2,"name":"init_test_mode","schema":{"command":"init_test_mode","fields":[{"name":"test_mode_duration","type":"uint8_t","optional":false},{"name":"current_zone_sensitivity_level","type":"uint8_t","optional":false}]},"direction":0,"is_manufacturer_specific":null}]},"id":"1:0x0500","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0500","status":"initialized","value_attribute":null}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"button,00:0d:6f:00:0f:3a:e3:69-1-3":{"platform":"button","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-3","class_name":"IdentifyButton","translation_key":null,"device_class":"identify","state_class":null,"entity_category":"diagnostic","entity_registry_enabled_default":true,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"IdentifyClusterHandler","generic_id":"cluster_handler_0x0003","endpoint_id":1,"cluster":{"id":3,"name":"Identify","type":"server","commands":[{"id":0,"name":"identify","schema":{"command":"identify","fields":[{"name":"identify_time","type":"uint16_t","optional":false}]},"direction":0,"is_manufacturer_specific":null},{"id":1,"name":"identify_query","schema":{"command":"identify_query","fields":[]},"direction":0,"is_manufacturer_specific":null},{"id":64,"name":"trigger_effect","schema":{"command":"trigger_effect","fields":[{"name":"effect_id","type":"EffectIdentifier","optional":false},{"name":"effect_variant","type":"EffectVariant","optional":false}]},"direction":0,"is_manufacturer_specific":null}]},"id":"1:0x0003","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0003","status":"initialized","value_attribute":null}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"sensor,00:0d:6f:00:0f:3a:e3:69-1-1":{"platform":"sensor","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-1","class_name":"Battery","translation_key":null,"device_class":"battery","state_class":"measurement","entity_category":"diagnostic","entity_registry_enabled_default":true,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"PowerConfigurationClusterHandler","generic_id":"cluster_handler_0x0001","endpoint_id":1,"cluster":{"id":1,"name":"Power Configuration","type":"server","commands":[]},"id":"1:0x0001","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0001","status":"initialized","value_attribute":"battery_voltage"}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"sensor,00:0d:6f:00:0f:3a:e3:69-1-1026":{"platform":"sensor","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-1026","class_name":"Temperature","translation_key":null,"device_class":"temperature","state_class":"measurement","entity_category":null,"entity_registry_enabled_default":true,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"TemperatureMeasurementClusterHandler","generic_id":"cluster_handler_0x0402","endpoint_id":1,"cluster":{"id":1026,"name":"Temperature Measurement","type":"server","commands":[]},"id":"1:0x0402","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0402","status":"initialized","value_attribute":"measured_value"}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"sensor,00:0d:6f:00:0f:3a:e3:69-1-0-rssi":{"platform":"sensor","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-0-rssi","class_name":"RSSISensor","translation_key":"rssi","device_class":"signal_strength","state_class":"measurement","entity_category":"diagnostic","entity_registry_enabled_default":false,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"BasicClusterHandler","generic_id":"cluster_handler_0x0000","endpoint_id":1,"cluster":{"id":0,"name":"Basic","type":"server","commands":[{"id":0,"name":"reset_fact_default","schema":{"command":"reset_fact_default","fields":[]},"direction":0,"is_manufacturer_specific":null}]},"id":"1:0x0000","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0000","status":"initialized","value_attribute":null}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"sensor,00:0d:6f:00:0f:3a:e3:69-1-0-lqi":{"platform":"sensor","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-0-lqi","class_name":"LQISensor","translation_key":"lqi","device_class":null,"state_class":"measurement","entity_category":"diagnostic","entity_registry_enabled_default":false,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"BasicClusterHandler","generic_id":"cluster_handler_0x0000","endpoint_id":1,"cluster":{"id":0,"name":"Basic","type":"server","commands":[{"id":0,"name":"reset_fact_default","schema":{"command":"reset_fact_default","fields":[]},"direction":0,"is_manufacturer_specific":null}]},"id":"1:0x0000","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0000","status":"initialized","value_attribute":null}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null},"update,00:0d:6f:00:0f:3a:e3:69-1-25-firmware_update":{"platform":"update","unique_id":"00:0d:6f:00:0f:3a:e3:69-1-25-firmware_update","class_name":"FirmwareUpdateEntity","translation_key":null,"device_class":"firmware","state_class":null,"entity_category":"config","entity_registry_enabled_default":true,"enabled":true,"fallback_name":null,"cluster_handlers":[{"class_name":"OtaClientClusterHandler","generic_id":"cluster_handler_0x0019","endpoint_id":1,"cluster":{"id":25,"name":"Ota","type":"client","commands":[{"id":3,"name":"image_block","schema":{"command":"image_block","fields":[{"name":"field_control","type":"FieldControl","optional":false},{"name":"manufacturer_code","type":"uint16_t","optional":false},{"name":"image_type","type":"uint16_t","optional":false},{"name":"file_version","type":"uint32_t","optional":false},{"name":"file_offset","type":"uint32_t","optional":false},{"name":"maximum_data_size","type":"uint8_t","optional":false},{"name":"request_node_addr","type":"EUI64","optional":false},{"name":"minimum_block_period","type":"uint16_t","optional":false}]},"direction":0,"is_manufacturer_specific":null},{"id":4,"name":"image_page","schema":{"command":"image_page","fields":[{"name":"field_control","type":"FieldControl","optional":false},{"name":"manufacturer_code","type":"uint16_t","optional":false},{"name":"image_type","type":"uint16_t","optional":false},{"name":"file_version","type":"uint32_t","optional":false},{"name":"file_offset","type":"uint32_t","optional":false},{"name":"maximum_data_size","type":"uint8_t","optional":false},{"name":"page_size","type":"uint16_t","optional":false},{"name":"response_spacing","type":"uint16_t","optional":false},{"name":"request_node_addr","type":"EUI64","optional":false}]},"direction":0,"is_manufacturer_specific":null},{"id":1,"name":"query_next_image","schema":{"command":"query_next_image","fields":[{"name":"field_control","type":"FieldControl","optional":false},{"name":"manufacturer_code","type":"uint16_t","optional":false},{"name":"image_type","type":"uint16_t","optional":false},{"name":"current_file_version","type":"uint32_t","optional":false},{"name":"hardware_version","type":"uint16_t","optional":false}]},"direction":0,"is_manufacturer_specific":null},{"id":8,"name":"query_specific_file","schema":{"command":"query_specific_file","fields":[{"name":"request_node_addr","type":"EUI64","optional":false},{"name":"manufacturer_code","type":"uint16_t","optional":false},{"name":"image_type","type":"uint16_t","optional":false},{"name":"file_version","type":"uint32_t","optional":false},{"name":"current_zigbee_stack_version","type":"uint16_t","optional":false}]},"direction":0,"is_manufacturer_specific":null},{"id":6,"name":"upgrade_end","schema":{"command":"upgrade_end","fields":[{"name":"status","type":"Status","optional":false},{"name":"manufacturer_code","type":"uint16_t","optional":false},{"name":"image_type","type":"uint16_t","optional":false},{"name":"file_version","type":"uint32_t","optional":false}]},"direction":0,"is_manufacturer_specific":null}]},"id":"1:0x0019","unique_id":"00:0d:6f:00:0f:3a:e3:69:1:0x0019","status":"initialized","value_attribute":null}],"device_ieee":"00:0d:6f:00:0f:3a:e3:69","endpoint_id":1,"available":true,"group_id":null}},"neighbors":[],"routes":[],"endpoint_names":[{"name":"IAS_ZONE"},{"name":"unknown 12 device_type of 0xc2df profile id"}],"device_automation_triggers":{"device_offline,device_offline":{"device_event_type":"device_offline"}}}
121 changes: 120 additions & 1 deletion tests/test_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
SIG_EP_TYPE,
create_mock_zigpy_device,
join_zigpy_device,
zigpy_device_from_json,
)
from zha.application import Platform
from zha.application.const import (
Expand All @@ -36,7 +37,13 @@
from zha.application.platforms.sensor import LQISensor, RSSISensor
from zha.application.platforms.switch import Switch
from zha.exceptions import ZHAException
from zha.zigbee.device import ClusterBinding, Device, get_device_automation_triggers
from zha.zigbee.device import (
ClusterBinding,
Device,
NeighborInfo,
RouteInfo,
get_device_automation_triggers,
)
from zha.zigbee.group import Group


Expand Down Expand Up @@ -848,3 +855,115 @@ async def test_device_properties(
assert zha_device.is_router is None
assert zha_device.is_end_device is None
assert zha_device.is_coordinator is None


def test_neighbor_info_ser_deser() -> None:
"""Test the serialization and deserialization of the neighbor info."""

neighbor_info = NeighborInfo(
ieee="00:0d:6f:00:0a:90:69:e7",
nwk="0x1234",
extended_pan_id="00:0d:6f:00:0a:90:69:e7",
lqi=255,
relationship=zdo_t._NeighborEnums.Relationship.Child.name,
depth=0,
device_type=zdo_t._NeighborEnums.DeviceType.Router.name,
rx_on_when_idle=zdo_t._NeighborEnums.RxOnWhenIdle.On.name,
permit_joining=zdo_t._NeighborEnums.PermitJoins.Accepting.name,
)

assert isinstance(neighbor_info.ieee, zigpy.types.EUI64)
assert isinstance(neighbor_info.nwk, zigpy.types.NWK)
assert isinstance(neighbor_info.extended_pan_id, zigpy.types.EUI64)
assert isinstance(neighbor_info.relationship, zdo_t._NeighborEnums.Relationship)
assert isinstance(neighbor_info.device_type, zdo_t._NeighborEnums.DeviceType)
assert isinstance(neighbor_info.rx_on_when_idle, zdo_t._NeighborEnums.RxOnWhenIdle)
assert isinstance(neighbor_info.permit_joining, zdo_t._NeighborEnums.PermitJoins)

assert neighbor_info.model_dump() == {
"ieee": "00:0d:6f:00:0a:90:69:e7",
"nwk": 0x1234,
"extended_pan_id": "00:0d:6f:00:0a:90:69:e7",
"lqi": 255,
"relationship": zdo_t._NeighborEnums.Relationship.Child.name,
"depth": 0,
"device_type": zdo_t._NeighborEnums.DeviceType.Router.name,
"rx_on_when_idle": zdo_t._NeighborEnums.RxOnWhenIdle.On.name,
"permit_joining": zdo_t._NeighborEnums.PermitJoins.Accepting.name,
}

assert neighbor_info.model_dump_json() == (
'{"device_type":"Router","rx_on_when_idle":"On","relationship":"Child",'
'"extended_pan_id":"00:0d:6f:00:0a:90:69:e7","ieee":"00:0d:6f:00:0a:90:69:e7","nwk":"0x1234",'
'"permit_joining":"Accepting","depth":0,"lqi":255}'
)


def test_route_info_ser_deser() -> None:
"""Test the serialization and deserialization of the route info."""

route_info = RouteInfo(
dest_nwk=0x1234,
next_hop=0x5678,
route_status=zdo_t.RouteStatus.Active.name,
memory_constrained=0,
many_to_one=1,
route_record_required=1,
)

assert isinstance(route_info.dest_nwk, zigpy.types.NWK)
assert isinstance(route_info.next_hop, zigpy.types.NWK)
assert isinstance(route_info.route_status, zdo_t.RouteStatus)

assert route_info.model_dump() == {
"dest_nwk": 0x1234,
"next_hop": 0x5678,
"route_status": zdo_t.RouteStatus.Active.name,
"memory_constrained": 0,
"many_to_one": 1,
"route_record_required": 1,
}

assert route_info.model_dump_json() == (
'{"dest_nwk":"0x1234","route_status":"Active","memory_constrained":0,"many_to_one":1,'
'"route_record_required":1,"next_hop":"0x5678"}'
)


def test_convert_extended_pan_id() -> None:
"""Test conversion of extended panid."""

extended_pan_id = zigpy.types.ExtendedPanId.convert("00:0d:6f:00:0a:90:69:e7")

assert NeighborInfo.convert_extended_pan_id(extended_pan_id) == extended_pan_id

converted_extended_pan_id = NeighborInfo.convert_extended_pan_id(
"00:0d:6f:00:0a:90:69:e7"
)
assert isinstance(converted_extended_pan_id, zigpy.types.ExtendedPanId)
assert converted_extended_pan_id == extended_pan_id


async def test_extended_device_info_ser_deser(zha_gateway: Gateway) -> None:
"""Test the serialization and deserialization of the extended device info."""

zigpy_dev = await zigpy_device_from_json(
zha_gateway.application_controller, "tests/data/devices/centralite-3320-l.json"
)
zha_device = await join_zigpy_device(zha_gateway, zigpy_dev)
assert zha_device is not None

assert isinstance(zha_device.extended_device_info.ieee, zigpy.types.EUI64)
assert isinstance(zha_device.extended_device_info.nwk, zigpy.types.NWK)

# last_seen changes so we exclude it from the comparison
json = zha_device.extended_device_info.model_dump_json(exclude=["last_seen"])

# load the json from a file as string
with open(
"tests/data/serialization_data/centralite-3320-l-extended-device-info.json",
encoding="UTF-8",
) as file:
expected_json = file.read()

assert json == expected_json
21 changes: 6 additions & 15 deletions tests/test_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,7 @@
join_zigpy_device,
)
from zha.application import Platform
from zha.application.const import (
CONF_USE_THREAD,
ZHA_GW_MSG,
ZHA_GW_MSG_CONNECTION_LOST,
RadioType,
)
from zha.application.const import CONF_USE_THREAD, ZHA_GW_MSG_CONNECTION_LOST, RadioType
from zha.application.gateway import (
ConnectionLostEvent,
DeviceJoinedDeviceInfo,
Expand Down Expand Up @@ -91,7 +86,7 @@ async def coordinator(zha_gateway: Gateway) -> Device:
}
},
ieee="00:15:8d:00:02:32:4f:32",
nwk=0x0000,
nwk=zigpy.types.NWK(0x0000),
node_descriptor=zdo_t.NodeDescriptor(
logical_type=zdo_t.LogicalType.Coordinator,
complex_descriptor_available=0,
Expand Down Expand Up @@ -535,7 +530,7 @@ async def test_startup_concurrency_limit(
}
},
ieee=f"11:22:33:44:{i:08x}",
nwk=0x1234 + i,
nwk=zigpy.types.NWK(0x1234 + i),
)
zigpy_dev.node_desc.mac_capability_flags |= (
zigpy.zdo.types.NodeDescriptor.MACCapabilityFlags.MainsPowered
Expand Down Expand Up @@ -645,7 +640,7 @@ def test_gateway_raw_device_initialized(
RawDeviceInitializedEvent(
device_info=RawDeviceInitializedDeviceInfo(
ieee=zigpy.types.EUI64.convert("00:0d:6f:00:0a:90:69:e7"),
nwk=0xB79C,
nwk=zigpy.types.NWK(0xB79C),
pairing_status=DevicePairingStatus.INTERVIEW_COMPLETE,
model="FakeModel",
manufacturer="FakeManufacturer",
Expand Down Expand Up @@ -676,9 +671,7 @@ def test_gateway_raw_device_initialized(
}
},
},
),
event_type="zha_gateway_message",
event="raw_device_initialized",
)
),
)

Expand All @@ -698,7 +691,7 @@ def test_gateway_device_joined(
DeviceJoinedEvent(
device_info=DeviceJoinedDeviceInfo(
ieee=zigpy.types.EUI64.convert("00:0d:6f:00:0a:90:69:e7"),
nwk=0xB79C,
nwk=zigpy.types.NWK(0xB79C),
pairing_status=DevicePairingStatus.PAIRED,
)
),
Expand All @@ -717,8 +710,6 @@ def test_gateway_connection_lost(zha_gateway: Gateway) -> None:
ZHA_GW_MSG_CONNECTION_LOST,
ConnectionLostEvent(
exception=exception,
event=ZHA_GW_MSG_CONNECTION_LOST,
event_type=ZHA_GW_MSG,
),
)

Expand Down
Loading