Manufacturing Expert

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)

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: