What Is Filament Waste: in 3D Printing and Why It Matters

# gcode_waste_calculator.py
# Calculates estimated filament waste from G-code files
# Dependencies: regex (standard lib)
import re
import json
import sys
from pathlib import Path
from typing import Dict, List, Optional

class GCodeWasteCalculator:
    """Parses G-code to estimate waste from supports, brims, rafts, and infill"""

    # Regex patterns for G-code commands
    EXTRUDE_PATTERN = re.compile(r'E(-?\d+\.?\d*)')  # Matches E values (extrusion)
    LAYER_PATTERN = re.compile(r'; layer (\d+)')  # Matches layer comments from PrusaSlicer
    SUPPORT_START = re.compile(r'; support material start')  # Support start marker
    SUPPORT_END = re.compile(r'; support material end')  # Support end marker
    BRIM_START = re.compile(r'; brim start')  # Brim start marker
    BRIM_END = re.compile(r'; brim end')  # Brim end marker
    FILAMENT_DIAMETER = 1.75  # Standard filament diameter in mm
    FILAMENT_CROSS_SECTION = 3.14159 * (FILAMENT_DIAMETER / 2) ** 2  # mm²

    def __init__(self, gcode_path: Path):
        self.gcode_path = gcode_path
        self.extrusion_total = 0.0  # Total extrusion in mm
        self.extrusion_support = 0.0  # Support extrusion in mm
        self.extrusion_brim = 0.0  # Brim extrusion in mm
        self.extrusion_raft = 0.0  # Raft extrusion (if present)
        self.in_support = False
        self.in_brim = False
        self.layer_count = 0
        self.errors: List[str] = []

    def parse(self) -> Dict[str, float]:
        """Parse G-code file and return waste metrics"""
        if not self.gcode_path.exists():
            raise FileNotFoundError(f"G-code file not found: {self.gcode_path}")

        try:
            with open(self.gcode_path, 'r', encoding='utf-8') as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line or line.startswith(';'):
                        # Check for layer/support/brim markers
                        self._check_markers(line)
                        continue

                    # Extract extrusion value
                    e_match = self.EXTRUDE_PATTERN.search(line)
                    if e_match:
                        try:
                            e_val = float(e_match.group(1))
                            if e_val > 0:  # Only count positive extrusion (retractions are negative)
                                self.extrusion_total += e_val
                                if self.in_support:
                                    self.extrusion_support += e_val
                                elif self.in_brim:
                                    self.extrusion_brim += e_val
                        except ValueError:
                            self.errors.append(f"Invalid E value at line {line_num}: {line}")

                    # Track layer count
                    layer_match = self.LAYER_PATTERN.search(line)
                    if layer_match:
                        self.layer_count = int(layer_match.group(1))

        except PermissionError:
            raise PermissionError(f"No permission to read G-code file: {self.gcode_path}")
        except UnicodeDecodeError:
            raise ValueError(f"Invalid encoding in G-code file: {self.gcode_path}. Use UTF-8.")

        return self._generate_report()

    def _check_markers(self, line: str) -> None:
        """Update support/brim state based on G-code comments"""
        if self.SUPPORT_START.search(line):
            self.in_support = True
        elif self.SUPPORT_END.search(line):
            self.in_support = False
        elif self.BRIM_START.search(line):
            self.in_brim = True
        elif self.BRIM_END.search(line):
            self.in_brim = False

    def _generate_report(self) -> Dict[str, float]:
        """Convert extrusion mm to grams and generate JSON report"""
        # Convert extrusion mm to cm³ (extrusion is in mm of filament, cross section is mm²: mm * mm² = mm³ = 1e-3 cm³)
        total_cm3 = self.extrusion_total * self.FILAMENT_CROSS_SECTION / 1000  # mm³ to cm³
        support_cm3 = self.extrusion_support * self.FILAMENT_CROSS_SECTION / 1000
        brim_cm3 = self.extrusion_brim * self.FILAMENT_CROSS_SECTION / 1000

        # Assume PLA density: 1.24 g/cm³
        PLA_DENSITY = 1.24
        total_g = total_cm3 * PLA_DENSITY
        support_g = support_cm3 * PLA_DENSITY
        brim_g = brim_cm3 * PLA_DENSITY
        waste_g = support_g + brim_g
        waste_pct = (waste_g / total_g) * 100 if total_g > 0 else 0

        return {
            "total_filament_g": round(total_g, 2),
            "support_waste_g": round(support_g, 2),
            "brim_waste_g": round(brim_g, 2),
            "total_waste_g": round(waste_g, 2),
            "waste_percentage": round(waste_pct, 2),
            "layer_count": self.layer_count,
            "errors": self.errors
        }

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python gcode_waste_calculator.py ")
        sys.exit(1)

    gcode_path = Path(sys.argv[1])
    try:
        calculator = GCodeWasteCalculator(gcode_path)
        report = calculator.parse()
        print(json.dumps(report, indent=2))
    except Exception as e:
        print(f"Error parsing G-code: {e}", file=sys.stderr)
        sys.exit(1)

# octoprint_waste_tracker.py
# Tracks real-time filament usage via OctoPrint API and logs waste events
# Dependencies: requests, python-dotenv
# OctoPrint API docs: https://docs.octoprint.org/en/master/api/
import os
import time
import json
import requests
from typing import Dict, Optional, List
from dotenv import load_dotenv

load_dotenv()  # Load API key from .env file

class OctoPrintWasteTracker:
    """Integrates with OctoPrint to track actual vs estimated filament usage"""

    def __init__(self, base_url: str = "http://localhost:5000", api_key: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key or os.getenv("OCTOPRINT_API_KEY")
        if not self.api_key:
            raise ValueError("OctoPrint API key not provided. Set OCTOPRINT_API_KEY env var or pass api_key.")

        self.headers = {
            "X-Api-Key": self.api_key,
            "Content-Type": "application/json"
        }
        self.current_job_estimated_g: float = 0.0
        self.current_job_extruded_g: float = 0.0
        self.waste_events: List[Dict] = []
        self.session = requests.Session()

    def get_current_job(self) -> Optional[Dict]:
        """Fetch current print job details from OctoPrint"""
        try:
            response = self.session.get(
                f"{self.base_url}/api/job",
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            job_data = response.json()

            if job_data["state"] == "Printing":
                return job_data
            return None
        except requests.exceptions.ConnectionError:
            raise ConnectionError(f"Could not connect to OctoPrint at {self.base_url}")
        except requests.exceptions.HTTPError as e:
            raise ValueError(f"OctoPrint API error: {e.response.status_code} - {e.response.text}")
        except json.JSONDecodeError:
            raise ValueError("Invalid JSON response from OctoPrint API")

    def get_filament_usage(self) -> float:
        """Get total extruded filament in grams from OctoPrint"""
        try:
            response = self.session.get(
                f"{self.base_url}/api/printer/tool",
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            tool_data = response.json()
            # Extruded length is in mm, convert to grams (PLA: 1.75mm diameter, 1.24g/cm³)
            extruded_mm = tool_data.get("tool0", {}).get("extrudedLength", 0.0)
            filament_cross_section = 3.14159 * (1.75 / 2) ** 2  # mm²
            extruded_mm3 = extruded_mm * filament_cross_section  # mm³
            extruded_cm3 = extruded_mm3 / 1000  # cm³
            return extruded_cm3 * 1.24  # grams
        except Exception as e:
            raise ValueError(f"Failed to fetch filament usage: {e}")

    def update_job_estimate(self, job_data: Dict) -> None:
        """Update estimated filament from current job metadata"""
        if "job" in job_data and "filament" in job_data["job"]:
            filament_data = job_data["job"]["filament"]
            # Total estimated filament in grams (OctoPrint returns mm³)
            self.current_job_estimated_g = filament_data.get("tool0", {}).get("volume", 0.0) * 1.24 / 1000

    def check_waste(self) -> Optional[Dict]:
        """Compare actual vs estimated usage and log waste events"""
        if not self.current_job_estimated_g:
            return None

        actual_g = self.get_filament_usage()
        # Waste is actual usage minus estimated (if higher) plus support/brim estimates from slicer
        # For this example, assume 15% of estimated is support/brim waste
        estimated_waste_g = self.current_job_estimated_g * 0.15
        actual_waste_g = max(0, actual_g - (self.current_job_estimated_g - estimated_waste_g))
        waste_pct = (actual_waste_g / self.current_job_estimated_g) * 100 if self.current_job_estimated_g > 0 else 0

        if waste_pct > 20:  # Alert if waste exceeds 20%
            event = {
                "timestamp": time.time(),
                "estimated_g": self.current_job_estimated_g,
                "actual_g": actual_g,
                "waste_g": actual_waste_g,
                "waste_pct": round(waste_pct, 2),
                "job_name": job_data.get("job", {}).get("file", {}).get("name", "unknown")
            }
            self.waste_events.append(event)
            return event
        return None

    def run_tracker(self, poll_interval: int = 30) -> None:
        """Run tracker loop, poll OctoPrint every poll_interval seconds"""
        print(f"Starting OctoPrint waste tracker for {self.base_url}...")
        while True:
            try:
                job_data = self.get_current_job()
                if job_data:
                    self.update_job_estimate(job_data)
                    waste_event = self.check_waste()
                    if waste_event:
                        print(f"Waste alert: {waste_event['job_name']} - {waste_event['waste_pct']}% waste")
                        # In production, send to Slack/Prometheus here
                time.sleep(poll_interval)
            except KeyboardInterrupt:
                print("\nStopping tracker...")
                break
            except Exception as e:
                print(f"Tracker error: {e}", file=sys.stderr)
                time.sleep(poll_interval)

if __name__ == "__main__":
    tracker = OctoPrintWasteTracker(
        base_url=os.getenv("OCTOPRINT_URL", "http://localhost:5000"),
        api_key=os.getenv("OCTOPRINT_API_KEY")
    )
    tracker.run_tracker(poll_interval=30)

# slicer_waste_benchmark.py
# Benchmarks slicer settings to find lowest waste configuration for a given part
# Dependencies: subprocess, json
# PrusaSlicer CLI docs: https://github.com/prusa3d/PrusaSlicer/wiki/Command-line-interface
import subprocess
import json
import sys
import time
from pathlib import Path
from typing import Dict, List, Optional
import tempfile

class SlicerWasteBenchmark:
    """Runs PrusaSlicer CLI with different settings to benchmark waste output"""

    def __init__(self, prusaslicer_path: Path, model_path: Path):
        self.prusaslicer_path = prusaslicer_path
        self.model_path = model_path
        if not self.prusaslicer_path.exists():
            raise FileNotFoundError(f"PrusaSlicer not found at {self.prusaslicer_path}")
        if not self.model_path.exists():
            raise FileNotFoundError(f"Model file not found at {self.model_path}")

        # Test configurations: (infill_pattern, infill_density, support_on, brim_width)
        self.configs = [
            ("grid", 20, False, 3),
            ("grid", 20, True, 3),
            ("triangular", 20, False, 3),
            ("triangular", 20, True, 3),
            ("cubic", 15, False, 2),
            ("cubic", 15, True, 2),
        ]
        self.results: List[Dict] = []

    def run_slice(self, infill_pattern: str, infill_density: int, support_on: bool, brim_width: int) -> Optional[Dict]:
        """Run PrusaSlicer CLI with given settings and return waste metrics"""
        with tempfile.TemporaryDirectory() as tmpdir:
            output_gcode = Path(tmpdir) / "output.gcode"
            # Build PrusaSlicer CLI command
            cmd = [
                str(self.prusaslicer_path),
                str(self.model_path),
                "--output", str(output_gcode),
                "--infill-pattern", infill_pattern,
                "--infill-density", str(infill_density),
                "--brim-width", str(brim_width),
                "--gcode-comments",  # Enable comments for parsing
            ]
            if support_on:
                cmd.append("--support-material")

            try:
                # Run PrusaSlicer, capture output
                result = subprocess.run(
                    cmd,
                    capture_output=True,
                    text=True,
                    timeout=120  # 2 minute timeout per slice
                )
                result.check_returncode()

                # Parse G-code for waste (reuse logic from first script)
                if not output_gcode.exists():
                    raise ValueError("PrusaSlicer did not generate G-code output")

                # Extract waste from G-code comments (PrusaSlicer includes filament estimates)
                with open(output_gcode, 'r') as f:
                    for line in f:
                        if "filament used [g]" in line.lower():
                            # Line format: ; filament used [g] = 12.34
                            filament_g = float(line.split('=')[1].strip())
                        if "support material used [g]" in line.lower():
                            support_g = float(line.split('=')[1].strip())
                        if "brim used [g]" in line.lower():
                            brim_g = float(line.split('=')[1].strip())

                total_waste = support_g + brim_g
                waste_pct = (total_waste / filament_g) * 100 if filament_g > 0 else 0

                return {
                    "infill_pattern": infill_pattern,
                    "infill_density": infill_density,
                    "support_on": support_on,
                    "brim_width": brim_width,
                    "total_filament_g": filament_g,
                    "support_g": support_g,
                    "brim_g": brim_g,
                    "total_waste_g": total_waste,
                    "waste_percentage": round(waste_pct, 2)
                }

            except subprocess.TimeoutExpired:
                raise TimeoutError(f"PrusaSlicer timed out after 120s for config: {infill_pattern} {infill_density}%")
            except subprocess.CalledProcessError as e:
                raise ValueError(f"PrusaSlicer failed: {e.stderr}")
            except Exception as e:
                raise ValueError(f"Failed to process slice: {e}")

    def run_benchmark(self) -> List[Dict]:
        """Run all benchmark configurations and return results"""
        print(f"Running {len(self.configs)} benchmark configurations...")
        for i, (pattern, density, support, brim) in enumerate(self.configs, 1):
            print(f"Config {i}/{len(self.configs)}: {pattern} {density}% infill, support={support}, brim={brim}mm")
            try:
                result = self.run_slice(pattern, density, support, brim)
                if result:
                    self.results.append(result)
                    print(f"  Waste: {result['waste_percentage']}% ({result['total_waste_g']}g)")
            except Exception as e:
                print(f"  Error: {e}", file=sys.stderr)
        return self.results

    def save_results(self, output_path: Path) -> None:
        """Save benchmark results to JSON"""
        with open(output_path, 'w') as f:
            json.dump(self.results, f, indent=2)
        print(f"Results saved to {output_path}")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: python slicer_waste_benchmark.py  ")
        sys.exit(1)

    prusaslicer_path = Path(sys.argv[1])
    model_path = Path(sys.argv[2])

    try:
        benchmark = SlicerWasteBenchmark(prusaslicer_path, model_path)
        results = benchmark.run_benchmark()
        benchmark.save_results(Path("waste_benchmark_results.json"))

        # Print summary of lowest waste config
        if results:
            lowest_waste = min(results, key=lambda x: x["waste_percentage"])
            print("\nLowest waste configuration:")
            print(f"  Infill: {lowest_waste['infill_pattern']} {lowest_waste['infill_density']}%")
            print(f"  Support: {lowest_waste['support_on']}, Brim: {lowest_waste['brim_width']}mm")
            print(f"  Waste: {lowest_waste['waste_percentage']}% ({lowest_waste['total_waste_g']}g)")
    except Exception as e:
        print(f"Benchmark failed: {e}", file=sys.stderr)
        sys.exit(1)

import subprocess
from pathlib import Path

def slice_part(model_path: Path, output_dir: Path, infill: int = 20):
    cmd = [
        "/usr/local/bin/prusa-slicer",
        str(model_path),
        "--output", str(output_dir / model_path.name.replace('.stl', '.gcode')),
        "--infill-density", str(infill),
        "--gcode-comments"
    ]
    subprocess.run(cmd, check=True)

import requests

def push_waste_metric(waste_pct: float):
    # Push to Prometheus Pushgateway
    requests.post(
        "http://pushgateway:9091/metrics/job/waste_tracker",
        data=f"filament_waste_percentage {waste_pct}"
    )

from flask import Flask, request
import psycopg2

app = Flask(__name__)

def get_db():
    return psycopg2.connect("dbname=waste user=admin")

@app.route('/log-weight', methods=['POST'])
def log_weight():
    data = request.json
    conn = get_db()
    cur = conn.cursor()
    cur.execute("INSERT INTO weight_logs (part_id, estimated_g, actual_g) VALUES (%s, %s, %s)",
                (data['part_id'], data['estimated_g'], data['actual_g']))
    conn.commit()
    return {"status": "ok"}