diff --git a/nanokvm/compatibility.py b/nanokvm/compatibility.py index 9a431b4..0b7da1b 100644 --- a/nanokvm/compatibility.py +++ b/nanokvm/compatibility.py @@ -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: diff --git a/nanokvm/components/hid.py b/nanokvm/components/hid.py index 485a057..c278ff9 100644 --- a/nanokvm/components/hid.py +++ b/nanokvm/components/hid.py @@ -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. diff --git a/nanokvm/components/network.py b/nanokvm/components/network.py index 5e507eb..6d42023 100644 --- a/nanokvm/components/network.py +++ b/nanokvm/components/network.py @@ -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( @@ -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: @@ -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( @@ -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( @@ -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( diff --git a/nanokvm/components/services.py b/nanokvm/components/services.py index edc4aef..9f3ae93 100644 --- a/nanokvm/components/services.py +++ b/nanokvm/components/services.py @@ -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], @@ -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( @@ -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, @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( diff --git a/nanokvm/components/storage.py b/nanokvm/components/storage.py index 9fccc5d..8d66332 100644 --- a/nanokvm/components/storage.py +++ b/nanokvm/components/storage.py @@ -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( diff --git a/nanokvm/components/system.py b/nanokvm/components/system.py index 612aba1..3117228 100644 --- a/nanokvm/components/system.py +++ b/nanokvm/components/system.py @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( @@ -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( diff --git a/nanokvm/components/users.py b/nanokvm/components/users.py index bfc44da..4d531c9 100644 --- a/nanokvm/components/users.py +++ b/nanokvm/components/users.py @@ -4,7 +4,7 @@ from aiohttp import hdrs -from ..compatibility import require_application_version, require_hardware +from ..compatibility import require_hardware from ..models.common import HWFamily from ..models.non_pro import ( CreateUserReq, @@ -27,8 +27,7 @@ def _user_path(username: str, suffix: str = "") -> str: class UserController(Controller): """Implement user account operations behind the public client facade.""" - @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_users(self) -> GetUsersRsp: """List the user accounts on the device.""" return await self._session.api_request_json( @@ -37,8 +36,7 @@ async def get_users(self) -> GetUsersRsp: response_model=GetUsersRsp, ) - @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 create_user( self, username: str, password: str, role: UserRole | str = UserRole.USER ) -> None: @@ -53,8 +51,7 @@ async def create_user( ), ) - @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 update_user( self, username: str, @@ -74,14 +71,12 @@ async def update_user( data=UpdateUserReq(role=role, enabled=enabled), ) - @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 delete_user(self, username: str) -> None: """Delete a user account and end its sessions.""" await self._session.api_request_json(hdrs.METH_DELETE, _user_path(username)) - @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_user_password(self, username: str, new_password: str) -> None: """Set another user's password and end that user's sessions. diff --git a/nanokvm/components/video.py b/nanokvm/components/video.py index ddf91f0..8a1bb9c 100644 --- a/nanokvm/components/video.py +++ b/nanokvm/components/video.py @@ -6,7 +6,7 @@ from aiohttp import hdrs -from ..compatibility import require_application_version, require_hardware +from ..compatibility import require_hardware from ..models.common import HWFamily, HWVersion from ..models.non_pro import ( GetHdmiStateRsp, @@ -45,8 +45,7 @@ class VideoController(Controller): """Implement video operations behind the public client facade.""" - @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_input_region(self) -> GetInputRegionRsp: """Get the configured non-Pro input region.""" return await self._session.api_request_json( @@ -55,8 +54,7 @@ async def get_input_region(self) -> GetInputRegionRsp: response_model=GetInputRegionRsp, ) - @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_input_region( self, mode: InputRegionMode, @@ -91,8 +89,7 @@ async def set_input_region( ), ) - @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_input_resolution(self) -> GetInputResolutionRsp: """Get the current non-Pro input frame resolution.""" return await self._session.api_request_json( @@ -110,8 +107,7 @@ async def set_screen(self, setting: ScreenSettingType, value: int) -> None: data=SetScreenReq(type=setting, value=value), ) - @require_hardware(HWVersion.PCIE) - @require_application_version(non_pro="2.2.8") + @require_hardware(HWVersion.PCIE, min_version="2.2.8") async def get_hdmi_state(self) -> GetHdmiStateRsp: """Get the HDMI state (PCIe variant).""" return await self._session.api_request_json( @@ -120,26 +116,22 @@ async def get_hdmi_state(self) -> GetHdmiStateRsp: response_model=GetHdmiStateRsp, ) - @require_hardware(HWVersion.PCIE) - @require_application_version(non_pro="2.1.5") + @require_hardware(HWVersion.PCIE, min_version="2.1.5") async def reset_hdmi(self) -> None: """Reset the HDMI connection.""" await self._session.api_request_json(hdrs.METH_POST, "/vm/hdmi/reset") - @require_hardware(HWVersion.PCIE) - @require_application_version(non_pro="2.2.8") + @require_hardware(HWVersion.PCIE, min_version="2.2.8") async def enable_hdmi(self) -> None: """Enable the HDMI connection.""" await self._session.api_request_json(hdrs.METH_POST, "/vm/hdmi/enable") - @require_hardware(HWVersion.PCIE) - @require_application_version(non_pro="2.2.8") + @require_hardware(HWVersion.PCIE, min_version="2.2.8") async def disable_hdmi(self) -> None: """Disable the HDMI connection.""" await self._session.api_request_json(hdrs.METH_POST, "/vm/hdmi/disable") - @require_hardware(HWVersion.PCIE) - @require_application_version(non_pro="2.5.0") + @require_hardware(HWVersion.PCIE, min_version="2.5.0") async def set_hdmi_idle_timeout(self, minutes: int) -> None: """Set the HDMI capture idle timeout in minutes; zero disables it.""" if not 0 <= minutes <= 10080: @@ -205,8 +197,7 @@ async def switch_edid(self, edid: EdidValue) -> None: data=SwitchEdidReq(edid=edid), ) - @require_hardware(HWFamily.PRO) - @require_application_version(pro="1.2.3") + @require_hardware(HWFamily.PRO, min_version="1.2.3") async def get_custom_edid_list(self) -> GetCustomEdidListRsp: """Get custom EDID list.""" return await self._session.api_request_json( @@ -215,8 +206,7 @@ async def get_custom_edid_list(self) -> GetCustomEdidListRsp: response_model=GetCustomEdidListRsp, ) - @require_hardware(HWFamily.PRO) - @require_application_version(pro="1.2.3") + @require_hardware(HWFamily.PRO, min_version="1.2.3") async def upload_edid(self, file_path: str | PathLike[str]) -> UploadEdidRsp: """Upload a custom EDID.""" return await self._session.upload_file( @@ -225,8 +215,7 @@ async def upload_edid(self, file_path: str | PathLike[str]) -> UploadEdidRsp: response_model=UploadEdidRsp, ) - @require_hardware(HWFamily.PRO) - @require_application_version(pro="1.2.3") + @require_hardware(HWFamily.PRO, min_version="1.2.3") async def delete_edid(self, edid: str) -> None: """Delete a custom EDID.""" await self._session.api_request_json( @@ -235,8 +224,7 @@ async def delete_edid(self, edid: str) -> None: data=DeleteEdidReq(edid=edid), ) - @require_hardware(HWFamily.PRO) - @require_application_version(pro="1.2.6") + @require_hardware(HWFamily.PRO, min_version="1.2.6") async def set_rate_control_mode(self, mode: RateControlMode) -> None: """Set the stream rate control mode (CBR/VBR).""" await self._session.api_request_json( @@ -272,8 +260,7 @@ async def set_gop(self, gop: int) -> None: data=SetGopReq(gop=gop), ) - @require_hardware(HWFamily.PRO) - @require_application_version(pro="1.2.8") + @require_hardware(HWFamily.PRO, min_version="1.2.8") async def set_fps(self, fps: int) -> None: """Set the stream FPS.""" await self._session.api_request_json( diff --git a/pyproject.toml b/pyproject.toml index 6ef1294..85cc4d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,24 +47,9 @@ required-version = ">=0.5.0" target-version = "py311" line-length = 88 -exclude = [ - ".venv", - ".git", - ".tox", - "docs", - "venv", - "bin", - "lib", - "deps", - "build", -] - [tool.ruff.lint] select = ["E", "F", "UP", "B", "SIM", "I"] -[tool.ruff.lint.per-file-ignores] -"tests/*.py" = ["F811"] - [tool.ruff.lint.isort] force-sort-within-sections = true known-first-party = [ @@ -94,7 +79,4 @@ source = ["nanokvm"] exclude_lines = [ "pragma: no cover", "if TYPE_CHECKING:", - "if typing.TYPE_CHECKING:", - "raise NotImplementedError", - "raise NotImplementedError()", ]