manufacturing-expert

name	manufacturing-expert
version	1.0.0
description	Expert-level manufacturing systems, Industry 4.0, production optimization, quality control, and smart factory solutions
category	domains
tags	["manufacturing","industry40","production","quality","mes","plc"]
allowed-tools	["Read","Write","Edit"]

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: 
    machine_type: 
    status: MachineStatus
    current_order: []
    production_rate:   
    uptime_percentage: 
    last_maintenance: datetime
    next_maintenance: datetime
    oee:   


 :
    
    timestamp: datetime
    line_id: 
    produced_units: 
    defective_units: 
    downtime_minutes: 
    cycle_time_seconds: 
    efficiency_percentage: 

 :
    

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

     () -> WorkOrder:
        
        order_id = ._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=,
            started_at=,
            completed_at=
        )

        .work_orders[order_id] = order
         order

     () -> []:
        
        
        pending_orders = [
            order  order  .work_orders.values()
             order.status == OrderStatus.PENDING
        ]

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

        
        available_machines = [
            machine  machine  .machines.values()
             machine.status  [MachineStatus.IDLE, MachineStatus.RUNNING]
        ]

        schedule = []

         order  sorted_orders:
            
            best_machine = ._find_best_machine(order, available_machines)

             best_machine:
                
                production_time = order.quantity / best_machine.production_rate
                estimated_completion = datetime.now() + timedelta(hours=production_time)

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

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

         schedule

     () -> [Machine]:
        
          machines:
             

        
        scored_machines = []

         machine  machines:
            score = 

            
            score += machine.oee * 

            
             machine.status == MachineStatus.IDLE:
                score += 

            
            days_since_maintenance = (datetime.now() - machine.last_maintenance).days
            score += (,  - days_since_maintenance)

            scored_machines.append((score, machine))

        
        scored_machines.sort(reverse=, key= x: x[])
         scored_machines[][]

     () -> :
        
        order = .work_orders.get(order_id)
          order:
             {: }

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

        
         order.actual_quantity >= order.quantity:
            order.status = OrderStatus.COMPLETED
            order.completed_at = datetime.now()

            
            duration = order.completed_at - order.started_at
            yield_rate = ((order.actual_quantity - order.defect_quantity) /
                         order.actual_quantity * )

             {
                : order_id,
                : ,
                : duration.total_seconds() / ,
                : yield_rate,
                : order.actual_quantity,
                : order.defect_quantity
            }

         {
            : order_id,
            : ,
            : (order.actual_quantity / order.quantity) * 
        }

     () -> :
        
        machine = .machines.get(machine_id)
          machine:
             {: }

        

        
        planned_time = time_period_hours *   
        downtime = ._get_downtime(machine_id, time_period_hours)
        operating_time = planned_time - downtime
        availability = operating_time / planned_time

        
        actual_production = ._get_production_count(machine_id, time_period_hours)
        ideal_production = machine.production_rate * time_period_hours
        performance = actual_production / ideal_production  ideal_production >   

        
        defects = ._get_defect_count(machine_id, time_period_hours)
        quality = (actual_production - defects) / actual_production  actual_production >   

        oee = availability * performance * quality

         {
            : machine_id,
            : time_period_hours,
            : oee * ,  
            : availability * ,
            : performance * ,
            : quality * ,
            :   
        }

     () -> :
        
        
        
         

     () -> :
        
        
         

     () -> :
        
        
         

     () -> :
        
         uuid
         

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[]  value < control_limits[]:
                violations.append({
                    : ,
                    : i,
                    : value,
                    : 
                })

        
        sigma_2 = control_limits[] * 
        ucl_2 = control_limits[] + sigma_2
        lcl_2 = control_limits[] - sigma_2

         i  ((measurements) - ):
            window = measurements[i:i+]
            beyond_2sigma = (  v  window  v > ucl_2  v < lcl_2)
             beyond_2sigma >= :
                violations.append({
                    : ,
                    : i,
                    : 
                })

        
         i  ((measurements) - ):
            window = measurements[i:i+]
            all_above = (v > control_limits[]  v  window)
            all_below = (v < control_limits[]  v  window)

             all_above  all_below:
                violations.append({
                    : ,
                    : i,
                    : 
                })

         {
            : (violations) == ,
            : violations,
            : (violations)
        }

     () -> :
        
        mean = np.mean(measurements)
        std_dev = np.std(measurements, ddof=)

        
        cp = (upper_spec_limit - lower_spec_limit) / ( * std_dev)

        
        cpu = (upper_spec_limit - mean) / ( * std_dev)
        cpl = (mean - lower_spec_limit) / ( * std_dev)
        cpk = (cpu, cpl)

        
         cpk >= :
            capability = 
         cpk >= :
            capability = 
         cpk >= :
            capability = 
        :
            capability = 

         {
            : cp,
            : cpk,
            : cpu,
            : cpl,
            : capability,
            : cpk *   cpk >   
        }

     () -> :
        
        
        data = measurements.reshape(n_parts, n_operators, n_trials)

        
        part_means = data.mean(axis=(, ))
        operator_means = data.mean(axis=(, ))
        overall_mean = data.mean()

        
        part_variance = np.var(part_means, ddof=)

        
        within_operator_variance = np.mean([
            np.var(data[:, op, :], ddof=)
             op  (n_operators)
        ])

        
        operator_variance = np.var(operator_means, ddof=)

        
        total_variance = np.var(data, ddof=)

        
        gage_rr = within_operator_variance + operator_variance
        gage_rr_percentage = (gage_rr / total_variance) * 

        
         gage_rr_percentage < :
            assessment = 
         gage_rr_percentage < :
            assessment = 
        :
            assessment = 

         {
            : gage_rr_percentage,
            : (within_operator_variance / total_variance) * ,
            : (operator_variance / total_variance) * ,
            : (part_variance / total_variance) * ,
            : 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[]

        
        X_scaled = .scaler.fit_transform(X)

        
        .model.fit(X_scaled, y)
        .trained = 

     () -> :
        
          .trained:
             {: }

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

        
        failure_probability = .model.predict_proba(features_scaled)[][]

        
        rul_hours = ._estimate_rul(failure_probability)

        
         failure_probability > :
            recommendation = 
            priority = 
         failure_probability > :
            recommendation = 
            priority = 
         failure_probability > :
            recommendation = 
            priority = 
        :
            recommendation = 
            priority = 

         {
            : failure_probability,
            : rul_hours,
            : recommendation,
            : priority,
            : datetime.now().isoformat()
        }

     () -> :
        
        
        
         failure_probability < :
               
         failure_probability < :
               
         failure_probability < :
               
         failure_probability < :
                
        :
                

     () -> []:
        
        failure_modes = []

        
         sensor_data[] > :
            failure_modes.append({
                : ,
                : ,
                : 
            })

        
         sensor_data[] > :
            failure_modes.append({
                : ,
                : ,
                : 
            })

        
         sensor_data[] > sensor_data.get(, ) * :
            failure_modes.append({
                : ,
                : ,
                : 
            })

         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) -> :
        
        
        simulated_state = .virtual_state.copy()

        
         param, value  scenario_params.items():
             param  simulated_state[]:
                simulated_state[][param] = value

        
        simulated_state[] = ._calculate_metrics(
            simulated_state[]
        )

         {
            : scenario_params,
            : simulated_state,
            : ._analyze_impact(simulated_state)
        }

     () -> :
        
        
        
        best_params = {}
        best_score = 

         {
            : optimization_goal,
            : best_params,
            : best_score
        }

     () -> :
        
           

     () -> :
        
           

     () -> :
        
         sensor_data.get(, ) * sensor_data.get(, )

     () -> :
        
         {: }

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/