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
15 changes: 13 additions & 2 deletions nanokvm/compatibility.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,21 @@ def _version_at_least(version: str, minimum: str) -> bool:
return normalized_version >= normalized_minimum


def require_hardware(*requirements: HWVersion | HWFamily) -> Callable[[F], F]:
"""Restrict a method to specific hardware versions or families."""
def require_hardware(
*requirements: HWVersion | HWFamily, min_version: str | None = None
) -> Callable[[F], F]:
"""Restrict a method to specific hardware versions or families.

``min_version`` also requires that application version on the matched
hardware.
"""

def decorator(func: F) -> F:
if min_version is not None:
func = require_application_version(non_pro=min_version, pro=min_version)(
func
)

@functools.wraps(func)
async def wrapper(self: Controller, *args: Any, **kwargs: Any) -> Any:
if self._session._hw_version is None:
Expand Down
3 changes: 1 addition & 2 deletions nanokvm/components/hid.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,7 @@ async def get_hid_mode(self) -> GetHidModeRsp:
response_model=GetHidModeRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def get_keyboard_led_status(self) -> GetKeyboardLedStatusRsp:
"""Get host keyboard LED state on non-Pro firmware 2.5.0 and newer.

Expand Down
15 changes: 5 additions & 10 deletions nanokvm/components/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,8 +111,7 @@ async def set_wol_mac_name(self, mac: str, name: str) -> None:
data=SetMacNameReq(mac=mac, name=name),
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.4.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.4.1")
async def get_dns(self) -> GetDNSRsp:
"""Get DNS configuration."""
return await self._session.api_request_json(
Expand All @@ -121,8 +120,7 @@ async def get_dns(self) -> GetDNSRsp:
response_model=GetDNSRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.4.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.4.1")
async def set_dns(
self, mode: DNSMode | str, servers: list[str] | None = None
) -> None:
Expand All @@ -133,8 +131,7 @@ async def set_dns(
data=SetDNSReq(mode=DNSMode(mode), servers=servers or []),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.3")
@require_hardware(HWFamily.PRO, min_version="1.2.3")
async def get_static_ip(self) -> GetStaticIPRsp:
"""Get static IP configuration."""
return await self._session.api_request_json(
Expand All @@ -143,8 +140,7 @@ async def get_static_ip(self) -> GetStaticIPRsp:
response_model=GetStaticIPRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.3")
@require_hardware(HWFamily.PRO, min_version="1.2.3")
async def set_static_ip(self, enabled: bool, ip: str) -> None:
"""Set static IP configuration."""
await self._session.api_request_json(
Expand All @@ -153,8 +149,7 @@ async def set_static_ip(self, enabled: bool, ip: str) -> None:
data=SetStaticIPReq(enabled=enabled, ip=ip),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.7")
@require_hardware(HWFamily.PRO, min_version="1.1.7")
async def scan_wifi(self) -> ScanWifiRsp:
"""Scan for available WiFi networks."""
return await self._session.api_request_json(
Expand Down
45 changes: 15 additions & 30 deletions nanokvm/components/services.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,7 @@ async def update_application(self) -> None:
await self._session.api_request_json(hdrs.METH_POST, "/application/update")
self._session.forget_versions()

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.3.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.3.1")
async def update_application_offline(
self,
file_path: str | PathLike[str],
Expand Down Expand Up @@ -142,8 +141,7 @@ async def update_application_offline(
raise
self._session.forget_versions()

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.1")
async def get_update_server(self) -> GetUpdateServerRsp:
"""Get the custom application update-server configuration."""
return await self._session.api_request_json(
Expand All @@ -152,8 +150,7 @@ async def get_update_server(self) -> GetUpdateServerRsp:
response_model=GetUpdateServerRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.1")
async def set_update_server(
self,
enabled: bool,
Expand All @@ -167,8 +164,7 @@ async def set_update_server(
data=SetUpdateServerReq(enabled=enabled, url=url),
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def get_mcp_config(self) -> GetMCPConfigRsp:
"""Get MCP configuration and coordinated-control state."""
return await self._session.api_request_json(
Expand All @@ -177,8 +173,7 @@ async def get_mcp_config(self) -> GetMCPConfigRsp:
response_model=GetMCPConfigRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def set_mcp_enabled(self, enabled: bool) -> GetMCPConfigRsp:
"""Enable or disable MCP control."""
return await self._session.api_request_json(
Expand All @@ -188,8 +183,7 @@ async def set_mcp_enabled(self, enabled: bool) -> GetMCPConfigRsp:
data=SetMCPConfigReq(enabled=enabled),
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def regenerate_mcp_api_key(self) -> GetMCPConfigRsp:
"""Generate and return a new MCP API key."""
return await self._session.api_request_json(
Expand All @@ -198,8 +192,7 @@ async def regenerate_mcp_api_key(self) -> GetMCPConfigRsp:
response_model=GetMCPConfigRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def get_ai_control_status(self) -> AIControlStatusRsp:
"""Get the current coordinated-control owner and transition state."""
return await self._session.api_request_json(
Expand All @@ -208,8 +201,7 @@ async def get_ai_control_status(self) -> AIControlStatusRsp:
response_model=AIControlStatusRsp,
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.5.0")
@require_hardware(HWFamily.NON_PRO, min_version="2.5.0")
async def set_ai_control_mode(self, mode: AIControlMode) -> SetAIControlModeRsp:
"""Select the owner of coordinated input control."""
return await self._session.api_request_json(
Expand Down Expand Up @@ -282,56 +274,49 @@ async def tailscale_restart(self) -> None:
hdrs.METH_POST, "/extensions/tailscale/restart"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.4")
@require_hardware(HWFamily.PRO, min_version="1.1.4")
async def assistant_install(self) -> None:
"""Install assistant dependencies."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/assistant/install"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.4")
@require_hardware(HWFamily.PRO, min_version="1.1.4")
async def assistant_start(self) -> None:
"""Start assistant."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/assistant/start"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.5")
@require_hardware(HWFamily.PRO, min_version="1.1.5")
async def kvmadmin_install(self) -> None:
"""Install kvmadmin."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/kvmadmin/install"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.5")
@require_hardware(HWFamily.PRO, min_version="1.1.5")
async def kvmadmin_uninstall(self) -> None:
"""Uninstall kvmadmin."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/kvmadmin/uninstall"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.5")
@require_hardware(HWFamily.PRO, min_version="1.1.5")
async def kvmadmin_start(self) -> None:
"""Start kvmadmin."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/kvmadmin/start"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.5")
@require_hardware(HWFamily.PRO, min_version="1.1.5")
async def kvmadmin_stop(self) -> None:
"""Stop kvmadmin."""
await self._session.api_request_json(
hdrs.METH_POST, "/extensions/kvmadmin/stop"
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.5")
@require_hardware(HWFamily.PRO, min_version="1.1.5")
async def kvmadmin_status(self) -> GetKvmadminStatusRsp:
"""Get kvmadmin status."""
return await self._session.api_request_json(
Expand Down
3 changes: 1 addition & 2 deletions nanokvm/components/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -375,8 +375,7 @@ async def delete_image(self, file: str) -> None:
data=DeleteImageReq(file=file),
)

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.2.1")
@require_hardware(HWFamily.NON_PRO, min_version="2.2.1")
async def get_cdrom_status(self) -> GetCdRomRsp:
"""Check if the mounted image is in CD-ROM mode."""
return await self._session.api_request_json(
Expand Down
39 changes: 13 additions & 26 deletions nanokvm/components/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -210,8 +210,7 @@ async def switch_to_pikvm(self) -> None:
"""Switch the system image to PiKVM."""
await self._session.api_request_json(hdrs.METH_POST, "/vm/system/pikvm")

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.2.6")
@require_hardware(HWFamily.NON_PRO, min_version="2.2.6")
async def get_swap_size(self) -> int:
"""Get Swap size."""
rsp = await self._session.api_request_json(
Expand All @@ -221,8 +220,7 @@ async def get_swap_size(self) -> int:
)
return rsp.size

@require_hardware(HWFamily.NON_PRO)
@require_application_version(non_pro="2.2.6")
@require_hardware(HWFamily.NON_PRO, min_version="2.2.6")
async def set_swap_size(self, size_mb: int) -> None:
"""Set the Swap size."""
await self._session.api_request_json(
Expand All @@ -247,8 +245,7 @@ async def set_memory_limit(self, enabled: bool, limit_mb: int) -> None:
data=SetMemoryLimitReq(enabled=enabled, limit=limit_mb),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.7")
@require_hardware(HWFamily.PRO, min_version="1.2.7")
async def refresh_virtual_device(self, device: str) -> None:
"""Refresh a virtual device (e.g. emmc)."""
await self._session.api_request_json(
Expand All @@ -257,8 +254,7 @@ async def refresh_virtual_device(self, device: str) -> None:
data=RefreshVirtualDeviceReq(device=device),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def get_lcd_time_format(self) -> GetLcdTimeFormatRsp:
"""Get the LCD time format."""
return await self._session.api_request_json(
Expand All @@ -267,8 +263,7 @@ async def get_lcd_time_format(self) -> GetLcdTimeFormatRsp:
response_model=GetLcdTimeFormatRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def set_lcd_time_format(self, fmt: LcdTimeFormat | str) -> None:
"""Set the LCD time format (12h/24h)."""
await self._session.api_request_json(
Expand All @@ -277,8 +272,7 @@ async def set_lcd_time_format(self, fmt: LcdTimeFormat | str) -> None:
data=SetLcdTimeFormatReq(format=LcdTimeFormat(fmt)),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.7")
@require_hardware(HWFamily.PRO, min_version="1.2.7")
async def get_low_power(self) -> GetLowPowerRsp:
"""Get low power status."""
return await self._session.api_request_json(
Expand All @@ -287,8 +281,7 @@ async def get_low_power(self) -> GetLowPowerRsp:
response_model=GetLowPowerRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.7")
@require_hardware(HWFamily.PRO, min_version="1.2.7")
async def set_low_power(self, enable: bool) -> None:
"""Set low power mode."""
await self._session.api_request_json(
Expand Down Expand Up @@ -339,8 +332,7 @@ async def set_led_strip(
data=SetLedStripReq.model_validate(settings),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def get_timezone(self) -> GetTimeZoneRsp:
"""Get the configured timezone."""
return await self._session.api_request_json(
Expand All @@ -349,8 +341,7 @@ async def get_timezone(self) -> GetTimeZoneRsp:
response_model=GetTimeZoneRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def set_timezone(self, timezone: str) -> None:
"""Set the timezone."""
await self._session.api_request_json(
Expand All @@ -359,8 +350,7 @@ async def set_timezone(self, timezone: str) -> None:
data=SetTimeZoneReq(timezone=timezone),
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def get_time_status(self) -> GetTimeStatusRsp:
"""Get time synchronization status."""
return await self._session.api_request_json(
Expand All @@ -369,14 +359,12 @@ async def get_time_status(self) -> GetTimeStatusRsp:
response_model=GetTimeStatusRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.1.6")
@require_hardware(HWFamily.PRO, min_version="1.1.6")
async def sync_time(self) -> None:
"""Synchronize time."""
await self._session.api_request_json(hdrs.METH_POST, "/vm/time/sync")

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.10")
@require_hardware(HWFamily.PRO, min_version="1.2.10")
async def get_menubar_config(self) -> GetMenuBarConfigRsp:
"""Get menu bar configuration."""
return await self._session.api_request_json(
Expand All @@ -385,8 +373,7 @@ async def get_menubar_config(self) -> GetMenuBarConfigRsp:
response_model=GetMenuBarConfigRsp,
)

@require_hardware(HWFamily.PRO)
@require_application_version(pro="1.2.10")
@require_hardware(HWFamily.PRO, min_version="1.2.10")
async def set_menubar_config(self, disabled_items: list[str]) -> None:
"""Set menu bar configuration."""
await self._session.api_request_json(
Expand Down
Loading
Loading