🇺🇸English

Manufacturing Expert

Expert guidance for manufacturing systems, Industry 4.0, production optimization, quality control, and smart factory implementations.

Core Concepts

Manufacturing Systems

• Manufacturing Execution Systems (MES)
• Enterprise Resource Planning (ERP)
• Computer-Aided Manufacturing (CAM)
• Programmable Logic Controllers (PLC)
• Industrial Internet of Things (IIoT)
• Supply Chain Management (SCM)
• Warehouse Management Systems (WMS)

Industry 4.0

• Smart factories
• Digital twins
• Predictive maintenance
• Autonomous robotics
• Augmented reality for operations
• Edge computing
• Cyber-physical systems

Standards and Protocols

• OPC UA (Open Platform Communications)
• ISA-95 (Enterprise-Control System Integration)
• MTConnect (manufacturing data exchange)
• MQTT for IIoT
• EtherCAT (real-time Ethernet)
• PROFINET
• ISO 9001 (Quality Management)

Manufacturing Execution System (MES)

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    ON_HOLD = "on_hold"
    CANCELLED = "cancelled"

class MachineStatus(Enum):
    IDLE = "idle"
    RUNNING = "running"
    MAINTENANCE = "maintenance"
    ERROR = "error"
    OFFLINE = "offline"

@dataclass
class WorkOrder:
    """Manufacturing work order"""
    order_id: str
    product_id: str
    quantity: int
    priority: int  # 1 (highest) to 5 (lowest)
    due_date: datetime
    status: OrderStatus
    assigned_line: Optional[str]
    started_at: Optional[datetime]
    completed_at: Optional[datetime]
    actual_quantity: int = 0
    defect_quantity: int = 0

@dataclass
class Machine:
    """Production machine/equipment"""
    machine_id: str
    machine_type: str
    status: MachineStatus
    current_order: Optional[str]
    production_rate: float  # units per hour
    uptime_percentage: float
    last_maintenance: datetime
    next_maintenance: datetime
    oee: float  # Overall Equipment Effectiveness

@dataclass
class ProductionMetrics:
    """Real-time production metrics"""
    timestamp: datetime
    line_id: str
    produced_units: int
    defective_units: int
    downtime_minutes: int
    cycle_time_seconds: float
    efficiency_percentage: float

class ManufacturingExecutionSystem:
    """MES for production management"""

    def __init__(self):
        self.work_orders = {}
        self.machines = {}
        self.production_data = []

    def create_work_order(self,
                         product_id: str,
                         quantity: int,
                         due_date: datetime,
                         priority: int = 3) -> WorkOrder:
        """Create new production work order"""
        order_id = self._generate_order_id()

        order = WorkOrder(
            order_id=order_id,
            product_id=product_id,
            quantity=quantity,
            priority=priority,
            due_date=due_date,
            status=OrderStatus.PENDING,
            assigned_line=None,
            started_at=None,
            completed_at=None
        )

        self.work_orders[order_id] = order
        return order

    def schedule_production(self) -> List[dict]:
        """Schedule work orders to production lines"""
        # Get pending orders sorted by priority and due date
        pending_orders = [
            order for order in self.work_orders.values()
            if order.status == OrderStatus.PENDING
        ]

        sorted_orders = sorted(
            pending_orders,
            key=lambda x: (x.priority, x.due_date)
        )

        # Get available machines
        available_machines = [
            machine for machine in self.machines.values()
            if machine.status in [MachineStatus.IDLE, MachineStatus.RUNNING]
        ]

        schedule = []

        for order in sorted_orders:
            # Find best machine for this order
            best_machine = self._find_best_machine(order, available_machines)

            if best_machine:
                # Calculate estimated completion time
                production_time = order.quantity / best_machine.production_rate
                estimated_completion = datetime.now() + timedelta(hours=production_time)

                schedule.append({
                    'order_id': order.order_id,
                    'machine_id': best_machine.machine_id,
                    'estimated_start': datetime.now(),
                    'estimated_completion': estimated_completion,
                    'estimated_duration_hours': production_time
                })

                # Update order
                order.assigned_line = best_machine.machine_id
                order.status = OrderStatus.IN_PROGRESS

        return schedule

    def _find_best_machine(self, order: WorkOrder, machines: List[Machine]) -> Optional[Machine]:
        """Find optimal machine for work order"""
        if not machines:
            return None

        # Score machines based on multiple factors
        scored_machines = []

        for machine in machines:
            score = 0

            # Prefer machines with higher OEE
            score += machine.oee * 50

            # Prefer machines that are idle
            if machine.status == MachineStatus.IDLE:
                score += 30

            # Prefer machines with recent maintenance
            days_since_maintenance = (datetime.now() - machine.last_maintenance).days
            score += max(0, 20 - days_since_maintenance)

            scored_machines.append((score, machine))

        # Return highest scoring machine
        scored_machines.sort(reverse=True, key=lambda x: x[0])
        return scored_machines[0][1]

    def record_production(self,
                         order_id: str,
                         produced: int,
                         defective: int = 0) -> dict:
        """Record production output"""
        order = self.work_orders.get(order_id)
        if not order:
            return {'error': 'Order not found'}

        order.actual_quantity += produced
        order.defect_quantity += defective

        # Check if order is complete
        if order.actual_quantity >= order.quantity:
            order.status = OrderStatus.COMPLETED
            order.completed_at = datetime.now()

            # Calculate metrics
            duration = order.completed_at - order.started_at
            yield_rate = ((order.actual_quantity - order.defect_quantity) /
                         order.actual_quantity * 100)

            return {
                'order_id': order_id,
                'status': 'completed',
                'duration_hours': duration.total_seconds() / 3600,
                'yield_rate': yield_rate,
                'total_produced': order.actual_quantity,
                'total_defective': order.defect_quantity
            }

        return {
            'order_id': order_id,
            'status': 'in_progress',
            'progress_percentage': (order.actual_quantity / order.quantity) * 100
        }

    def calculate_oee(self,
                     machine_id: str,
                     time_period_hours: int = 24) -> dict:
        """Calculate Overall Equipment Effectiveness"""
        machine = self.machines.get(machine_id)
        if not machine:
            return {'error': 'Machine not found'}

        # OEE = Availability × Performance × Quality

        # Availability: (Operating Time / Planned Production Time)
        planned_time = time_period_hours * 60  # minutes
        downtime = self._get_downtime(machine_id, time_period_hours)
        operating_time = planned_time - downtime
        availability = operating_time / planned_time

        # Performance: (Actual Production / Ideal Production)
        actual_production = self._get_production_count(machine_id, time_period_hours)
        ideal_production = machine.production_rate * time_period_hours
        performance = actual_production / ideal_production if ideal_production > 0 else 0

        # Quality: (Good Units / Total Units)
        defects = self._get_defect_count(machine_id, time_period_hours)
        quality = (actual_production - defects) / actual_production if actual_production > 0 else 0

        oee = availability * performance * quality

        return {
            'machine_id': machine_id,
            'period_hours': time_period_hours,
            'oee': oee * 100,  # Percentage
            'availability': availability * 100,
            'performance': performance * 100,
            'quality': quality * 100,
            'world_class_oee': 85.0  # Benchmark
        }

    def _get_downtime(self, machine_id: str, hours: int) -> float:
        """Get machine downtime in minutes"""
        # Query production data for downtime
        # Implementation would aggregate from time-series data
        return 0.0

    def _get_production_count(self, machine_id: str, hours: int) -> int:
        """Get production count for machine"""
        # Implementation would query production records
        return 0

    def _get_defect_count(self, machine_id: str, hours: int) -> int:
        """Get defect count for machine"""
        # Implementation would query quality records
        return 0

    def _generate_order_id(self) -> str:
        """Generate unique order ID"""
        import uuid
        return f"WO-{uuid.uuid4().hex[:8].upper()}"

Quality Control System

from scipy import stats
import numpy as np

class StatisticalProcessControl:
    """Statistical Process Control (SPC) for quality management"""

    def __init__(self):
        self.measurement_history = {}

    def calculate_control_limits(self,
                                 measurements: List[float],
                                 sigma_level: float = 3.0) -> dict:
        """Calculate control limits for control charts"""
        mean = np.mean(measurements)
        std_dev = np.std(measurements, ddof=1)

        ucl = mean + (sigma_level * std_dev)  # Upper Control Limit
        lcl = mean - (sigma_level * std_dev)  # Lower Control Limit

        return {
            'mean': mean,
            'std_dev': std_dev,
            'ucl': ucl,
            'lcl': lcl,
            'sigma_level': sigma_level
        }

    def detect_out_of_control(self,
                             measurements: List[float],
                             control_limits: dict) -> dict:
        """Detect out-of-control conditions"""
        violations = []

        # Rule 1: Point beyond control limits
        for i, value in enumerate(measurements):
            if value > control_limits['ucl'] or value < control_limits['lcl']:
                violations.append({
                    'rule': 'beyond_limits',
                    'index': i,
                    'value': value,
                    'severity': 'critical'
                })

        # Rule 2: 2 out of 3 consecutive points beyond 2σ
        sigma_2 = control_limits['std_dev'] * 2
        ucl_2 = control_limits['mean'] + sigma_2
        lcl_2 = control_limits['mean'] - sigma_2

        for i in range(len(measurements) - 2):
            window = measurements[i:i+3]
            beyond_2sigma = sum(1 for v in window if v > ucl_2 or v < lcl_2)
            if beyond_2sigma >= 2:
                violations.append({
                    'rule': '2_of_3_beyond_2sigma',
                    'index': i,
                    'severity': 'warning'
                })

        # Rule 3: 9 consecutive points on same side of mean
        for i in range(len(measurements) - 8):
            window = measurements[i:i+9]
            all_above = all(v > control_limits['mean'] for v in window)
            all_below = all(v < control_limits['mean'] for v in window)

            if all_above or all_below:
                violations.append({
                    'rule': '9_consecutive_same_side',
                    'index': i,
                    'severity': 'warning'
                })

        return {
            'in_control': len(violations) == 0,
            'violations': violations,
            'total_violations': len(violations)
        }

    def calculate_cpk(self,
                     measurements: List[float],
                     lower_spec_limit: float,
                     upper_spec_limit: float) -> dict:
        """Calculate Process Capability Index (Cpk)"""
        mean = np.mean(measurements)
        std_dev = np.std(measurements, ddof=1)

        # Cp: Process Capability
        cp = (upper_spec_limit - lower_spec_limit) / (6 * std_dev)

        # Cpk: Process Capability Index (accounts for centering)
        cpu = (upper_spec_limit - mean) / (3 * std_dev)
        cpl = (mean - lower_spec_limit) / (3 * std_dev)
        cpk = min(cpu, cpl)

        # Interpret Cpk
        if cpk >= 2.0:
            capability = "Excellent"
        elif cpk >= 1.33:
            capability = "Adequate"
        elif cpk >= 1.0:
            capability = "Marginal"
        else:
            capability = "Inadequate"

        return {
            'cp': cp,
            'cpk': cpk,
            'cpu': cpu,
            'cpl': cpl,
            'capability': capability,
            'sigma_level': cpk * 3 if cpk > 0 else 0
        }

    def perform_gage_rr(self,
                       measurements: np.ndarray,
                       n_parts: int,
                       n_operators: int,
                       n_trials: int) -> dict:
        """Perform Gage Repeatability and Reproducibility study"""
        # Reshape data: (parts × operators × trials)
        data = measurements.reshape(n_parts, n_operators, n_trials)

        # Calculate variance components
        part_means = data.mean(axis=(1, 2))
        operator_means = data.mean(axis=(0, 2))
        overall_mean = data.mean()

        # Part variation
        part_variance = np.var(part_means, ddof=1)

        # Repeatability (equipment variation)
        within_operator_variance = np.mean([
            np.var(data[:, op, :], ddof=1)
            for op in range(n_operators)
        ])

        # Reproducibility (operator variation)
        operator_variance = np.var(operator_means, ddof=1)

        # Total variation
        total_variance = np.var(data, ddof=1)

        # Gage R&R
        gage_rr = within_operator_variance + operator_variance
        gage_rr_percentage = (gage_rr / total_variance) * 100

        # Interpretation
        if gage_rr_percentage < 10:
            assessment = "Acceptable"
        elif gage_rr_percentage < 30:
            assessment = "Marginal"
        else:
            assessment = "Unacceptable"

        return {
            'gage_rr_percentage': gage_rr_percentage,
            'repeatability_percentage': (within_operator_variance / total_variance) * 100,
            'reproducibility_percentage': (operator_variance / total_variance) * 100,
            'part_variation_percentage': (part_variance / total_variance) * 100,
            'assessment': assessment
        }

Predictive Maintenance

from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
import pandas as pd

class PredictiveMaintenanceSystem:
    """Predictive maintenance using machine learning"""

    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.scaler = StandardScaler()
        self.trained = False

    def extract_features(self, sensor_data: dict) -> np.ndarray:
        """Extract features from sensor data"""
        features = [
            sensor_data['vibration_rms'],
            sensor_data['vibration_peak'],
            sensor_data['temperature_c'],
            sensor_data['current_a'],
            sensor_data['pressure_bar'],
            sensor_data['speed_rpm'],
            sensor_data['operating_hours'],
            sensor_data['cycles_completed']
        ]

        return np.array(features).reshape(1, -1)

    def train_model(self, historical_data: pd.DataFrame):
        """Train predictive maintenance model"""
        # Extract features and labels
        X = historical_data.drop(['machine_id', 'timestamp', 'failure'], axis=1)
        y = historical_data['failure']

        # Scale features
        X_scaled = self.scaler.fit_transform(X)

        # Train model
        self.model.fit(X_scaled, y)
        self.trained = True

    def predict_failure(self, sensor_data: dict) -> dict:
        """Predict equipment failure probability"""
        if not self.trained:
            return {'error': 'Model not trained'}

        features = self.extract_features(sensor_data)
        features_scaled = self.scaler.transform(features)

        # Get failure probability
        failure_probability = self.model.predict_proba(features_scaled)[0][1]

        # Calculate remaining useful life (simplified)
        rul_hours = self._estimate_rul(failure_probability)

        # Generate recommendation
        if failure_probability > 0.8:
            recommendation = "Schedule immediate maintenance"
            priority = "critical"
        elif failure_probability > 0.5:
            recommendation = "Schedule maintenance within 1 week"
            priority = "high"
        elif failure_probability > 0.3:
            recommendation = "Monitor closely, schedule maintenance"
            priority = "medium"
        else:
            recommendation = "Continue normal operation"
            priority = "low"

        return {
            'failure_probability': failure_probability,
            'remaining_useful_life_hours': rul_hours,
            'recommendation': recommendation,
            'priority': priority,
            'timestamp': datetime.now().isoformat()
        }

    def _estimate_rul(self, failure_probability: float) -> float:
        """Estimate Remaining Useful Life"""
        # Simplified RUL estimation
        # In production, use more sophisticated models (LSTM, CNN)
        if failure_probability < 0.1:
            return 720.0  # 30 days
        elif failure_probability < 0.3:
            return 360.0  # 15 days
        elif failure_probability < 0.5:
            return 168.0  # 7 days
        elif failure_probability < 0.8:
            return 48.0   # 2 days
        else:
            return 12.0   # 12 hours

    def analyze_failure_modes(self, sensor_data: dict) -> List[dict]:
        """Identify potential failure modes"""
        failure_modes = []

        # Check for bearing failure indicators
        if sensor_data['vibration_rms'] > 10.0:
            failure_modes.append({
                'mode': 'bearing_failure',
                'indicator': 'high_vibration',
                'severity': 'high'
            })

        # Check for overheating
        if sensor_data['temperature_c'] > 80.0:
            failure_modes.append({
                'mode': 'thermal_failure',
                'indicator': 'high_temperature',
                'severity': 'high'
            })

        # Check for electrical issues
        if sensor_data['current_a'] > sensor_data.get('rated_current', 100) * 1.2:
            failure_modes.append({
                'mode': 'electrical_failure',
                'indicator': 'overcurrent',
                'severity': 'medium'
            })

        return failure_modes

Digital Twin Implementation

class DigitalTwin:
    """Digital twin for manufacturing equipment"""

    def __init__(self, physical_asset_id: str):
        self.asset_id = physical_asset_id
        self.virtual_state = {}
        self.historical_data = []
        self.simulation_model = None

    def sync_with_physical(self, sensor_data: dict):
        """Synchronize digital twin with physical asset"""
        self.virtual_state.update({
            'timestamp': datetime.now(),
            'sensors': sensor_data,
            'calculated_metrics': self._calculate_metrics(sensor_data)
        })

        self.historical_data.append(self.virtual_state.copy())

    def _calculate_metrics(self, sensor_data: dict) -> dict:
        """Calculate derived metrics from sensor data"""
        return {
            'efficiency': self._calculate_efficiency(sensor_data),
            'health_score': self._calculate_health_score(sensor_data),
            'energy_consumption': self._calculate_energy(sensor_data)
        }

    def simulate_scenario(self, scenario_params: dict) -> dict:
        """Simulate what-if scenarios"""
        # Simulate different operating conditions
        simulated_state = self.virtual_state.copy()

        # Apply scenario parameters
        for param, value in scenario_params.items():
            if param in simulated_state['sensors']:
                simulated_state['sensors'][param] = value

        # Recalculate metrics
        simulated_state['calculated_metrics'] = self._calculate_metrics(
            simulated_state['sensors']
        )

        return {
            'scenario': scenario_params,
            'predicted_state': simulated_state,
            'impact_analysis': self._analyze_impact(simulated_state)
        }

    def optimize_parameters(self, optimization_goal: str) -> dict:
        """Optimize operating parameters"""
        # Use digital twin to find optimal settings
        # This would use optimization algorithms
        best_params = {}
        best_score = 0

        return {
            'optimization_goal': optimization_goal,
            'recommended_parameters': best_params,
            'expected_improvement': best_score
        }

    def _calculate_efficiency(self, sensor_data: dict) -> float:
        """Calculate equipment efficiency"""
        return 85.0  # Simplified

    def _calculate_health_score(self, sensor_data: dict) -> float:
        """Calculate equipment health score (0-100)"""
        return 90.0  # Simplified

    def _calculate_energy(self, sensor_data: dict) -> float:
        """Calculate energy consumption"""
        return sensor_data.get('current_a', 0) * sensor_data.get('voltage_v', 0)

    def _analyze_impact(self, state: dict) -> dict:
        """Analyze impact of state change"""
        return {'impact': 'positive'}

Best Practices

Production Management

• Implement real-time monitoring dashboards
• Use automated scheduling algorithms
• Maintain digital work instructions
• Track genealogy and traceability
• Implement kanban or just-in-time systems
• Monitor key performance indicators (KPIs)

Quality Management

• Implement Statistical Process Control (SPC)
• Use automated inspection systems
• Maintain calibration records
• Conduct regular gage R&R studies
• Implement root cause analysis (RCA)
• Track first pass yield (FPY)

Maintenance Strategy

• Implement predictive maintenance
• Maintain spare parts inventory
• Use CMMS (Computerized Maintenance Management System)
• Schedule preventive maintenance
• Track Mean Time Between Failures (MTBF)
• Implement condition-based monitoring

Data Management

• Use time-series databases for sensor data
• Implement data historians
• Maintain data integrity and quality
• Enable real-time analytics
• Support machine learning workloads
• Archive historical data appropriately

Anti-Patterns

❌ Manual data entry for production records ❌ No preventive maintenance program ❌ Ignoring quality control data ❌ Siloed systems (no integration) ❌ No standard operating procedures ❌ Inadequate operator training ❌ No backup systems for critical equipment ❌ Poor inventory management

Resources

• ISA-95 Standard: https://www.isa.org/standards/isa95
• OPC UA: https://opcfoundation.org/
• MTConnect: https://www.mtconnect.org/
• Industry 4.0: https://www.plattform-i40.de/
• MESA International: https://www.mesa.org/
• SME (Society of Manufacturing Engineers): https://www.sme.org/
• Six Sigma: https://www.isixsigma.com/

Weekly Installs

141

Repository

personamanagmen…ayer/pcl

GitHub Stars

11

First Seen

Jan 24, 2026

Security Audits

Gen Agent Trust HubPassSocketPassSnykPass

Installed on

opencode126

gemini-cli125

codex124

cursor118

github-copilot114

amp108