Add Virtual Machine Manager support

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/vscode/devcontainers/python:0-3.9
+FROM mcr.microsoft.com/vscode/devcontainers/python:3.9-bookworm
 
 # install test requirements
 COPY requirements*.txt /tmp/pip-tmp/

README.md

@@ -379,6 +379,86 @@ if __name__ == "__main__":
     asyncio.run(main())
 ```
 
+## Virtual Machine Manager usage
+
+### Show information about all existing guests
+
+```python
+import asyncio
+import aiohttp
+from synology_dsm import SynologyDSM
+
+
+async def main():
+    print("Creating Valid API")
+    async with aiohttp.ClientSession(
+        connector=aiohttp.TCPConnector(verify_ssl=False)
+    ) as session:
+        await do(session)
+
+
+async def do(session: aiohttp.ClientSession):
+    api = SynologyDSM(session, "<IP/DNS>", "<port>", "<username>", "<password>")
+    await api.login()
+
+    await api.virtual_machine_manager.update()
+
+    guests = api.virtual_machine_manager.get_all_guests()
+
+    for guest in guests:
+        print(f"############### {guest.name} ###############")
+        print(f"autorun: {guest.autorun}")
+        print(f"description: {guest.description}")
+        print(f"guest_id: {guest.guest_id}")
+        print(f"status: {guest.status}")
+        print(f"vcpu_num: {guest.vcpu_num}")
+        print(f"vram_size: {guest.vram_size / 1024} MiBytes")
+        print(f"host_cpu_usage: {guest.host_cpu_usage / 10} %")
+        print(f"host_ram_usage: {round(guest.host_ram_usage / 1024,1)} MiBytes")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Perform power actions on guests
+
+```python
+import asyncio
+import aiohttp
+from synology_dsm import SynologyDSM
+
+
+async def main():
+    print("Creating Valid API")
+    async with aiohttp.ClientSession(
+        connector=aiohttp.TCPConnector(verify_ssl=False)
+    ) as session:
+        await do(session)
+
+
+async def do(session: aiohttp.ClientSession):
+    api = SynologyDSM(session, "<IP/DNS>", "<port>", "<username>", "<password>")
+    await api.login()
+
+    await api.virtual_machine_manager.update()
+
+    # start a guest
+    await api.virtual_machine_manager.guest_poweron("{guest.guest_id}")
+
+    # power off a guest
+    await api.virtual_machine_manager.guest_poweroff("{guest.guest_id}")
+
+    # graceful shutdown a guest (needs working guest-agent, else it is not graceful)
+    await api.virtual_machine_manager.guest_shutdown("{guest.guest_id}")
+
+    # graceful restart a guest (needs working guest-agent, else it is not graceful)
+    await api.virtual_machine_manager.guest_restart("{guest.guest_id}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
 # Credits / Special Thanks
 
 - [@florianeinfalt](https://github.com/florianeinfalt)

src/synology_dsm/api/virtual_machine_manager/__init__.py

@@ -0,0 +1,68 @@
+"""Synology Virtual Machine Manager API models."""
+
+from __future__ import annotations
+
+from synology_dsm.api import SynoBaseApi
+
+from .guest import SynoVmmGuest
+
+
+class SynoVirtualMachineManager(SynoBaseApi["dict[str, SynoVmmGuest]"]):
+    """Class containing Virtual Machine Guests."""
+
+    API_KEY = "SYNO.Virtualization.*"
+    GUEST_API_KEY = "SYNO.Virtualization.Guest"
+    ACTION_API_KEY = "SYNO.Virtualization.Guest.Action"
+
+    async def update(self) -> None:
+        """Updates Virtual Machine Manager data."""
+        raw_data = await self._dsm.get(self.GUEST_API_KEY, "list")
+        print(raw_data)
+        if not isinstance(raw_data, dict) or (data := raw_data.get("data")) is None:
+            return
+
+        for guest in data["guests"]:
+            if guest["guest_id"] in self._data:
+                self._data[guest["guest_id"]].update(guest)
+            else:
+                self._data[guest["guest_id"]] = SynoVmmGuest(guest)
+
+    def get_all_guests(self) -> list[SynoVmmGuest]:
+        """Return a list of all vmm guests."""
+        return list(self._data.values())
+
+    def get_guest(self, guest_id: str) -> SynoVmmGuest | None:
+        """Return vmm guest by guest_id."""
+        return self._data.get(guest_id)
+
+    async def _guest_action(self, guest_id: str, action: str) -> bool | None:
+        raw_data = await self._dsm.post(
+            self.ACTION_API_KEY,
+            "pwr_ctl",
+            {
+                "guest_id": guest_id,
+                "action": action,
+            },
+        )
+        if (
+            isinstance(raw_data, dict)
+            and (result := raw_data.get("success")) is not None
+        ):
+            return bool(result)
+        return None
+
+    async def guest_poweron(self, guest_id: str) -> bool | None:
+        """Power on a vmm guest."""
+        return await self._guest_action(guest_id, "poweron")
+
+    async def guest_poweroff(self, guest_id: str) -> bool | None:
+        """Power off a vmm guest."""
+        return await self._guest_action(guest_id, "poweroff")
+
+    async def guest_shutdown(self, guest_id: str) -> bool | None:
+        """Graceful shutdown a vmm guest."""
+        return await self._guest_action(guest_id, "shutdown")
+
+    async def guest_restart(self, guest_id: str) -> bool | None:
+        """Graceful restart a vmm guest."""
+        return await self._guest_action(guest_id, "reboot")

src/synology_dsm/api/virtual_machine_manager/guest.py

@@ -0,0 +1,80 @@
+"""VirtualMachineManager guest."""
+
+from __future__ import annotations
+
+from typing import TypedDict, Union
+
+SynoVmmGuestData = TypedDict(
+    "SynoVmmGuestData",
+    {
+        "autorun": int,
+        "desc": str,
+        "guest_id": str,
+        "name": str,
+        "ram_used": int,
+        "status": str,
+        "vcpu_num": int,
+        "vcpu_usage": Union[str, int],  # empty str when offline
+        "vram_size": int,
+    },
+    total=False,
+)
+
+
+class SynoVmmGuest:
+    """An representation of a Synology Virtual Machine Manager guest."""
+
+    def __init__(self, data: SynoVmmGuestData) -> None:
+        """Initialize a Virtual Machine Manager guest."""
+        self._data: SynoVmmGuestData = data
+
+    def update(self, data: SynoVmmGuestData) -> None:
+        """Update the vmm guest."""
+        self._data = data
+
+    @property
+    def autorun(self) -> bool:
+        """Return autorun of the vmm guest."""
+        return bool(self._data["autorun"])
+
+    @property
+    def description(self) -> str:
+        """Return description of the vmm guest."""
+        return self._data["desc"]
+
+    @property
+    def guest_id(self) -> str:
+        """Return guest_id of the vmm guest."""
+        return self._data["guest_id"]
+
+    @property
+    def name(self) -> str:
+        """Return name of the vmm guest."""
+        return self._data["name"]
+
+    @property
+    def status(self) -> str:
+        """Return status of the vmm guest."""
+        return self._data["status"]
+
+    @property
+    def host_cpu_usage(self) -> int:
+        """Return host cpu usage in one thousandth of the vmm guest."""
+        if isinstance(self._data["vcpu_usage"], str):
+            return 0
+        return self._data["vcpu_usage"]
+
+    @property
+    def host_ram_usage(self) -> int:
+        """Return host ram usage in KiByte of the vmm guest."""
+        return self._data["ram_used"]
+
+    @property
+    def vcpu_num(self) -> int:
+        """Return number of vcpu of the vmm guest."""
+        return self._data["vcpu_num"]
+
+    @property
+    def vram_size(self) -> int:
+        """Return size of vram in KiByte of the vmm guest."""
+        return self._data["vram_size"]

src/synology_dsm/const.py

@@ -155,15 +155,16 @@
     900: "The storage is in invalid",
     901: "Failed to set a host to a virtual machine",
     902: "The virtual machine does not have a host",
-    903: "Failed to power on a virtual machine due to insufficient CPU threads",
-    904: "Failed to power on a virtual machine due to insufficient memory",
-    905: "The status of virtual machine is online",
+    903: "Can't shutdown the guest, it is not running",
+    904: "Can't power off the guest, it is not running",
+    905: "Can't restart the guest, it is not running",
     906: "MAC conflict",
     907: "Failed to create virtual machine because the selected image is not found",
     908: "The status of virtual machine is offline",
     909: "Failed to power on a virtual machine due to insufficient CPU threads for reservation on the host",  # pylint: disable=line-too-long
     910: "Failed to power on the virtual machine because there is no corresponding networking on the host",  # pylint: disable=line-too-long
     911: "Only the VirtIO hard disk controller can be used to boot the virtual machine remotely. Virtual machines with UEFI enabled cannot be powered on remotely",  # pylint: disable=line-too-long
+    939: "Guest already running",
     1000: "Cannot find task_id",
     1001: "Need Virtual Machine Manager Pro",
     1400: "The result of image creating is partial success",

src/synology_dsm/synology_dsm.py

@@ -26,6 +26,7 @@
 from .api.photos import SynoPhotos
 from .api.storage.storage import SynoStorage
 from .api.surveillance_station import SynoSurveillanceStation
+from .api.virtual_machine_manager import SynoVirtualMachineManager
 from .const import API_AUTH, API_INFO, SENSITIV_PARAMS
 from .exceptions import (
     SynologyDSMAPIErrorException,
@@ -104,6 +105,7 @@ def __init__(
         self._system: SynoCoreSystem | None = None
         self._utilisation: SynoCoreUtilization | None = None
         self._upgrade: SynoCoreUpgrade | None = None
+        self._vmm: SynoVirtualMachineManager | None = None
 
         try:
             IPv6Address(dsm_ip)
@@ -437,6 +439,9 @@ async def update(
         if self._upgrade:
             update_methods.append(self._upgrade.update())
 
+        if self._vmm:
+            update_methods.append(self._vmm.update())
+
         await asyncio.gather(*update_methods)
 
     def reset(self, api: SynoBaseApi | str) -> bool:
@@ -477,6 +482,9 @@ def reset(self, api: SynoBaseApi | str) -> bool:
             if api == SynoSurveillanceStation.API_KEY:
                 self._surveillance = None
                 return True
+            if api == SynoVirtualMachineManager.API_KEY:
+                self._vmm = None
+                return True
         if isinstance(api, SynoCoreExternalUSB):
             self._external_usb = None
             return True
@@ -507,6 +515,9 @@ def reset(self, api: SynoBaseApi | str) -> bool:
         if isinstance(api, SynoSurveillanceStation):
             self._surveillance = None
             return True
+        if isinstance(api, SynoVirtualMachineManager):
+            self._vmm = None
+            return True
         return False
 
     @property
@@ -592,3 +603,10 @@ def utilisation(self) -> SynoCoreUtilization:
         if not self._utilisation:
             self._utilisation = SynoCoreUtilization(self)
         return self._utilisation
+
+    @property
+    def virtual_machine_manager(self) -> SynoVirtualMachineManager:
+        """Gets NAS virtual machine manager informations."""
+        if not self._vmm:
+            self._vmm = SynoVirtualMachineManager(self)
+        return self._vmm

tests/__init__.py

@@ -18,6 +18,7 @@
 from synology_dsm.api.photos import SynoPhotos
 from synology_dsm.api.storage.storage import SynoStorage
 from synology_dsm.api.surveillance_station import SynoSurveillanceStation
+from synology_dsm.api.virtual_machine_manager import SynoVirtualMachineManager
 from synology_dsm.const import API_AUTH, API_INFO
 from synology_dsm.exceptions import SynologyDSMRequestException
 
@@ -74,6 +75,7 @@
     DSM_7_FOTO_ITEMS_SEARCHED,
     DSM_7_FOTO_ITEMS_SHARED_ALBUM,
     DSM_7_FOTO_SHARED_ITEMS,
+    DSM_7_VMM_GUESTS,
 )
 from .const import (
     DEVICE_TOKEN,
@@ -124,6 +126,7 @@
         "DSM_INFORMATION": DSM_7_DSM_INFORMATION,
         "FOTO_ALBUMS": DSM_7_FOTO_ALBUMS,
         "FOTO_ITEMS": DSM_7_FOTO_ITEMS,
+        "VMM_GUESTS": DSM_7_VMM_GUESTS,
     },
 }
 
@@ -297,6 +300,9 @@ async def _execute_request(self, method, url, params, **kwargs):
             if SynoPhotos.BROWSE_ITEM_FOTOTEAM_API_KEY in url:
                 return DSM_7_FOTO_SHARED_ITEMS
 
+            if SynoVirtualMachineManager.GUEST_API_KEY in url:
+                return DSM_7_VMM_GUESTS
+
             if SynoStorage.API_KEY in url:
                 return API_SWITCHER[self.dsm_version]["STORAGE_STORAGE"][
                     self.disks_redundancy

tests/api_data/dsm_7/__init__.py

@@ -18,6 +18,7 @@
     DSM_7_FOTO_ITEMS_SHARED_ALBUM,
     DSM_7_FOTO_SHARED_ITEMS,
 )
+from .virtual_machine_manager.const_7_vmm import DSM_7_VMM_GUESTS
 
 __all__ = [
     "DSM_7_AUTH_LOGIN",
@@ -33,4 +34,5 @@
     "DSM_7_FOTO_ITEMS_SHARED_ALBUM",
     "DSM_7_FOTO_ITEMS_SEARCHED",
     "DSM_7_FOTO_SHARED_ITEMS",
+    "DSM_7_VMM_GUESTS",
 ]