Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds UPnP (Universal Plug and Play) discovery as a fallback mechanism to mDNS and introduces a new tree command to visualize the topology of Devialet devices, systems, and groups. The implementation merges results from both mDNS and UPnP discovery methods, deduplicating based on (address, port, base_path) tuples.
Changes:
- Added UPnP/SSDP discovery gateway for device discovery via MediaRenderer:2 protocol
- Implemented
treecommand to display hierarchical device topology with optional JSON output - Added logging configuration via
--log-levelCLI argument andDEVIALETCTL_LOG_LEVELenvironment variable
Reviewed changes
Copilot reviewed 8 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/devialetctl/infrastructure/upnp_gateway.py | New UPnP discovery gateway using SSDP protocol for device discovery |
| src/devialetctl/interfaces/cli.py | Added tree command, _discover_targets helper, logging configuration, and topology building logic |
| tests/test_cli_regression.py | Added autouse fixture to disable UPnP discovery during tests and two new tests for tree command |
| src/devialetctl/discovery.py | Updated discover function to merge results from both mDNS and UPnP gateways |
| src/devialetctl/infrastructure/devialet_gateway.py | Added fetch_json_async convenience method |
| src/devialetctl/infrastructure/init.py | Exported new UPnP classes |
| src/devialetctl/domain/events.py | No functional change (formatting only) |
| README.md | Updated features list to mention UPnP fallback |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
src/devialetctl/interfaces/cli.py
Outdated
| system_data = group_data["systems"][system_id] | ||
| devices_items = [ | ||
| { | ||
| "device_id": dev["device_id"], | ||
| "device_name": dev["device_name"], |
There was a problem hiding this comment.
Hardcoding the base_path to "/ipcontrol/v1" discards the actual base_path from the discovered target. If a device uses a different base_path, queries to this gateway will fail. Use the base_path from the target that was used to fetch this device's information, or store base_path in the device dict at lines 73-83 and use it here.
| if base_path != "/ipcontrol/v1": | ||
| LOG.debug( | ||
| "UPnP host rejected (unexpected base_path) host=%s base_path=%s", | ||
| host, | ||
| base_path, | ||
| ) | ||
| continue |
There was a problem hiding this comment.
This validation logic is incorrect and will reject all UPnP devices. The normalize_base_path function always returns "/ipcontrol/v1" when passed "/ipcontrol/v1" (as defined in _DEFAULT_BASE_PATH), so the condition at line 101 will always be False. This means lines 102-107 will never execute, making this check ineffective. The check should either be removed entirely or the logic should be inverted to continue only when base_path is what's expected.
| if base_path != "/ipcontrol/v1": | |
| LOG.debug( | |
| "UPnP host rejected (unexpected base_path) host=%s base_path=%s", | |
| host, | |
| base_path, | |
| ) | |
| continue |
| class UpnpDiscoveryGateway(DiscoveryPort): | ||
| def discover(self, timeout_s: float = 3.0) -> list[Target]: | ||
| uniq: dict[str, UpnpService] = {} | ||
| LOG.debug( | ||
| "UPnP discovery begin timeout_s=%.2f target=%s", | ||
| timeout_s, | ||
| _SSDP_SEARCH_TARGET, | ||
| ) | ||
| for headers in _iter_ssdp_responses(timeout_s): | ||
| location = headers.get("location", "") | ||
| parsed = urlparse(location) | ||
| host = parsed.hostname | ||
| if not host: | ||
| LOG.debug("UPnP response ignored (missing host in location): %s", location) | ||
| continue | ||
| if host in uniq: | ||
| continue | ||
|
|
||
| # Use known Devialet IP-control endpoint defaults for command operations. | ||
| base_path = normalize_base_path(_DEFAULT_BASE_PATH) | ||
| if base_path != "/ipcontrol/v1": | ||
| LOG.debug( | ||
| "UPnP host rejected (unexpected base_path) host=%s base_path=%s", | ||
| host, | ||
| base_path, | ||
| ) | ||
| continue | ||
| uniq[host] = UpnpService( | ||
| name=f"UPnP:{host}", | ||
| address=host, | ||
| port=_DEFAULT_PORT, | ||
| base_path=base_path, | ||
| ) | ||
| LOG.debug("UPnP device accepted host=%s base_path=%s", host, base_path) | ||
|
|
||
| targets = [ | ||
| Target(address=s.address, port=s.port, base_path=s.base_path, name=s.name) | ||
| for s in uniq.values() | ||
| ] | ||
| LOG.debug("UPnP discovery done found=%d", len(targets)) | ||
| return targets |
There was a problem hiding this comment.
The new UPnP discovery gateway lacks unit tests. The repository has comprehensive test coverage for similar components (e.g., test_mdns_gateway_listener.py, test_mdns_gateway_cleanup.py). Consider adding similar unit tests for the UPnP gateway to cover: SSDP header parsing, response iteration with timeouts, socket error handling, deduplication logic, and the discovery method.
src/devialetctl/interfaces/cli.py
Outdated
| gateway = DevialetHttpGateway( | ||
| probe_device["address"], | ||
| probe_device["port"], | ||
| "/ipcontrol/v1", | ||
| ) |
There was a problem hiding this comment.
Hardcoding the base_path to "/ipcontrol/v1" discards the actual base_path from the discovered target. If a device uses a different base_path, queries to this gateway will fail. Use probe_device["base_path"] or target.base_path instead to preserve the discovered configuration.
clementperon
changed the title
1 participant
No description provided.