Attempt too support both API versions

pyprusalink/__init__.py

@@ -3,70 +3,70 @@
 
 from aiohttp import ClientSession
 from pyprusalink.client import ApiClient
-from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus, VersionInfo
-from pyprusalink.types_legacy import LegacyPrinterStatus
+from pyprusalink.types import FileInfo, FilesInfo, JobInfo, PrinterInfo, VersionInfo
 
 
 class PrusaLink:
     """Wrapper for the Prusalink API.
 
     Data format can be found here:
-    https://github.com/prusa3d/Prusa-Link-Web/blob/master/spec/openapi.yaml
+    https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/master/lib/WUI/link_content/basic_gets.cpp
     """
 
-    def __init__(
-        self, session: ClientSession, host: str, username: str, password: str
-    ) -> None:
+    def __init__(self, session: ClientSession, host: str, api_key: str) -> None:
         """Initialize the PrusaLink class."""
-        self.client = ApiClient(
-            session=session, host=host, username=username, password=password
-        )
+        self.client = ApiClient.get_api_key_client(session, host, api_key)
 
-    async def cancel_job(self, jobId: int) -> None:
+    async def cancel_job(self) -> None:
         """Cancel the current job."""
-        async with self.client.request("DELETE", f"/api/v1/job/{jobId}"):
+        async with self.client.request("POST", "/api/job", {"command": "cancel"}):
             pass
 
-    async def pause_job(self, jobId: int) -> None:
-        """Pause a job."""
-        async with self.client.request("PUT", f"/api/v1/job/{jobId}/pause"):
+    async def resume_job(self) -> None:
+        """Resume a paused job."""
+        async with self.client.request(
+            "POST", "/api/job", {"command": "pause", "action": "resume"}
+        ):
             pass
 
-    async def resume_job(self, jobId: int) -> None:
-        """Resume a paused job."""
-        async with self.client.request("PUT", f"/api/v1/job/{jobId}/resume"):
+    async def pause_job(self) -> None:
+        """Pause the current job."""
+        async with self.client.request(
+            "POST", "/api/job", {"command": "pause", "action": "pause"}
+        ):
             pass
 
     async def get_version(self) -> VersionInfo:
         """Get the version."""
         async with self.client.request("GET", "/api/version") as response:
             return await response.json()
 
-    async def get_legacy_printer(self) -> LegacyPrinterStatus:
-        """Get the legacy printer endpoint."""
+    async def get_printer(self) -> PrinterInfo:
+        """Get the printer."""
         async with self.client.request("GET", "/api/printer") as response:
             return await response.json()
 
-    async def get_info(self) -> PrinterInfo:
-        """Get the printer."""
-        async with self.client.request("GET", "/api/v1/info") as response:
+    async def get_job(self) -> JobInfo:
+        """Get current job."""
+        async with self.client.request("GET", "/api/job") as response:
             return await response.json()
 
-    async def get_status(self) -> PrinterStatus:
-        """Get the printer."""
-        async with self.client.request("GET", "/api/v1/status") as response:
+    async def get_file(self, path: str) -> FileInfo:
+        """Get specific file info."""
+        async with self.client.request("GET", f"/api/files{path}") as response:
             return await response.json()
 
-    async def get_job(self) -> JobInfo:
-        """Get current job."""
-        async with self.client.request("GET", "/api/v1/job") as response:
-            # when there is no job running we'll an empty document that will fail to parse
-            if response.status == 204:
-                return {}
+    async def get_files(self) -> FilesInfo:
+        """Get all files."""
+        async with self.client.request("GET", "/api/files?recursive=true") as response:
             return await response.json()
 
-    # Prusa Link Web UI still uses the old endpoints and it seems that the new v1 endpoint doesn't support this yet
-    async def get_file(self, path: str) -> bytes:
-        """Get a files such as Thumbnails or Icons. Path comes from the current job['file']['refs']['thumbnail']"""
-        async with self.client.request("GET", path) as response:
+    async def get_small_thumbnail(self, path: str) -> bytes:
+        """Get a small thumbnail."""
+        async with self.client.request("GET", f"/thumb/s{path}") as response:
+            return await response.read()
+
+    async def get_large_thumbnail(self, path: str) -> bytes:
+        """Get a large thumbnail."""
+        async with self.client.request("GET", f"/thumb/l{path}") as response:
             return await response.read()

pyprusalink/client.py

@@ -6,24 +6,42 @@
 import hashlib
 
 from aiohttp import ClientResponse, ClientSession
-from pyprusalink.types import Conflict, InvalidAuth
+from pyprusalink.error import Conflict, InvalidAuth
 
 
 class ApiClient:
     def __init__(
-        self, session: ClientSession, host: str, username: str, password: str
+        self,
+        session: ClientSession,
+        host: str,
+        username: str | None = None,
+        password: str | None = None,
+        api_key: str | None = None,
     ) -> None:
         self._session = session
         self.host = host
         self._username = username
         self._password = password
+        self._api_key = api_key
         self._lock = asyncio.Lock()
 
         self._realm = ""
         self._nonce = ""
         self._qop = ""
 
-    def _generate_headers(self, method: str, path: str) -> dict:
+    @classmethod
+    def get_http_digest_client(
+        cls, session: ClientSession, host: str, username: str, password: str
+    ) -> ApiClient:
+        return cls(session, host, username, password)
+
+    @classmethod
+    def get_api_key_client(
+        cls, session: ClientSession, host: str, api_key: str
+    ) -> ApiClient:
+        return cls(session, host, api_key=api_key)
+
+    def _generate_digest_headers(self, method: str, path: str) -> dict:
         """Generates new Authorization with the current _nonce, method and path."""
         ha1 = hashlib.md5(
             f"{self._username}:{self._realm}:{self._password}".encode()
@@ -38,6 +56,12 @@ def _generate_headers(self, method: str, path: str) -> dict:
 
         return headers
 
+    def _generate_headers(self, method: str, path: str) -> dict:
+        if self._username or self._password:
+            return self._generate_digest_headers(method, path)
+        else:
+            return {"X-Api-Key": self._api_key}
+
     def _extract_digest_params(self, headers: dict[str]) -> None:
         """Extract realm, nonce key from Digest Auth header"""
         header_value = headers.get("WWW-Authenticate", "")

pyprusalink/error.py

@@ -0,0 +1,10 @@
+class PrusaLinkError(Exception):
+    """Base class for PrusaLink errors."""
+
+
+class InvalidAuth(PrusaLinkError):
+    """Error to indicate there is invalid auth."""
+
+
+class Conflict(PrusaLinkError):
+    """Error to indicate the command hit a conflict."""

pyprusalink/types.py

@@ -1,131 +1,42 @@
-from enum import Enum
 from typing import TypedDict
 
-"""Types of the v1 API. Source: https://github.com/prusa3d/Prusa-Link-Web/blob/master/spec/openapi.yaml"""
-
-
-class PrusaLinkError(Exception):
-    """Base class for PrusaLink errors."""
-
-
-class InvalidAuth(PrusaLinkError):
-    """Error to indicate there is invalid auth."""
-
-
-class Conflict(PrusaLinkError):
-    """Error to indicate the command hit a conflict."""
-
-
-class Capabilities(TypedDict):
-    """API Capabilities"""
-
-    upload_by_put: bool | None
+"""Types of the non-versioned API. Source: https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/v4.4.1/lib/WUI/link_content/basic_gets.cpp"""
 
 
 class VersionInfo(TypedDict):
     """Version data."""
 
     api: str
-    version: str
-    printer: str
+    server: str
     text: str
-    firmware: str
-    sdk: str | None
-    capabilities: Capabilities | None
+    hostname: str
 
 
 class PrinterInfo(TypedDict):
-    """Printer informations."""
-
-    mmu: bool | None
-    name: str | None
-    location: str | None
-    farm_mode: bool | None
-    nozzle_diameter: float | None
-    min_extrusion_temp: int | None
-    serial: str | None
-    sd_ready: bool | None
-    active_camera: bool | None
-    hostname: str | None
-    port: str | None
-    network_error_chime: bool | None
-
-
-class StatusInfo(TypedDict):
-    """Status of the printer."""
-
-    ok: bool | None
-    message: str | None
+    """Printer data."""
 
+    telemetry: dict
+    temperature: dict
+    state: dict
 
-class PrinterState(Enum):
-    """Printer state as Enum."""
 
-    IDLE = "IDLE"
-    BUSY = "BUSY"
-    PRINTING = "PRINTING"
-    PAUSED = "PAUSED"
-    FINISHED = "FINISHED"
-    STOPPED = "STOPPED"
-    ERROR = "ERROR"
-    ATTENTION = "ATTENTION"
-    READY = "READY"
-
-
-class PrinterStatusInfo(TypedDict):
-    """Printer information."""
-
-    state: PrinterState
-    temp_nozzle: float | None
-    target_nozzle: float | None
-    temp_bed: float | None
-    target_bed: float | None
-    axis_x: float | None
-    axis_y: float | None
-    axis_z: float | None
-    flow: int | None
-    speed: int | None
-    fan_hotend: int | None
-    fan_print: int | None
-    status_printer: StatusInfo | None
-    status_connect: StatusInfo | None
-
-
-class PrinterStatus(TypedDict):
-    """Printer status."""
-
-    printer: PrinterStatusInfo
-
-
-class PrintFileRefs(TypedDict):
-    """Additional Files for the current Job"""
+class JobInfo(TypedDict):
+    """Job data."""
 
-    download: str | None
-    icon: str | None
-    thumbnail: str | None
+    state: str
+    job: dict | None
 
 
-class JobFilePrint(TypedDict):
-    """Currently printed file informations."""
+class FileInfo(TypedDict):
+    """File data."""
 
     name: str
-    display_name: str | None
-    path: str
-    display_path: str | None
-    size: int | None
-    m_timestamp: int
-    meta: dict | None
-    refs: PrintFileRefs | None
+    origin: str
+    size: int
+    refs: dict
 
 
-class JobInfo(TypedDict):
-    """Job information."""
+class FilesInfo(TypedDict):
+    """Files data."""
 
-    id: int
-    state: str
-    progress: int
-    time_remaining: int | None
-    time_printing: int
-    inaccurate_estimates: bool | None
-    serial_print: bool | None
-    file: JobFilePrint | None
+    files: list[FileInfo]

pyprusalink/v1/__init__.py

@@ -0,0 +1,72 @@
+"""Prusalink API."""
+from __future__ import annotations
+
+from aiohttp import ClientSession
+from pyprusalink.client import ApiClient
+from pyprusalink.types_legacy import LegacyPrinterStatus
+from pyprusalink.v1.types import JobInfo, PrinterInfo, PrinterStatus, VersionInfo
+
+
+class PrusaLink:
+    """Wrapper for the Prusalink API.
+
+    Data format can be found here:
+    https://github.com/prusa3d/Prusa-Link-Web/blob/master/spec/openapi.yaml
+    """
+
+    def __init__(
+        self, session: ClientSession, host: str, username: str, password: str
+    ) -> None:
+        """Initialize the PrusaLink class."""
+        self.client = ApiClient.get_http_digest_client(
+            session=session, host=host, username=username, password=password
+        )
+
+    async def cancel_job(self, jobId: int) -> None:
+        """Cancel the current job."""
+        async with self.client.request("DELETE", f"/api/v1/job/{jobId}"):
+            pass
+
+    async def pause_job(self, jobId: int) -> None:
+        """Pause a job."""
+        async with self.client.request("PUT", f"/api/v1/job/{jobId}/pause"):
+            pass
+
+    async def resume_job(self, jobId: int) -> None:
+        """Resume a paused job."""
+        async with self.client.request("PUT", f"/api/v1/job/{jobId}/resume"):
+            pass
+
+    async def get_version(self) -> VersionInfo:
+        """Get the version."""
+        async with self.client.request("GET", "/api/version") as response:
+            return await response.json()
+
+    async def get_legacy_printer(self) -> LegacyPrinterStatus:
+        """Get the legacy printer endpoint."""
+        async with self.client.request("GET", "/api/printer") as response:
+            return await response.json()
+
+    async def get_info(self) -> PrinterInfo:
+        """Get the printer."""
+        async with self.client.request("GET", "/api/v1/info") as response:
+            return await response.json()
+
+    async def get_status(self) -> PrinterStatus:
+        """Get the printer."""
+        async with self.client.request("GET", "/api/v1/status") as response:
+            return await response.json()
+
+    async def get_job(self) -> JobInfo:
+        """Get current job."""
+        async with self.client.request("GET", "/api/v1/job") as response:
+            # when there is no job running we'll an empty document that will fail to parse
+            if response.status == 204:
+                return {}
+            return await response.json()
+
+    # Prusa Link Web UI still uses the old endpoints and it seems that the new v1 endpoint doesn't support this yet
+    async def get_file(self, path: str) -> bytes:
+        """Get a files such as Thumbnails or Icons. Path comes from the current job['file']['refs']['thumbnail']"""
+        async with self.client.request("GET", path) as response:
+            return await response.read()