diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5860ef7..41f4307 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -15,8 +15,8 @@ jobs: strategy: matrix: python-version: - - "3.9" - - "3.10" + - "3.11" + - "3.12" steps: - uses: actions/checkout@v4.1.1 diff --git a/.gitignore b/.gitignore index c71ccfa..1a2e7bd 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ venv *.egg-info *.log dist +main.py diff --git a/pyprusalink/__init__.py b/pyprusalink/__init__.py index 37286ae..e80120b 100644 --- a/pyprusalink/__init__.py +++ b/pyprusalink/__init__.py @@ -1,146 +1,72 @@ """Prusalink API.""" from __future__ import annotations -from collections.abc import AsyncGenerator -from contextlib import asynccontextmanager -from typing import TypedDict - -from aiohttp import ClientResponse, ClientSession - - -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 VersionInfo(TypedDict): - """Version data.""" - - api: str - server: str - text: str - hostname: str - - -class PrinterInfo(TypedDict): - """Printer data.""" - - telemetry: dict - temperature: dict - state: dict - - -class JobInfo(TypedDict): - """Job data.""" - - state: str - job: dict | None - - -class FileInfo(TypedDict): - """File data.""" - - name: str - origin: str - size: int - refs: dict - - -class FilesInfo(TypedDict): - """Files data.""" - - files: list[FileInfo] +from aiohttp import ClientSession +from pyprusalink.client import ApiClient +from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus, VersionInfo +from pyprusalink.types_legacy import LegacyPrinterStatus class PrusaLink: """Wrapper for the Prusalink API. Data format can be found here: - https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/master/lib/WUI/link_content/basic_gets.cpp + https://github.com/prusa3d/Prusa-Link-Web/blob/master/spec/openapi.yaml """ - def __init__(self, session: ClientSession, host: str, api_key: str) -> None: + def __init__( + self, session: ClientSession, host: str, username: str, password: str + ) -> None: """Initialize the PrusaLink class.""" - self._session = session - self.host = host - self._api_key = api_key + self.client = ApiClient( + session=session, host=host, username=username, password=password + ) - async def cancel_job(self) -> None: + async def cancel_job(self, jobId: int) -> None: """Cancel the current job.""" - async with self.request("POST", "api/job", {"command": "cancel"}): + async with self.client.request("DELETE", f"/api/v1/job/{jobId}"): pass - async def resume_job(self) -> None: - """Resume a paused job.""" - async with self.request( - "POST", "api/job", {"command": "pause", "action": "resume"} - ): + 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 pause_job(self) -> None: - """Pause the current job.""" - async with self.request( - "POST", "api/job", {"command": "pause", "action": "pause"} - ): + 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.request("GET", "api/version") as response: + async with self.client.request("GET", "/api/version") as response: return await response.json() - async def get_printer(self) -> PrinterInfo: - """Get the printer.""" - async with self.request("GET", "api/printer") as response: + 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_job(self) -> JobInfo: - """Get current job.""" - async with self.request("GET", "api/job") as response: + 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_file(self, path: str) -> FileInfo: - """Get specific file info.""" - async with self.request("GET", f"api/files{path}") as response: + 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_files(self) -> FilesInfo: - """Get all files.""" - async with self.request("GET", "api/files?recursive=true") as response: + 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() - async def get_small_thumbnail(self, path: str) -> bytes: - """Get a small thumbnail.""" - async with self.request("GET", f"thumb/s{path}") as response: + # 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() - - async def get_large_thumbnail(self, path: str) -> bytes: - """Get a large thumbnail.""" - async with self.request("GET", f"thumb/l{path}") as response: - return await response.read() - - @asynccontextmanager - async def request( - self, method: str, path: str, json: dict | None = None - ) -> AsyncGenerator[ClientResponse, None]: - """Make a request to the PrusaLink API.""" - url = f"{self.host}/{path}" - headers = {"X-Api-Key": self._api_key} - - async with self._session.request( - method, url, headers=headers, json=json - ) as response: - if response.status == 401: - raise InvalidAuth() - if response.status == 409: - raise Conflict() - - response.raise_for_status() - yield response diff --git a/pyprusalink/client.py b/pyprusalink/client.py new file mode 100644 index 0000000..c4e5f5f --- /dev/null +++ b/pyprusalink/client.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager +import hashlib + +from aiohttp import ClientResponse, ClientSession +from pyprusalink.types import Conflict, InvalidAuth + + +class ApiClient: + def __init__( + self, session: ClientSession, host: str, username: str, password: str + ) -> None: + self._session = session + self.host = host + self._username = username + self._password = password + self._lock = asyncio.Lock() + + self._realm = "" + self._nonce = "" + self._qop = "" + + def _generate_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() + ).hexdigest() + ha2 = hashlib.md5(f"{method}:{path}".encode()).hexdigest() + response_value = hashlib.md5(f"{ha1}:{self._nonce}:{ha2}".encode()).hexdigest() + + headers = { + "Authorization": f'Digest username="{self._username}", realm="{self._realm}", ' + f'nonce="{self._nonce}", uri="{path}", response="{response_value}"' + } + + return headers + + def _extract_digest_params(self, headers: dict[str]) -> None: + """Extract realm, nonce key from Digest Auth header""" + header_value = headers.get("WWW-Authenticate", "") + if not header_value.startswith("Digest"): + return + + header_value = header_value[len("Digest ") :] + + params = {} + parts = header_value.split(",") + for part in parts: + key, value = part.strip().split("=", 1) + params[key.strip()] = value.strip(' "') + + self._realm = params["realm"] + self._nonce = params["nonce"] + self._qop = params.get("qop", "auth") + + @asynccontextmanager + async def request( + self, + method: str, + path: str, + json_data: dict | None = None, + try_auth: bool = True, + ) -> AsyncGenerator[ClientResponse, None]: + """Make a request to the PrusaLink API.""" + url = f"{self.host}{path}" + headers = self._generate_headers(method=method, path=path) + async with self._session.request( + method, url, json=json_data, headers=headers + ) as response: + if response.status == 401: + if try_auth: + self._extract_digest_params(response.headers) + async with self.request(method, path, json_data, False) as response: + yield response + return + else: + raise InvalidAuth() + + if response.status == 409: + raise Conflict() + + response.raise_for_status() + yield response diff --git a/pyprusalink/types.py b/pyprusalink/types.py new file mode 100644 index 0000000..9916391 --- /dev/null +++ b/pyprusalink/types.py @@ -0,0 +1,131 @@ +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 + + +class VersionInfo(TypedDict): + """Version data.""" + + api: str + version: str + printer: str + text: str + firmware: str + sdk: str | None + capabilities: Capabilities | None + + +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 + + +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""" + + download: str | None + icon: str | None + thumbnail: str | None + + +class JobFilePrint(TypedDict): + """Currently printed file informations.""" + + name: str + display_name: str | None + path: str + display_path: str | None + size: int | None + m_timestamp: int + meta: dict | None + refs: PrintFileRefs | None + + +class JobInfo(TypedDict): + """Job information.""" + + id: int + state: str + progress: int + time_remaining: int | None + time_printing: int + inaccurate_estimates: bool | None + serial_print: bool | None + file: JobFilePrint | None diff --git a/pyprusalink/types_legacy.py b/pyprusalink/types_legacy.py new file mode 100644 index 0000000..4437e77 --- /dev/null +++ b/pyprusalink/types_legacy.py @@ -0,0 +1,15 @@ +from typing import TypedDict + +"""Legacy Types of the non v1 API. Source: https://github.com/prusa3d/Prusa-Link-Web/blob/master/spec/openapi-legacy.yaml""" + + +class LegacyPrinterTelemetry(TypedDict): + """Legacy Printer telemetry.""" + + material: str | None + + +class LegacyPrinterStatus(TypedDict): + """Legacy Printer status.""" + + telemetry: LegacyPrinterTelemetry | None