spoolman2/scripts/slicer_post_process.py

#!/usr/bin/env python3
"""
Elegoo Slicer / Orca Slicer Post-Processing Script for Spoolman Integration.

This script parses G-code files to extract filament usage information and creates
a pending print job in Spoolman. After the print completes, use the Spoolman
Print Queue UI to confirm the print (deducts filament) or cancel it (flags for weighing).

Usage:
    1. In Elegoo Slicer, go to Printer Settings > Post-processing Scripts
    2. Add the path to this script: /path/to/slicer_post_process.py
    3. Configure the environment variables below or create a config file

Configuration:
    Set these environment variables:
    - SPOOLMAN_URL: Your Spoolman instance URL (e.g., http://192.168.0.11:8000)
    - SPOOLMAN_SPOOL_ID: Default spool ID to use (or set per-filament in slicer)

    Or create a file at ~/.config/spoolman/slicer.conf:
        [spoolman]
        url = http://192.168.0.11:8000
        spool_id = 1

The script will:
    1. Read the G-code file passed as argument
    2. Parse Elegoo/Orca slicer comments for filament usage
    3. Create a pending print job in Spoolman via API
    4. The job appears in the Print Queue for confirmation after printing
"""

import configparser
import os
import re
import sys
import urllib.request
import json
from pathlib import Path


def get_config() -> tuple[str, int]:
    """Get Spoolman URL and default spool ID from environment or config file."""
    url = os.environ.get("SPOOLMAN_URL")
    spool_id = os.environ.get("SPOOLMAN_SPOOL_ID")

    if not url:
        config_path = Path.home() / ".config" / "spoolman" / "slicer.conf"
        if config_path.exists():
            config = configparser.ConfigParser()
            config.read(config_path)
            url = config.get("spoolman", "url", fallback=None)
            spool_id = config.get("spoolman", "spool_id", fallback=None)

    if not url:
        print("ERROR: SPOOLMAN_URL not set. Set environment variable or create config file.", file=sys.stderr)
        sys.exit(1)

    if not spool_id:
        print("ERROR: SPOOLMAN_SPOOL_ID not set. Set environment variable or create config file.", file=sys.stderr)
        sys.exit(1)

    return url.rstrip("/"), int(spool_id)


def parse_gcode(filepath: str) -> dict:
    """
    Parse G-code file for filament usage information.

    Looks for Elegoo/Orca Slicer comments like:
    ; total filament used [g] = 12.34
    ; filament used [mm] = 4567.89
    ; filament used [g] = 12.34

    Also checks for PrusaSlicer format:
    ; filament used [mm] = 4567.89
    ; filament used [g] = 12.34
    """
    result = {
        "filename": Path(filepath).name,
        "filament_used_g": None,
        "filament_used_mm": None,
    }

    # Patterns for different slicers
    patterns = {
        # Elegoo/Orca format
        "total_filament_g": re.compile(r";\s*total filament used \[g\]\s*=\s*([\d.]+)", re.IGNORECASE),
        "filament_g": re.compile(r";\s*filament used \[g\]\s*=\s*([\d.]+)", re.IGNORECASE),
        "filament_mm": re.compile(r";\s*filament used \[mm\]\s*=\s*([\d.]+)", re.IGNORECASE),
        # Alternative formats
        "filament_weight": re.compile(r";\s*(?:total )?filament (?:weight|cost)\s*=\s*([\d.]+)\s*g", re.IGNORECASE),
        "filament_length": re.compile(r";\s*(?:total )?filament\s*=\s*([\d.]+)\s*mm", re.IGNORECASE),
    }

    try:
        with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
            # Only read first 500 lines and last 500 lines for efficiency
            lines = f.readlines()
            header = lines[:500]
            footer = lines[-500:] if len(lines) > 500 else []
            check_lines = header + footer

            for line in check_lines:
                if not line.startswith(";"):
                    continue

                # Check each pattern
                for key, pattern in patterns.items():
                    match = pattern.search(line)
                    if match:
                        value = float(match.group(1))
                        if "g" in key or "weight" in key:
                            if result["filament_used_g"] is None:
                                result["filament_used_g"] = value
                        elif "mm" in key or "length" in key:
                            if result["filament_used_mm"] is None:
                                result["filament_used_mm"] = value

    except Exception as e:
        print(f"Warning: Error parsing G-code: {e}", file=sys.stderr)

    return result


def create_print_job(url: str, spool_id: int, gcode_info: dict) -> bool:
    """Create a pending print job in Spoolman."""
    if gcode_info["filament_used_g"] is None:
        print("Warning: Could not find filament weight in G-code. Skipping job creation.", file=sys.stderr)
        return False

    api_url = f"{url}/api/v1/print-job"

    payload = {
        "spool_id": spool_id,
        "filename": gcode_info["filename"],
        "filament_used_g": gcode_info["filament_used_g"],
    }

    if gcode_info["filament_used_mm"] is not None:
        payload["filament_used_mm"] = gcode_info["filament_used_mm"]

    try:
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            api_url,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )

        with urllib.request.urlopen(req, timeout=10) as response:
            if response.status in (200, 201):
                result = json.loads(response.read().decode("utf-8"))
                print(f"Created print job #{result['id']}: {gcode_info['filename']} ({gcode_info['filament_used_g']:.1f}g)")
                return True
            else:
                print(f"Error: API returned status {response.status}", file=sys.stderr)
                return False

    except urllib.error.HTTPError as e:
        error_body = e.read().decode("utf-8") if e.fp else ""
        print(f"Error: API request failed: {e.code} {e.reason} - {error_body}", file=sys.stderr)
        return False
    except urllib.error.URLError as e:
        print(f"Error: Could not connect to Spoolman: {e.reason}", file=sys.stderr)
        return False
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        return False


def main():
    """Main entry point."""
    if len(sys.argv) < 2:
        print("Usage: slicer_post_process.py <gcode_file>", file=sys.stderr)
        print(__doc__)
        sys.exit(1)

    gcode_path = sys.argv[1]

    if not os.path.exists(gcode_path):
        print(f"Error: G-code file not found: {gcode_path}", file=sys.stderr)
        sys.exit(1)

    # Get configuration
    url, spool_id = get_config()

    # Parse G-code
    print(f"Parsing G-code: {gcode_path}")
    gcode_info = parse_gcode(gcode_path)

    if gcode_info["filament_used_g"]:
        print(f"  Filament: {gcode_info['filament_used_g']:.2f}g", end="")
        if gcode_info["filament_used_mm"]:
            print(f" ({gcode_info['filament_used_mm']:.1f}mm)")
        else:
            print()
    else:
        print("  Warning: Could not parse filament usage from G-code")

    # Create print job
    success = create_print_job(url, spool_id, gcode_info)

    if success:
        print(f"Print job created in Spoolman ({url})")
        print("After printing, use the Print Queue to confirm or cancel the job.")
    else:
        print("Failed to create print job. The print will proceed without tracking.", file=sys.stderr)

    # Always exit 0 so the slicer continues with the print
    sys.exit(0)


if __name__ == "__main__":
    main()