What Is Batch Production in 3D Printing and Why It Matters

import requests
import time
from typing import List, Dict, Optional
from dataclasses import dataclass
import logging

# Configure logging for batch job audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("batch_scheduler.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

@dataclass
class OctoPrintInstance:
    """Represents a single OctoPrint-connected 3D printer"""
    url: str  # e.g., http://192.168.1.100:5000
    api_key: str
    max_bed_temp: int = 250  # Default FDM max bed temp
    max_extruder_temp: int = 300  # Default FDM max extruder temp

class BatchSchedulerError(Exception):
    """Custom exception for batch scheduling failures"""
    pass

class OctoPrintBatchScheduler:
    def __init__(self, printers: List[OctoPrintInstance], max_retries: int = 3):
        self.printers = printers
        self.max_retries = max_retries
        self.session = requests.Session()
        # Set default headers for all OctoPrint API requests
        for printer in self.printers:
            self.session.headers.update({"X-Api-Key": printer.api_key})

    def _make_api_request(self, printer: OctoPrintInstance, method: str, endpoint: str, **kwargs) -> Dict:
        """Handle rate limits and connection errors for OctoPrint API calls"""
        url = f"{printer.url}/api/{endpoint.lstrip('/')}"
        for attempt in range(self.max_retries):
            try:
                response = self.session.request(method, url, **kwargs)
                response.raise_for_status()
                return response.json()
            except requests.exceptions.ConnectionError as e:
                logger.warning(f"Connection failed to {printer.url} (attempt {attempt+1}/{self.max_retries}): {e}")
                time.sleep(2 ** attempt)  # Exponential backoff
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 429:
                    retry_after = int(e.response.headers.get("Retry-After", 5))
                    logger.warning(f"Rate limited by {printer.url}, retrying after {retry_after}s")
                    time.sleep(retry_after)
                else:
                    logger.error(f"HTTP error from {printer.url}: {e}")
                    raise BatchSchedulerError(f"API error: {e}") from e
        raise BatchSchedulerError(f"Max retries exceeded for {printer.url}")

    def get_printer_status(self, printer: OctoPrintInstance) -> Dict:
        """Fetch current operational status of a printer"""
        return self._make_api_request(printer, "GET", "job")

    def validate_stl(self, printer: OctoPrintInstance, stl_path: str) -> bool:
        """Check if an STL is compatible with the target printer's build volume"""
        # In production, this would use Trimesh to check mesh dimensions
        # For brevity, we mock the validation here
        logger.info(f"Validating {stl_path} for {printer.url}")
        return True  # Replace with actual Trimesh validation

    def schedule_batch(self, stl_paths: List[str], material: str, layer_height: float = 0.2) -> Dict:
        """Schedule a batch of STL files across available printers"""
        available_printers = []
        for printer in self.printers:
            status = self.get_printer_status(printer)
            if status.get("state", {}).get("text") == "Operational":
                available_printers.append(printer)

        if not available_printers:
            raise BatchSchedulerError("No available printers for batch job")

        batch_id = f"batch_{int(time.time())}"
        logger.info(f"Scheduling batch {batch_id} with {len(stl_paths)} files across {len(available_printers)} printers")

        results = {"batch_id": batch_id, "scheduled": [], "failed": []}
        for i, stl_path in enumerate(stl_paths):
            printer = available_printers[i % len(available_printers)]  # Round-robin scheduling
            if not self.validate_stl(printer, stl_path):
                results["failed"].append({"stl": stl_path, "reason": "Validation failed"})
                continue
            # Upload STL to printer (simplified for example)
            # In production, use OctoPrint's file upload API
            results["scheduled"].append({"stl": stl_path, "printer": printer.url})

        logger.info(f"Batch {batch_id} scheduled: {len(results['scheduled'])} success, {len(results['failed'])} failed")
        return results

# Example usage
if __name__ == "__main__":
    printers = [
        OctoPrintInstance(url="http://192.168.1.100:5000", api_key="API_KEY_1"),
        OctoPrintInstance(url="http://192.168.1.101:5000", api_key="API_KEY_2"),
    ]
    scheduler = OctoPrintBatchScheduler(printers)
    try:
        batch_result = scheduler.schedule_batch(
            stl_paths=["/path/to/part1.stl", "/path/to/part2.stl"],
            material="PLA"
        )
        print(f"Scheduled batch: {batch_result}")
    except BatchSchedulerError as e:
        logger.error(f"Batch scheduling failed: {e}")

const { PrismaClient } = require('@prisma/client');
const prisma = new PrismaClient();

/**
 * Batch cost calculator for 3D printing workflows
 * Compares batch vs single-unit printing costs using real-time material and machine rates
 */
class BatchCostCalculator {
  constructor() {
    this.MACHINE_HOURLY_RATES = {
      'FDM': 2.50,    // USD per hour for FDM printers
      'SLS': 18.00,   // USD per hour for SLS printers
      'RESIN': 6.50   // USD per hour for resin printers
    };
    this.MATERIAL_COST_PER_KG = {
      'PLA': 22.00,
      'ABS': 28.00,
      'Nylon12': 85.00,
      'ResinStandard': 45.00
    };
  }

  /**
   * Calculate material usage for a single part (simplified, uses STL volume in production)
   * @param {string} material - Material type
   * @param {number} printWeightG - Weight of single print in grams
   * @returns {number} Material cost for single part
   */
  calculateSingleMaterialCost(material, printWeightG) {
    if (!this.MATERIAL_COST_PER_KG[material]) {
      throw new Error(`Unsupported material: ${material}`);
    }
    if (printWeightG <= 0) {
      throw new Error('Print weight must be positive');
    }
    return (printWeightG / 1000) * this.MATERIAL_COST_PER_KG[material];
  }

  /**
   * Calculate machine time cost for a single part
   * @param {string} machineType - FDM, SLS, RESIN
   * @param {number} printTimeHrs - Print time per part in hours
   * @returns {number} Machine cost for single part
   */
  calculateSingleMachineCost(machineType, printTimeHrs) {
    if (!this.MACHINE_HOURLY_RATES[machineType]) {
      throw new Error(`Unsupported machine type: ${machineType}`);
    }
    if (printTimeHrs <= 0) {
      throw new Error('Print time must be positive');
    }
    return printTimeHrs * this.MACHINE_HOURLY_RATES[machineType];
  }

  /**
   * Calculate total batch cost with failure rate and setup overhead
   * @param {Object} params - Batch parameters
   * @param {number} params.batchSize - Number of units in batch
   * @param {number} params.singlePartCost - Pre-calculated cost per part
   * @param {number} params.failureRate - Percentage of failed builds (0-1)
   * @param {number} params.setupTimeHrs - Batch setup time in hours
   * @returns {number} Total batch cost
   */
  calculateBatchCost(params) {
    const { batchSize, singlePartCost, failureRate, setupTimeHrs } = params;
    if (batchSize < 1) throw new Error('Batch size must be at least 1');
    if (failureRate < 0 || failureRate > 1) throw new Error('Failure rate must be between 0 and 1');

    // Account for failed builds: need to print (batchSize / (1 - failureRate)) parts
    const totalPartsToPrint = batchSize / (1 - failureRate);
    const materialAndMachineCost = totalPartsToPrint * singlePartCost;
    const setupCost = setupTimeHrs * this.MACHINE_HOURLY_RATES['FDM']; // Assume setup uses FDM rate

    return materialAndMachineCost + setupCost;
  }

  /**
   * Compare batch vs single-unit printing costs
   * @param {Object} batchParams - Batch parameters
   * @param {Object} singleParams - Single print parameters
   * @returns {Object} Cost comparison
   */
  async compareBatchVsSingle(batchParams, singleParams) {
    try {
      const singlePartCost = singleParams.materialCost + singleParams.machineCost;
      const batchCost = this.calculateBatchCost({
        ...batchParams,
        singlePartCost
      });
      const singleTotalCost = singleParams.materialCost + singleParams.machineCost;
      const totalSingleCost = singleTotalCost * batchParams.batchSize;

      // Log comparison to database for audit
      await prisma.costComparison.create({
        data: {
          batchSize: batchParams.batchSize,
          batchCost,
          singleCost: totalSingleCost,
          savings: totalSingleCost - batchCost,
          timestamp: new Date()
        }
      });

      return {
        batchCost: parseFloat(batchCost.toFixed(2)),
        singleCost: parseFloat(totalSingleCost.toFixed(2)),
        savings: parseFloat((totalSingleCost - batchCost).toFixed(2)),
        savingsPercentage: parseFloat((((totalSingleCost - batchCost) / totalSingleCost) * 100).toFixed(1))
      };
    } catch (error) {
      console.error('Cost comparison failed:', error.message);
      throw error;
    }
  }
}

// Example usage
async function main() {
  const calculator = new BatchCostCalculator();
  try {
    const result = await calculator.compareBatchVsSingle(
      {
        batchSize: 100,
        failureRate: 0.023, // 2.3% failure rate for batched FDM
        setupTimeHrs: 0.5
      },
      {
        materialCost: calculator.calculateSingleMaterialCost('PLA', 120), // 120g per part
        machineCost: calculator.calculateSingleMachineCost('FDM', 3.1) // 3.1h per part
      }
    );
    console.log('Cost Comparison:', result);
  } catch (error) {
    console.error('Error:', error.message);
  } finally {
    await prisma.$disconnect();
  }
}

main();

import trimesh
import os
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass
import sys
import math

# Trimesh is available at https://github.com/mikedh/trimesh
# Install via: pip install trimesh[easy]

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ValidationConfig:
    """Configuration for STL validation in batch workflows"""
    max_build_volume_mm: tuple = (250, 250, 250)  # X, Y, Z in mm (standard FDM bed)
    max_overhang_angle_deg: int = 45  # Maximum overhang without support
    require_manifold: bool = True
    max_file_size_mb: int = 50

class BatchSTLValidatorError(Exception):
    """Custom exception for validation failures"""
    pass

class BatchSTLValidator:
    def __init__(self, config: Optional[ValidationConfig] = None):
        self.config = config or ValidationConfig()
        self.validation_results = []

    def _check_file_size(self, stl_path: str) -> bool:
        """Validate STL file size is within limits"""
        try:
            file_size_mb = os.path.getsize(stl_path) / (1024 * 1024)
            if file_size_mb > self.config.max_file_size_mb:
                logger.error(f"File {stl_path} too large: {file_size_mb:.2f}MB > {self.config.max_file_size_mb}MB")
                return False
            return True
        except OSError as e:
            logger.error(f"Failed to check file size for {stl_path}: {e}")
            return False

    def _check_manifold(self, mesh: trimesh.Trimesh) -> bool:
        """Check if the mesh is manifold (watertight)"""
        if not self.config.require_manifold:
            return True
        if not mesh.is_manifold:
            logger.error("Mesh is non-manifold (has holes or disconnected faces)")
            return False
        if not mesh.is_watertight:
            logger.error("Mesh is not watertight")
            return False
        return True

    def _check_build_volume(self, mesh: trimesh.Trimesh) -> bool:
        """Check if mesh fits within configured build volume"""
        mesh_bounds = mesh.bounds  # [[min_x, min_y, min_z], [max_x, max_y, max_z]]
        mesh_dimensions = mesh_bounds[1] - mesh_bounds[0]
        for dim, max_dim in zip(mesh_dimensions, self.config.max_build_volume_mm):
            if dim > max_dim:
                logger.error(f"Mesh dimension {dim:.2f}mm exceeds max build volume {max_dim}mm")
                return False
        return True

    def _check_overhangs(self, mesh: trimesh.Trimesh) -> bool:
        """Simplified overhang check (production would use slice simulation)"""
        # For this example, we check if any face normal has a z-component < cos(90 - max_overhang_angle)
        # This is a rough approximation, not a replacement for slice-based checks
        max_z_normal = math.cos(math.radians(90 - self.config.max_overhang_angle_deg))
        face_normals = mesh.face_normals
        for normal in face_normals:
            if normal[2] < max_z_normal and normal[2] < 0:  # Overhang pointing downward
                logger.warning(f"Potential overhang detected with angle > {self.config.max_overhang_angle_deg}deg")
                return False
        return True

    def validate_stl(self, stl_path: str) -> Dict:
        """Run all validation checks on a single STL file"""
        result = {
            "stl_path": stl_path,
            "valid": False,
            "errors": [],
            "warnings": []
        }
        # Check file size first
        if not self._check_file_size(stl_path):
            result["errors"].append("File size exceeds limit")
            return result
        # Load mesh
        try:
            mesh = trimesh.load(stl_path)
            if not isinstance(mesh, trimesh.Trimesh):
                result["errors"].append("File is not a valid single STL mesh")
                return result
        except Exception as e:
            result["errors"].append(f"Failed to load STL: {str(e)}")
            return result
        # Run validation checks
        if self.config.require_manifold and not self._check_manifold(mesh):
            result["errors"].append("Mesh is non-manifold or not watertight")
        if not self._check_build_volume(mesh):
            result["errors"].append("Mesh exceeds build volume")
        if not self._check_overhangs(mesh):
            result["errors"].append(f"Mesh has overhangs > {self.config.max_overhang_angle_deg}deg")
        # Mark as valid if no errors
        if not result["errors"]:
            result["valid"] = True
            logger.info(f"STL {stl_path} passed all validation checks")
        else:
            logger.error(f"STL {stl_path} failed validation: {result['errors']}")
        return result

    def validate_batch(self, stl_paths: List[str]) -> List[Dict]:
        """Validate a batch of STL files"""
        self.validation_results = []
        for stl_path in stl_paths:
            if not os.path.exists(stl_path):
                self.validation_results.append({
                    "stl_path": stl_path,
                    "valid": False,
                    "errors": ["File not found"]
                })
                continue
            result = self.validate_stl(stl_path)
            self.validation_results.append(result)
        return self.validation_results

# Example usage
if __name__ == "__main__":
    validator = BatchSTLValidator()
    test_stls = ["/path/to/part1.stl", "/path/to/part2.stl", "/path/to/nonexistent.stl"]
    results = validator.validate_batch(test_stls)
    for res in results:
        print(f"STL: {res['stl_path']}, Valid: {res['valid']}, Errors: {res['errors']}")

import trimesh
def validate_stl(stl_path: str) -> bool:
    try:
        mesh = trimesh.load(stl_path)
        return mesh.is_manifold and mesh.is_watertight
    except Exception:
        return False

import redis
import hashlib
import uuid

r = redis.Redis(host='localhost', port=6379, db=0)

def submit_batch(stl_paths: list, material: str) -> str:
    batch_hash = hashlib.sha256(f"{stl_paths}{material}".encode()).hexdigest()
    existing_job = r.get(batch_hash)
    if existing_job:
        return existing_job.decode()
    job_id = str(uuid.uuid7())
    r.setex(batch_hash, 86400, job_id)  # 24h TTL
    return job_id

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider

meter = metrics.get_meter("batch_production")
failed_builds = meter.create_counter("batch.failed_builds", description="Number of failed builds per batch")

def record_failed_build(batch_id: str, material: str):
    failed_builds.add(1, {"batch_id": batch_id, "material": material})