Skip to content
Merged
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 .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -17,3 +17,4 @@ fixtures/*
*.sedbck
tmp
.cache
appdata_folder
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
- PR [#261](https://github.com/plugwise/python-plugwise-usb/pull/261): Sense: bugfix parsing of humidity value as an unsigned int
- PR #263 Maintenance chores and re-instatement of ruff, deprecate pre-commit cloud runs (just leveraging renovate)
- PR #264 Maintenance chores Rework Github Actions workflow
- PR [#269](https://github.com/plugwise/python-plugwise-usb/pull/269) Implementation of ruff fixes
- PR [#271](https://github.com/plugwise/python-plugwise-usb/pull/271) Ruff fix: split update_missing_registrations in quick and full version

## v0.44.3 - 2025-06-12

Expand Down
69 changes: 41 additions & 28 deletions plugwise_usb/network/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@
if self._cache_enabled:
await self.restore_network_cache()
await self.load_registry_from_cache()
await self.update_missing_registrations(quick=True)
await self.update_missing_registrations_quick()

async def restore_network_cache(self) -> None:
"""Restore previously saved cached network and node information."""
Expand Down Expand Up @@ -179,10 +179,10 @@
if self._network_cache is not None:
self._network_cache.update_registration(address, mac, node_type)

async def update_missing_registrations(self, quick: bool = False) -> None: # noqa: PLR0912
"""Retrieve all unknown network registrations from network controller."""
async def update_missing_registrations_full(self) -> None:
"""Full retrieval of all unknown network registrations from network controller."""
for address in range(0, 64):
if self._registry.get(address) is not None and not quick:
if self._registry.get(address) is not None:
mac, _ = self._registry[address]
if mac == "":
self._first_free_address = min(self._first_free_address, address)
Expand All @@ -194,37 +194,50 @@
self._first_free_address = min(
self._first_free_address, nextaddress
)
if quick:
break
_LOGGER.debug(
"Network registration at address %s is %s",
str(nextaddress),
"'empty'" if mac == "" else f"set to {mac}",
)
self.update_network_registration(nextaddress, mac, None)
await sleep(0.1)
if not quick:
await sleep(10)
if quick:
if self._registration_task is None or self._registration_task.done():
self._registration_task = create_task(
self.update_missing_registrations(quick=False)
await sleep(10)
_LOGGER.debug("Full network registration finished")
self._scan_completed = True
if self._cache_enabled:
_LOGGER.debug("Full network registration finished, save to cache")
await self.save_registry_to_cache()
_LOGGER.debug("Full network registration finished, post")
_LOGGER.info("Full network discovery completed")
if self._full_scan_finished is not None:
await self._full_scan_finished()
self._full_scan_finished = None

Check warning on line 213 in plugwise_usb/network/registry.py

View check run for this annotation

Codecov / codecov/patch

plugwise_usb/network/registry.py#L204-L213

Added lines #L204 - L213 were not covered by tests

async def update_missing_registrations_quick(self) -> None:
"""Quick retrieval of all unknown network registrations from network controller."""
for address in range(0, 64):
registration = await self.retrieve_network_registration(address, False)
if registration is not None:
nextaddress, mac = registration
if mac == "":
self._first_free_address = min(
self._first_free_address, nextaddress
)
break
_LOGGER.debug(
"Network registration at address %s is %s",
str(nextaddress),
"'empty'" if mac == "" else f"set to {mac}",
)
if self._quick_scan_finished is not None:
await self._quick_scan_finished()
self._quick_scan_finished = None
_LOGGER.info("Quick network registration discovery finished")
else:
_LOGGER.debug("Full network registration finished")
self._scan_completed = True
if self._cache_enabled:
_LOGGER.debug("Full network registration finished, save to cache")
await self.save_registry_to_cache()
_LOGGER.debug("Full network registration finished, post")
_LOGGER.info("Full network discovery completed")
if self._full_scan_finished is not None:
await self._full_scan_finished()
self._full_scan_finished = None
self.update_network_registration(nextaddress, mac, None)
await sleep(0.1)
if self._registration_task is None or self._registration_task.done():
self._registration_task = create_task(
self.update_missing_registrations_full()
)
if self._quick_scan_finished is not None:
await self._quick_scan_finished()
self._quick_scan_finished = None
_LOGGER.info("Quick network registration discovery finished")

def update_node_registration(self, mac: str) -> int:
"""Register (re)joined node to Plugwise network and return network address."""
Expand Down
Loading