Skip to content
Open
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 packages/api/src/api/controllers/devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ async def scan_devices(
"device_name": device.device_name,
"firmware_version": device.firmware_version,
"available_firmware_version": device.available_firmware_version,
"available_firmware_channel": device.available_firmware_channel,
"response_time": device.response_time,
"error_message": device.error_message,
"last_seen": device.last_seen.isoformat() if device.last_seen else None,
Expand Down
2 changes: 2 additions & 0 deletions packages/api/tests/unit/controllers/test_devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ async def execute(self, scan_request):
device_name="Test Device",
firmware_version="1.0.0",
available_firmware_version="1.2.0",
available_firmware_channel="stable",
response_time=0.5,
last_seen=datetime.now(),
)
Expand All @@ -64,6 +65,7 @@ async def execute(self, scan_request):
assert data[0]["ip"] == "192.168.1.100"
assert data[0]["status"] == "detected"
assert data[0]["available_firmware_version"] == "1.2.0"
assert data[0]["available_firmware_channel"] == "stable"
assert data[0]["device_type"] == "SHSW-PM"
assert data[0]["model_name"] == "Shelly 1PM"

Expand Down
8 changes: 8 additions & 0 deletions packages/cli/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ shelly-manager scan --target 192.168.1.0/24 --export csv --export-file devices.c
- `--use-mdns`: Use mDNS service discovery
- `--timeout`: Timeout per device (default: 3.0s)
- `--workers`: Concurrent workers (default: 50)
- `--include-beta`: Count beta-only firmware updates as available (hidden by default)
- `--export`: Export format (json, csv)
- `--export-file`: Output file path

Expand All @@ -94,12 +95,19 @@ shelly-manager scan --target 192.168.1.0/24 --export csv --export-file devices.c
# Check device status
shelly-manager device status 192.168.1.100
shelly-manager device status 192.168.1.100 192.168.1.101
shelly-manager device status 192.168.1.100 --include-beta # Also surface beta-only updates

# List known devices in a table
shelly-manager device list 192.168.1.0/24
shelly-manager device list 192.168.1.0/24 --include-beta

# Reboot devices
shelly-manager device reboot 192.168.1.100
shelly-manager device reboot 192.168.1.100 --force # Skip confirmation
```

By default, `scan`, `device list`, and `device status` report a device as up to date when the only firmware update available is on the beta channel — pass `--include-beta` to see it. A beta release is always installable explicitly regardless of this flag, via `device update --channel beta`.

**Device Status Output:**

```
Expand Down
33 changes: 28 additions & 5 deletions packages/cli/src/cli/commands/device_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,11 @@ def device_commands() -> None:
@device_commands.command()
@click.argument("targets", nargs=-1)
@click.option("--use-mdns", is_flag=True, help="Use mDNS to discover devices")
@click.option(
"--include-beta",
is_flag=True,
help="Count beta-only firmware updates as available (hidden by default)",
)
@device_targeting_options
@common_options
@click.pass_context
Expand All @@ -44,6 +49,7 @@ async def scan(
targets: tuple[str, ...],
targets_opt: tuple[str, ...],
use_mdns: bool,
include_beta: bool,
timeout: int,
workers: int,
) -> None:
Expand All @@ -55,6 +61,7 @@ async def scan(
shelly-manager scan 192.168.1.0/24
shelly-manager scan -t 192.168.1.100 -t 192.168.1.101
shelly-manager scan --use-mdns
shelly-manager scan 192.168.1.0/24 --include-beta
"""
console = ctx.obj.console
container = ctx.obj.container
Expand All @@ -70,11 +77,16 @@ async def scan(
)

devices_found = await scan_use_case.execute(request)
scan_use_case.display_results(devices_found)
scan_use_case.display_results(devices_found, include_beta=include_beta)


@device_commands.command("list")
@click.argument("targets", nargs=-1)
@click.option(
"--include-beta",
is_flag=True,
help="Count beta-only firmware updates as available (hidden by default)",
)
@device_targeting_options
@common_options
@click.pass_context
Expand All @@ -83,15 +95,17 @@ async def list_devices(
ctx: click.Context,
targets: tuple[str, ...],
targets_opt: tuple[str, ...],
include_beta: bool,
timeout: int,
workers: int,
) -> None:
"""
Similar to scan but optimized for listing known devices with full details in a table format.

Examples:
shelly-manager list 192.168.1.0/24
shelly-manager list -t 192.168.1.100 -t 192.168.1.101
shelly-manager device list 192.168.1.0/24
shelly-manager device list -t 192.168.1.100 -t 192.168.1.101
shelly-manager device list 192.168.1.0/24 --include-beta
"""
console = ctx.obj.console
container = ctx.obj.container
Expand All @@ -108,13 +122,20 @@ async def list_devices(
devices_found = await scan_use_case.execute(request)

if devices_found:
scan_use_case.display_results(devices_found, show_table=True)
scan_use_case.display_results(
devices_found, show_table=True, include_beta=include_beta
)
else:
console.print(f"\n{Messages.warning('No devices found')}")


@device_commands.command()
@click.argument("targets", nargs=-1, required=False)
@click.option(
"--include-beta",
is_flag=True,
help="Include beta-only firmware updates in the update list (hidden by default)",
)
@device_targeting_options
@common_options
@click.pass_context
Expand All @@ -123,6 +144,7 @@ async def status(
ctx: click.Context,
targets: tuple[str, ...],
targets_opt: tuple[str, ...],
include_beta: bool,
timeout: int,
workers: int,
) -> None:
Expand All @@ -133,6 +155,7 @@ async def status(
Examples:
shelly-manager status 192.168.1.100 192.168.1.101
shelly-manager status -t 192.168.1.0/24
shelly-manager status 192.168.1.100 --include-beta
"""
console = ctx.obj.console
container = ctx.obj.container
Expand All @@ -156,7 +179,7 @@ async def status(
"shelly-manager device status 192.168.1.0/24",
)
sys.exit(EXIT_VALIDATION)
status_use_case.display_results(results)
status_use_case.display_results(results, include_beta=include_beta)


@click.group()
Expand Down
58 changes: 48 additions & 10 deletions packages/cli/src/cli/use_cases/common/result_formatting.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from typing import Any

from core.domain.entities import DeviceStatus, DiscoveredDevice
from core.domain.enums.enums import Status
from rich.console import Console
from rich.table import Table

Expand All @@ -21,27 +22,45 @@ def __init__(self, console: Console):
self._console = console

def format_device_table(
self, devices: list[Any], title: str = "Shelly Devices"
self,
devices: list[Any],
title: str = "Shelly Devices",
include_beta: bool = False,
) -> None:
"""
Format and display a table of devices.

Args:
devices: List of device objects or dictionaries
title: Table title
include_beta: Whether beta-only updates should be surfaced as
available, instead of reported as up to date
"""
if not devices:
return

if devices and isinstance(devices[0], DiscoveredDevice):
self._format_discovered_devices_table(devices, title)
self._format_discovered_devices_table(devices, title, include_beta)
elif devices and isinstance(devices[0], DeviceStatus):
self._format_device_status_table(devices)
self._format_device_status_table(devices, include_beta)
else:
self._format_legacy_device_table(devices, title)

def _effective_status(self, device: DiscoveredDevice, include_beta: bool) -> str:

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't avoid the ts copy, but for the Python side what do you think about a small method on DiscoveredDevice in core, so the rule and its tests live in one place?

"""The status to display, downgrading a beta-only update to "no
update needed" when beta visibility is off — mirrors the same
allowed-channels rule the web UI applies."""
status = device.status
if (
not include_beta
and str(status) == Status.UPDATE_AVAILABLE.value
and getattr(device, "available_firmware_channel", None) == "beta"
):
return Status.NO_UPDATE_NEEDED.value
return status

def _format_discovered_devices_table(
self, devices: list[DiscoveredDevice], title: str
self, devices: list[DiscoveredDevice], title: str, include_beta: bool = False
) -> None:
"""Format table for DiscoveredDevice entities."""
table = Table(title=title)
Expand All @@ -61,17 +80,19 @@ def _format_discovered_devices_table(
device.model_name or device.device_type or "Unknown",
device.device_name or "Unknown",
device.firmware_version or "Unknown",
format_device_status(device.status),
format_device_status(self._effective_status(device, include_beta)),
response_time,
)

self._console.print(table)

def _format_device_status_table(self, devices: list[DeviceStatus]) -> None:
def _format_device_status_table(
self, devices: list[DeviceStatus], include_beta: bool = False
) -> None:
"""Format table for DeviceStatus entities."""

for device_status in devices:
self.format_detailed_device_status(device_status)
self.format_detailed_device_status(device_status, include_beta)

def _format_legacy_device_table(self, devices: list[Any], title: str) -> None:
"""Legacy format for backward compatibility."""
Expand Down Expand Up @@ -106,7 +127,9 @@ def _format_legacy_device_table(self, devices: list[Any], title: str) -> None:

self._console.print(table)

def format_detailed_device_status(self, device_status: DeviceStatus) -> None:
def format_detailed_device_status(
self, device_status: DeviceStatus, include_beta: bool = False
) -> None:
"""Format detailed component information for a single device."""
from rich.columns import Columns
from rich.panel import Panel
Expand All @@ -132,18 +155,33 @@ def format_detailed_device_status(self, device_status: DeviceStatus) -> None:
# Show available firmware updates with version information
if system_info.available_updates:
device_summary = device_status.get_device_summary()
available_updates = device_summary.get("available_updates", {})
raw_available_updates = device_summary.get("available_updates", {})
available_updates = (
raw_available_updates
if include_beta
else {
channel: info
for channel, info in raw_available_updates.items()
if channel != "beta"
}
)

if available_updates:
system_content.append("[yellow]Updates Available:[/yellow]")
for update_type, update_info in available_updates.items():
version = update_info.get("version", "Unknown")
name = update_info.get("name", update_type) or update_type
system_content.append(f" [cyan]• {name}:[/cyan] {version}")
else:
elif not raw_available_updates:
# The raw component data has entries but none carried a
# usable version, so get_device_summary filtered
# everything out — unrelated to beta visibility.
system_content.append(
f"[yellow]Updates Available:[/yellow] {len(system_info.available_updates)}"
)
# else: only a beta release exists and beta visibility is
# off — show nothing, exactly like a device with no
# updates at all. No partial hint.

system_panel = Panel(
"\n".join(system_content),
Expand Down
4 changes: 2 additions & 2 deletions packages/cli/src/cli/use_cases/device/device_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,6 @@ async def _check_device_status(

return results

def display_results(self, results: list[Any]) -> None:
def display_results(self, results: list[Any], include_beta: bool = False) -> None:
"""Display status results to console."""
self._result_formatter.format_device_table(results)
self._result_formatter.format_device_table(results, include_beta=include_beta)
10 changes: 8 additions & 2 deletions packages/cli/src/cli/use_cases/device/scan_devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,17 +60,23 @@ async def execute(self, request: DeviceScanRequest) -> list[Any]:
return devices_found

def display_results(
self, devices_found: list[Any], show_table: bool = True
self,
devices_found: list[Any],
show_table: bool = True,
include_beta: bool = False,
) -> None:
"""
Display scan results to console.

Args:
devices_found: List of discovered devices
show_table: Whether to show device table
include_beta: Whether beta-only updates count as available
"""
if show_table:
self._result_formatter.format_device_table(devices_found)
self._result_formatter.format_device_table(
devices_found, include_beta=include_beta
)

if devices_found:
self._console.print(
Expand Down
Loading
Loading