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/