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
181 changes: 181 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,181 @@
# Prusa Connect Python Client

[![PyPI version](https://badge.fury.io/py/prusa-connect.svg)](https://badge.fury.io/py/prusa-connect)
[![License: AGPL v3](https://img.shields.io/badge/License-AGPL%20v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
[![Python Versions](https://img.shields.io/pypi/pyversions/prusa-connect.svg)](https://pypi.org/project/prusa-connect/)

Control your Prusa 3D printers programmatically with Python. This library provides a frictionless, strongly-typed interface for the Prusa Connect API.

**Features:**
* **Zero-Config Authentication:** Log in once via CLI, use everywhere in Python.
* **Strong Typing:** Full Pydantic models for printers, jobs, cameras, and files.
* **Batteries Included:** Retries, timeouts, and error handling out of the box.
* **CLI Tool:** Managing printers from the terminal.

---

## 1. Installation

Install the package with the CLI tools (recommended for easiest setup):

```bash
pip install "prusa-connect[cli]"
```

Or install the lightweight library only:

```bash
pip install prusa-connect
```

## 2. Quickstart

### Step 1: Authenticate

Run the following command in your terminal to log in to your Prusa Account. This will save a secure token locally.

```bash
prusactl list-printers
```
*Follow the interactive prompts to enter your credentials and 2FA code.*

### Step 2: Hello World

Create a Python script (`hello_prusa.py`) to list your printers. The client automatically loads the credentials you just saved.

```python
from prusa_connect import PrusaConnectClient

# Credentials are automatically loaded from your environment or local file
client = PrusaConnectClient()

print("My Printers:")
for printer in client.get_printers():
status = printer.printer_state or "UNKNOWN"
print(f"- {printer.name} ({status})")

if printer.telemetry:
print(f" Temp: {printer.telemetry.temp_nozzle}°C")
```

Run it:

```bash
python hello_prusa.py
```

---

## 3. How-to Guides

### Headless Authentication (CI/CD)

For environments where interactive login isn't possible (like CI/CD pipelines or servers), you can use environment variables.

1. **Option A: Raw Token**
Set the `PRUSA_TOKEN` environment variable with your raw JWT access token.

```bash
export PRUSA_TOKEN="ey..."
```

2. **Option B: Token JSON**
Set the `PRUSA_TOKENS_JSON` environment variable with the full JSON object containing access and refresh tokens.

```bash
export PRUSA_TOKENS_JSON='{"access_token": "...", "refresh_token": "..."}'
```

The `PrusaConnectClient` will automatically detect these variables.

### Controlling a Printer

You can send commands like PAUSE, RESUME, or STOP.

```python
client = PrusaConnectClient()

# Get your printer's UUID (e.g., from client.get_printers())
printer_uuid = "c0ffee-uuid-1234"

# Pause the print
client.send_command(printer_uuid, "PAUSE_PRINT")
print("Printer paused.")
```

### Accessing Cameras

Fetch the latest snapshot from your printer's camera.

```python
cameras = client.get_cameras()

if cameras:
cam = cameras[0]
print(f"Taking snapshot from {cam.name}...")

# Get binary image data
image_data = client.get_snapshot(cam.id)

with open("snapshot.jpg", "wb") as f:
f.write(image_data)
print("Saved to snapshot.jpg")
```

### Managing Files

List files on your team's storage.

```python
teams = client.get_teams()
if teams:
my_team_id = teams[0].id
files = client.get_file_list(my_team_id)

for file in files:
print(f"{file.name} ({file.size} bytes)")
```

---

## 4. Reference

### `PrusaConnectClient`

The main entry point for the API.

**Initialization:**
```python
client = PrusaConnectClient(
credentials=None, # Auto-loads if None
base_url="https://connect.prusa3d.com/app",
timeout=30.0
)
```

**Key Methods:**
* `get_printers() -> list[Printer]`
* `get_printer(uuid) -> Printer`
* `send_command(uuid, command)`
* `get_cameras() -> list[Camera]`
* `get_snapshot(camera_id) -> bytes`
* `get_team_jobs(team_id) -> list[Job]`

### Data Models

All responses are validated Pydantic models.

* `Printer`: `uuid`, `name`, `printer_state`, `telemetry` (temps), `job` (current status).
* `Job`: `state`, `progress`, `time_remaining`, `file`.
* `Camera`: `id`, `name`, `resolution`.

---

## 5. Explanation

### Authentication Flow

Prusa Connect uses a secure OAuth2-like flow with PKCE.
* **Interactive:** The CLI (`prusactl`) handles the complex exchange of username, password, and 2FA to obtain a **Refresh Token** and **Access Token**.
* **Refresh:** The `PrusaConnectClient` automatically checks if the Access Token is expired and uses the Refresh Token to get a new one, ensuring your long-running scripts don't break.
* **Storage:** Tokens are stored in `prusa_tokens.json` by default. Treat this file like a password.
47 changes: 46 additions & 1 deletion src/prusa_connect/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -261,8 +261,11 @@ def before_request(self, headers: MutableMapping[str, str | bytes]) -> None:
headers["Authorization"] = f"Bearer {self.tokens.access_token.raw_token}"

@classmethod
def from_file(cls, path: Path) -> "PrusaConnectCredentials | None":
def from_file(cls, path: Path | str) -> "PrusaConnectCredentials | None":
"""Factory: Load credentials from a JSON file."""
if isinstance(path, str):
path = Path(path)

try:
logger.debug(f"Loading credentials from {path}")
with path.open() as f:
Expand All @@ -279,6 +282,48 @@ def save_to_disk(new_data):
logger.info(f"No credentials found at {path.absolute()}")
return None

@classmethod
def from_env(cls) -> "PrusaConnectCredentials | None":
"""Factory: Load credentials from environment variables.

Checks:
1. PRUSA_TOKENS_JSON: A JSON string containing the full token set.
2. PRUSA_TOKEN: A raw Access Token (JWT).
"""
if json_str := os.environ.get("PRUSA_TOKENS_JSON"):
try:
return cls(json.loads(json_str))
except json.JSONDecodeError:
logger.warning("Invalid JSON in PRUSA_TOKENS_JSON")

if token := os.environ.get("PRUSA_TOKEN"):
try:
# Pydantic will attempt to parse the string into the AccessToken model
return cls({"access_token": token})
except Exception as e:
logger.debug(f"Could not create credentials from PRUSA_TOKEN: {e}")

return None

@classmethod
def load_default(cls) -> "PrusaConnectCredentials | None":
"""Factory: Attempt to load credentials from default locations.

Priority:
1. Environment Variables (PRUSA_TOKENS_JSON, PRUSA_TOKEN)
2. Local file 'prusa_tokens.json'
"""
# 1. Environment
if creds := cls.from_env():
return creds

# 2. Default file
default_path = Path("prusa_tokens.json")
if default_path.exists():
return cls.from_file(default_path)

return None


# --- PKCE & Login Flow Helpers ---

Expand Down
15 changes: 13 additions & 2 deletions src/prusa_connect/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from urllib3.util import Retry

from prusa_connect.__version__ import __version__
from prusa_connect.auth import PrusaConnectCredentials
from prusa_connect.exceptions import (
PrusaApiError,
PrusaAuthError,
Expand Down Expand Up @@ -62,19 +63,29 @@ class PrusaConnectClient:

def __init__(
self,
credentials: AuthStrategy,
credentials: AuthStrategy | None = None,
base_url: str = DEFAULT_BASE_URL,
timeout: float = DEFAULT_TIMEOUT,
) -> None:
"""Initializes the client.

Args:
credentials: An object adhering to the `AuthStrategy` protocol.
(e.g. `PrusaConnectCredentials`)
If None, attempts to load from environment or 'prusa_tokens.json'.
base_url: Optional override for the API endpoint.
timeout: Default timeout for API requests in seconds.
"""
self._base_url = base_url.rstrip("/")

if credentials is None:
credentials = PrusaConnectCredentials.load_default()

if credentials is None:
raise PrusaAuthError(
"No credentials provided and none found in default locations. "
"Please login via CLI (`prusactl list-printers`) or provide credentials explicitly."
)

self._credentials = credentials
self._timeout = timeout
self._session = requests.Session()
Expand Down
Loading