174 lines
7.2 KiB
Python
174 lines
7.2 KiB
Python
import requests
|
|
from urllib.parse import urljoin
|
|
|
|
class OctoPrintClient:
|
|
"""
|
|
Client for interacting with the OctoPrint API using Application Keys or standard API Keys.
|
|
Designed to be easily extensible.
|
|
"""
|
|
def __init__(self, base_url, api_key=None):
|
|
self.base_url = base_url.rstrip('/')
|
|
self.api_key = api_key
|
|
self.session = requests.Session()
|
|
if self.api_key:
|
|
self.session.headers.update({"X-Api-Key": self.api_key})
|
|
|
|
def _request(self, method, endpoint, **kwargs):
|
|
"""Internal method to handle API requests and standard parsing."""
|
|
url = urljoin(self.base_url, endpoint)
|
|
response = self.session.request(method, url, **kwargs)
|
|
response.raise_for_status()
|
|
|
|
# Octoprint often returns 204 No Content for successful commands
|
|
if response.status_code == 204:
|
|
return True
|
|
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return response.text
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Application Key Workflow
|
|
# -------------------------------------------------------------------------
|
|
def request_app_key(self, app_name="My OctoPrint App"):
|
|
"""
|
|
Step 1: Start the Application Key authorization flow.
|
|
Returns: (app_token, client_token)
|
|
You should poll verify_app_key(client_token) until the user allows it in OctoPrint UI.
|
|
"""
|
|
data = self._request("POST", "/plugin/appkeys/request", json={"app": app_name})
|
|
return data.get("app_token"), data.get("client_token")
|
|
|
|
def verify_app_key(self, client_token):
|
|
"""
|
|
Step 2: Check if the requested app key is approved.
|
|
Returns True if authorized, False if still pending.
|
|
Raises an exception if denied or timed out.
|
|
"""
|
|
url = urljoin(self.base_url, "/plugin/appkeys/probe")
|
|
# App Key probe requires passing the client_token as a Bearer token
|
|
# Don't use self._request since it uses the established session/api_key
|
|
response = requests.get(url, headers={"Authorization": f"Bearer {client_token}"})
|
|
|
|
if response.status_code == 204:
|
|
return True
|
|
elif response.status_code == 202:
|
|
return False # Pending approval
|
|
else:
|
|
raise Exception(f"App key request denied or expired (HTTP {response.status_code})")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Files
|
|
# -------------------------------------------------------------------------
|
|
def get_files(self, location="local"):
|
|
"""
|
|
Retrieve all files available on OctoPrint.
|
|
location: 'local' (internal storage) or 'sdcard' (SD card on printer)
|
|
"""
|
|
return self._request("GET", f"/api/files/{location}")
|
|
|
|
def select_file(self, location, path, print_after_select=False):
|
|
"""Select a file, and optionally start printing it immediately."""
|
|
payload = {"command": "select", "print": print_after_select}
|
|
return self._request("POST", f"/api/files/{location}/{path}", json=payload)
|
|
|
|
def upload_file(self, location, path_to_file, filename_override=None):
|
|
"""Upload a file to OctoPrint"""
|
|
with open(path_to_file, 'rb') as f:
|
|
files = {'file': (filename_override or path_to_file.split('/')[-1], f, 'application/octet-stream')}
|
|
url = urljoin(self.base_url, f"/api/files/{location}")
|
|
response = self.session.post(url, files=files)
|
|
response.raise_for_status()
|
|
if response.status_code == 204:
|
|
return True
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return response.text
|
|
|
|
def delete_file(self, location, path):
|
|
"""Delete a file from OctoPrint"""
|
|
return self._request("DELETE", f"/api/files/{location}/{path}")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Printer Status
|
|
# -------------------------------------------------------------------------
|
|
def get_printer_status(self):
|
|
"""
|
|
Get the current printer state (e.g., temperatures, operational state).
|
|
Note: If printer is disconnected, this may return an HTTP error.
|
|
"""
|
|
try:
|
|
return self._request("GET", "/api/printer")
|
|
except requests.HTTPError as e:
|
|
if e.response.status_code == 409:
|
|
return {"state": {"text": "Offline/Disconnected"}}
|
|
raise
|
|
|
|
def get_job_info(self):
|
|
"""Get information about the current print job and progress."""
|
|
return self._request("GET", "/api/job")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Printer Control
|
|
# -------------------------------------------------------------------------
|
|
def start_print(self):
|
|
"""Start the print job (Requires a file to be selected first)."""
|
|
return self._request("POST", "/api/job", json={"command": "start"})
|
|
|
|
def pause_print(self, action="pause"):
|
|
"""
|
|
Pause, resume, or toggle the print job.
|
|
action: 'pause', 'resume', or 'toggle'
|
|
"""
|
|
return self._request("POST", "/api/job", json={"command": "pause", "action": action})
|
|
|
|
def cancel_print(self):
|
|
"""Cancel the current print job."""
|
|
return self._request("POST", "/api/job", json={"command": "cancel"})
|
|
|
|
def send_gcode(self, commands):
|
|
"""
|
|
Send a string or list of G-Code commands directly to the printer.
|
|
Example: send_gcode("G28") or send_gcode(["G28", "G29"])
|
|
"""
|
|
if isinstance(commands, str):
|
|
commands = [commands]
|
|
return self._request("POST", "/api/printer/command", json={"commands": commands})
|
|
|
|
def home_axes(self, axes=["x", "y", "z"]):
|
|
"""Convenience method to home the printer axes."""
|
|
return self._request("POST", "/api/printer/printhead", json={"command": "home", "axes": axes})
|
|
|
|
def auto_leveling(self):
|
|
return self.send_gcode("G29")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Webcam / Video
|
|
# -------------------------------------------------------------------------
|
|
def get_webcam_stream_url(self):
|
|
"""
|
|
Attempts to fetch the configured webcam stream URL from OctoPrint settings.
|
|
Provides a fallback if settings are inaccessible.
|
|
"""
|
|
try:
|
|
settings = self._request("GET", "/api/settings")
|
|
stream_url = settings.get("webcam", {}).get("streamUrl", "/webcam/?action=stream")
|
|
if stream_url.startswith("/"):
|
|
return urljoin(self.base_url, stream_url)
|
|
return stream_url
|
|
except requests.HTTPError:
|
|
# Fallback standard URL
|
|
return urljoin(self.base_url, "/webcam/?action=stream")
|
|
|
|
# --- Example Usage / Extensibility test ---
|
|
if __name__ == "__main__":
|
|
# Example snippet of how to use the client:
|
|
OCTOPRINT_URL = "http://octopi.local"
|
|
# OCTOPRINT_KEY = "YOUR_APP_KEY_HERE"
|
|
|
|
# client = OctoPrintClient(OCTOPRINT_URL, OCTOPRINT_KEY)
|
|
# print(client.get_printer_status())
|
|
pass
|