tmp

app/octoprint_client.py

@@ -0,0 +1,152 @@
+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)
+
+    # -------------------------------------------------------------------------
+    # 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})
+
+    # -------------------------------------------------------------------------
+    # 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