制造业数据分析师面试题库：MES/ERP/SCADA系统解析与OEE建模

ERP (计划层)
    ↓ 生产订单、物料需求
MES (执行层)
    ↓ 生产指令、质量标准
SCADA (控制层)
    ↑ 实时数据、设备状态
    ↑ 生产进度、质量数据

class OEEAnalyzer:
    def __init__(self):
        self.availability_threshold = 0.85
        self.performance_threshold = 0.95
        self.quality_threshold = 0.99

    def calculate_oee(self, production_data):
        """计算OEE = 可用率 × 性能率 × 质量率"""
        # 可用率 = 实际运行时间 / 计划生产时间
        availability = production_data['actual_runtime'] / production_data['planned_runtime']

        # 性能率 = 实际产量 / (实际运行时间 × 理论产能)
        performance = (production_data['actual_output'] /
                      (production_data['actual_runtime'] * production_data['theoretical_speed']))

        # 质量率 = 合格品数量 / 总产量
        quality = production_data['good_output'] / production_data['actual_output']

        oee = availability * performance * quality

        return {
            'oee': oee,
            'availability': availability,
            'performance': performance,
            'quality': quality,
            'improvement_priorities': self.identify_bottlenecks(availability, performance, quality)
        }

    def identify_bottlenecks(self, availability, performance, quality):
        """识别改进重点"""
        priorities = []
        if availability < self.availability_threshold:
            priorities.append('设备可用率')
        if performance < self.performance_threshold:
            priorities.append('生产效率')
        if quality < self.quality_threshold:
            priorities.append('质量控制')
        return priorities

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

class SupplierRiskAnalyzer:
    def __init__(self):
        self.risk_factors = [
            'delivery_performance',  # 交付表现
            'quality_score',        # 质量评分
            'financial_health',     # 财务健康度
            'capacity_utilization', # 产能利用率
            'geographic_risk',      # 地理风险
            'compliance_score'      # 合规评分
        ]

    def calculate_risk_score(self, supplier_data):
        """计算供应商风险评分"""
        # 标准化处理
        scaler = StandardScaler()
        normalized_data = scaler.fit_transform(supplier_data[self.risk_factors])

        # 权重设置
        weights = {
            'delivery_performance': 0.25,
            'quality_score': 0.25,
            'financial_health': 0.20,
            'capacity_utilization': 0.15,
            'geographic_risk': 0.10,
            'compliance_score': 0.05
        }

        # 计算加权风险评分
        risk_scores = []
        for i, supplier in enumerate(normalized_data):
            weighted_score = sum(supplier[j] * weights[factor]
                               for j, factor in enumerate(self.risk_factors))
            risk_scores.append(weighted_score)

        supplier_data['risk_score'] = risk_scores
        supplier_data['risk_level'] = pd.cut(risk_scores,
                                           bins=[0, 0.3, 0.6, 1.0],
                                           labels=['低风险', '中风险', '高风险'])

        return supplier_data

    def recommend_actions(self, supplier_data):
        """推荐风险应对措施"""
        recommendations = []
        for _, supplier in supplier_data.iterrows():
            if supplier['risk_level'] == '高风险':
                recommendations.append({
                    'supplier_id': supplier['supplier_id'],
                    'actions': ['寻找备用供应商', '增加库存缓冲', '加强监控'],
                    'priority': 'High'
                })
            elif supplier['risk_level'] == '中风险':
                recommendations.append({
                    'supplier_id': supplier['supplier_id'],
                    'actions': ['定期评估', '改进计划'],
                    'priority': 'Medium'
                })

        return recommendations

class QualityRootCauseAnalyzer:
    def __init__(self):
        self.process_parameters = [
            'temperature', 'pressure', 'humidity', 'speed',
            'material_batch', 'operator_id', 'equipment_id'
        ]

    def analyze_defect_patterns(self, quality_data, process_data):
        """分析缺陷模式"""
        import scipy.stats as stats

        # 合并质量和工艺数据
        merged_data = pd.merge(quality_data, process_data, on='timestamp')

        # 按缺陷类型分组分析
        defect_analysis = {}
        for defect_type in merged_data['defect_type'].unique():
            if defect_type != 'normal':
                defect_data = merged_data[merged_data['defect_type'] == defect_type]
                normal_data = merged_data[merged_data['defect_type'] == 'normal']

                significant_factors = []
                for param in self.process_parameters:
                    if param in merged_data.columns:
                        # 进行t检验
                        t_stat, p_value = stats.ttest_ind(
                            defect_data[param].dropna(),
                            normal_data[param].dropna()
                        )

                        if p_value < 0.05:  # 显著性水平
                            significant_factors.append({
                                'parameter': param,
                                'p_value': p_value,
                                'defect_mean': defect_data[param].mean(),
                                'normal_mean': normal_data[param].mean(),
                                'impact_direction': 'higher' if defect_data[param].mean() > normal_data[param].mean() else 'lower'
                            })

                defect_analysis[defect_type] = significant_factors

        return defect_analysis

    def generate_improvement_suggestions(self, root_cause_analysis):
        """生成改进建议"""
        suggestions = {}

        for defect_type, factors in root_cause_analysis.items():
            defect_suggestions = []

            for factor in factors:
                param = factor['parameter']
                direction = factor['impact_direction']

                if param == 'temperature':
                    if direction == 'higher':
                        defect_suggestions.append('降低工艺温度，加强冷却控制')
                    else:
                        defect_suggestions.append('提高工艺温度，确保充分反应')

                elif param == 'pressure':
                    if direction == 'higher':
                        defect_suggestions.append('降低工艺压力，检查压力控制系统')
                    else:
                        defect_suggestions.append('增加工艺压力，提高压实效果')

                # 可以继续添加其他参数的建议逻辑

            suggestions[defect_type] = defect_suggestions

        return suggestions

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest, RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import TimeSeriesSplit
import warnings
warnings.filterwarnings('ignore')

class PredictiveMaintenanceModel:
    def __init__(self):
        self.feature_window = 24  # 24小时特征窗口
        self.prediction_horizon = 72  # 提前72小时预警

    def engineer_features(self, sensor_data):
        """构造预测性维护特征"""
        features = []

        # 按设备分组处理
        for equipment_id in sensor_data['equipment_id'].unique():
            equipment_data = sensor_data[sensor_data['equipment_id'] == equipment_id].copy()
            equipment_data = equipment_data.sort_values('timestamp')

            # 时间序列特征
            for col in ['temperature', 'vibration', 'pressure', 'current']:
                if col in equipment_data.columns:
                    # 滑动窗口统计特征
                    equipment_data[f'{col}_mean_{self.feature_window}h'] = equipment_data[col].rolling(
                        window=self.feature_window).mean()
                    equipment_data[f'{col}_std_{self.feature_window}h'] = equipment_data[col].rolling(
                        window=self.feature_window).std()
                    equipment_data[f'{col}_max_{self.feature_window}h'] = equipment_data[col].rolling(
                        window=self.feature_window).max()
                    equipment_data[f'{col}_min_{self.feature_window}h'] = equipment_data[col].rolling(
                        window=self.feature_window).min()

                    # 趋势特征
                    equipment_data[f'{col}_trend'] = equipment_data[col].diff().rolling(
                        window=12).mean()

                    # 异常检测特征
                    isolation_forest = IsolationForest(contamination=0.1)
                    equipment_data[f'{col}_anomaly_score'] = isolation_forest.fit_predict(
                        equipment_datacol.fillna(method='ffill'))

            # 运行时间特征
            equipment_data['runtime_hours'] = (equipment_data['timestamp'] -
                                             equipment_data['timestamp'].iloc[0]).dt.total_seconds() / 3600

            # 维护历史特征
            if 'last_maintenance' in equipment_data.columns:
                equipment_data['days_since_maintenance'] = (
                    equipment_data['timestamp'] - equipment_data['last_maintenance']).dt.days

            features.append(equipment_data)

        return pd.concat(features, ignore_index=True)

    def create_failure_labels(self, equipment_data):
        """创建故障预测标签"""
        # 基于未来故障时间创建标签
        equipment_data['failure_in_next_72h'] = 0

        for equipment_id in equipment_data['equipment_id'].unique():
            equipment_mask = equipment_data['equipment_id'] == equipment_id
            equipment_subset = equipment_data[equipment_mask].copy()

            # 找到故障时间点
            failure_times = equipment_subset[equipment_subset['failure_occurred'] == 1]['timestamp']

            for failure_time in failure_times:
                # 在故障前72小时内的数据点标记为正样本
                prediction_window = pd.Timedelta(hours=self.prediction_horizon)
                prediction_mask = (
                    (equipment_subset['timestamp'] >= failure_time - prediction_window) &
                    (equipment_subset['timestamp'] <= failure_time)
                )
                equipment_data.loc[equipment_mask & prediction_mask, 'failure_in_next_72h'] = 1

        return equipment_data

    def train_model(self, feature_data):
        """训练预测模型"""
        # 准备特征和标签
        feature_columns = [col for col in feature_data.columns
                          if col not in ['timestamp', 'equipment_id', 'failure_occurred', 'failure_in_next_72h']]

        X = feature_data[feature_columns].fillna(method='ffill').fillna(0)
        y = feature_data['failure_in_next_72h']

        # 时间序列交叉验证
        tscv = TimeSeriesSplit(n_splits=5)

        # 训练随机森林模型
        model = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            min_samples_split=20,
            class_weight='balanced',  # 处理不平衡数据
            random_state=42
        )

        # 标准化特征
        scaler = StandardScaler()
        X_scaled = scaler.fit_transform(X)

        model.fit(X_scaled, y)

        # 特征重要性分析
        feature_importance = pd.DataFrame({
            'feature': feature_columns,
            'importance': model.feature_importances_
        }).sort_values('importance', ascending=False)

        return {
            'model': model,
            'scaler': scaler,
            'feature_columns': feature_columns,
            'feature_importance': feature_importance
        }

    def predict_failures(self, model_dict, new_data):
        """预测设备故障"""
        model = model_dict['model']
        scaler = model_dict['scaler']
        feature_columns = model_dict['feature_columns']

        # 特征工程
        engineered_data = self.engineer_features(new_data)

        # 预测
        X_new = engineered_data[feature_columns].fillna(method='ffill').fillna(0)
        X_new_scaled = scaler.transform(X_new)

        failure_probability = model.predict_proba(X_new_scaled)[:, 1]
        failure_prediction = model.predict(X_new_scaled)

        # 添加预测结果
        engineered_data['failure_probability'] = failure_probability
        engineered_data['failure_prediction'] = failure_prediction
        engineered_data['risk_level'] = pd.cut(failure_probability,
                                              bins=[0, 0.3, 0.7, 1.0],
                                              labels=['低风险', '中风险', '高风险'])

        return engineered_data

from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import GridSearchCV
from scipy.optimize import minimize
import numpy as np

class ProcessOptimizer:
    def __init__(self):
        self.process_params = [
            'temperature', 'pressure', 'flow_rate',
            'catalyst_concentration', 'reaction_time'
        ]
        self.quality_metrics = ['yield', 'purity', 'viscosity']

    def build_process_model(self, historical_data):
        """构建工艺参数-质量关系模型"""
        models = {}

        for quality_metric in self.quality_metrics:
            if quality_metric in historical_data.columns:
                X = historical_data[self.process_params]
                y = historical_data[quality_metric]

                # 网格搜索优化超参数
                param_grid = {
                    'n_estimators': [100, 200],
                    'max_depth': [5, 10, 15],
                    'learning_rate': [0.01, 0.1, 0.2]
                }

                gbr = GradientBoostingRegressor(random_state=42)
                grid_search = GridSearchCV(gbr, param_grid, cv=5, scoring='r2')
                grid_search.fit(X, y)

                models[quality_metric] = {
                    'model': grid_search.best_estimator_,
                    'score': grid_search.best_score_,
                    'params': grid_search.best_params_
                }

        return models

    def optimize_parameters(self, models, constraints, objectives):
        """多目标工艺参数优化"""
        def objective_function(params):
            """目标函数：最大化质量指标加权和"""
            param_dict = dict(zip(self.process_params, params))
            param_array = np.array([params])

            total_score = 0
            for metric, weight in objectives.items():
                if metric in models:
                    predicted_quality = models[metric]['model'].predict(param_array)[0]
                    total_score += weight * predicted_quality

            return -total_score  # 最小化负值等于最大化

        # 参数约束
        bounds = []
        for param in self.process_params:
            if param in constraints:
                bounds.append((constraints[param]['min'], constraints[param]['max']))
            else:
                bounds.append((0, 100))  # 默认约束

        # 优化求解
        result = minimize(
            objective_function,
            x0=[np.mean([bound[0], bound[1]]) for bound in bounds],  # 初始值
            bounds=bounds,
            method='L-BFGS-B'
        )

        optimal_params = dict(zip(self.process_params, result.x))

        # 预测优化后的质量指标
        predicted_qualities = {}
        param_array = np.array([result.x])
        for metric in self.quality_metrics:
            if metric in models:
                predicted_qualities[metric] = models[metric]['model'].predict(param_array)[0]

        return {
            'optimal_parameters': optimal_params,
            'predicted_qualities': predicted_qualities,
            'optimization_success': result.success,
            'improvement_potential': -result.fun
        }

    def sensitivity_analysis(self, models, base_params):
        """参数敏感性分析"""
        sensitivity_results = {}

        for param in self.process_params:
            param_effects = {}
            base_array = np.array([list(base_params.values())])
            base_predictions = {}

            # 基准预测
            for metric in self.quality_metrics:
                if metric in models:
                    base_predictions[metric] = models[metric]['model'].predict(base_array)[0]

            # 参数变化影响分析
            param_index = self.process_params.index(param)
            change_percentages = [-20, -10, -5, 5, 10, 20]

            for change_pct in change_percentages:
                modified_params = base_array.copy()
                modified_params[0, param_index] *= (1 + change_pct / 100)

                effects = {}
                for metric in self.quality_metrics:
                    if metric in models:
                        new_prediction = models[metric]['model'].predict(modified_params)[0]
                        effect = ((new_prediction - base_predictions[metric]) /
                                base_predictions[metric] * 100)
                        effects[metric] = effect

                param_effects[f'{change_pct}%'] = effects

            sensitivity_results[param] = param_effects

        return sensitivity_results

import asyncio
import json
from datetime import datetime
from typing import Dict, List
import kafka
from influxdb_client import InfluxDBClient
import redis

class IoTDataPipeline:
    def __init__(self, config):
        self.config = config
        self.kafka_producer = kafka.KafkaProducer(
            bootstrap_servers=config['kafka']['servers'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.influx_client = InfluxDBClient(
            url=config['influxdb']['url'],
            token=config['influxdb']['token'],
            org=config['influxdb']['org']
        )
        self.redis_client = redis.Redis(
            host=config['redis']['host'],
            port=config['redis']['port']
        )

    def validate_sensor_data(self, data):
        """传感器数据验证"""
        required_fields = ['device_id', 'timestamp', 'value', 'sensor_type']

        # 基础字段检查
        if not all(field in data for field in required_fields):
            return False, "缺少必需字段"

        # 数据类型检查
        if not isinstance(data['value'], (int, float)):
            return False, "数值类型错误"

        # 时间戳检查
        try:
            timestamp = datetime.fromisoformat(data['timestamp'])
            now = datetime.now()
            if abs((timestamp - now).total_seconds()) > 300:  # 5分钟容忍度
                return False, "时间戳异常"
        except:
            return False, "时间戳格式错误"

        # 数值范围检查
        sensor_limits = {
            'temperature': (-50, 1000),
            'pressure': (0, 1000),
            'vibration': (0, 100),
            'flow_rate': (0, 1000)
        }

        if data['sensor_type'] in sensor_limits:
            min_val, max_val = sensor_limits[data['sensor_type']]
            if not (min_val <= data['value'] <= max_val):
                return False, f"数值超出范围 [{min_val}, {max_val}]"

        return True, "验证通过"

    async def process_sensor_data(self, raw_data):
        """处理传感器数据"""
        # 数据验证
        is_valid, message = self.validate_sensor_data(raw_data)
        if not is_valid:
            await self.handle_invalid_data(raw_data, message)
            return

        # 数据清洗和转换
        cleaned_data = self.clean_data(raw_data)

        # 异常检测
        anomaly_score = await self.detect_anomaly(cleaned_data)
        cleaned_data['anomaly_score'] = anomaly_score

        # 数据分发
        await asyncio.gather(
            self.send_to_real_time_processing(cleaned_data),
            self.store_to_time_series_db(cleaned_data),
            self.update_device_status(cleaned_data)
        )

    def clean_data(self, data):
        """数据清洗"""
        cleaned = data.copy()

        # 数值平滑（移动平均）
        device_id = data['device_id']
        sensor_type = data['sensor_type']

        # 从Redis获取历史数据
        history_key = f"sensor_history:{device_id}:{sensor_type}"
        history = self.redis_client.lrange(history_key, 0, 4)  # 获取最近5个值

        if history:
            history_values = [float(val) for val in history]
            history_values.append(data['value'])
            smoothed_value = sum(history_values) / len(history_values)
            cleaned['smoothed_value'] = smoothed_value
        else:
            cleaned['smoothed_value'] = data['value']

        # 更新历史数据
        self.redis_client.lpush(history_key, data['value'])
        self.redis_client.ltrim(history_key, 0, 9)  # 保留最近10个值
        self.redis_client.expire(history_key, 3600)  # 1小时过期

        return cleaned

    async def detect_anomaly(self, data):
        """异常检测"""
        device_id = data['device_id']
        sensor_type = data['sensor_type']
        current_value = data['value']

        # 从Redis获取统计信息
        stats_key = f"sensor_stats:{device_id}:{sensor_type}"
        stats = self.redis_client.hgetall(stats_key)

        if stats:
            mean = float(stats.get(b'mean', current_value))
            std = float(stats.get(b'std', 0))
            count = int(stats.get(b'count', 1))

            # 更新统计信息（在线算法）
            new_count = count + 1
            new_mean = (mean * count + current_value) / new_count

            if count > 1:
                # 在线方差更新
                old_variance = std ** 2
                new_variance = ((count - 1) * old_variance +
                              (current_value - mean) * (current_value - new_mean)) / count
                new_std = new_variance ** 0.5
            else:
                new_std = 0

            # 异常评分（基于z-score）
            if new_std > 0:
                z_score = abs(current_value - new_mean) / new_std
                anomaly_score = min(z_score / 3.0, 1.0)  # 标准化到[0,1]
            else:
                anomaly_score = 0

            # 更新Redis统计信息
            self.redis_client.hset(stats_key, mapping={
                'mean': new_mean,
                'std': new_std,
                'count': new_count
            })
            self.redis_client.expire(stats_key, 86400)  # 24小时过期

        else:
            # 初始化统计信息
            self.redis_client.hset(stats_key, mapping={
                'mean': current_value,
                'std': 0,
                'count': 1
            })
            anomaly_score = 0

        return anomaly_score

    async def send_to_real_time_processing(self, data):
        """发送到实时处理系统"""
        topic_mapping = {
            'temperature': 'sensor_temperature',
            'pressure': 'sensor_pressure',
            'vibration': 'sensor_vibration',
            'flow_rate': 'sensor_flow'
        }

        topic = topic_mapping.get(data['sensor_type'], 'sensor_general')

        # 添加分区键（按设备ID分区）
        partition_key = data['device_id']

        self.kafka_producer.send(
            topic,
            value=data,
            key=partition_key.encode('utf-8')
        )

    async def store_to_time_series_db(self, data):
        """存储到时序数据库"""
        write_api = self.influx_client.write_api()

        point = {
            "measurement": f"sensor_{data['sensor_type']}",
            "tags": {
                "device_id": data['device_id'],
                "factory": data.get('factory', 'unknown'),
                "line": data.get('production_line', 'unknown')
            },
            "fields": {
                "value": data['value'],
                "smoothed_value": data['smoothed_value'],
                "anomaly_score": data['anomaly_score']
            },
            "time": data['timestamp']
        }

        write_api.write(
            bucket=self.config['influxdb']['bucket'],
            record=point
        )

    async def update_device_status(self, data):
        """更新设备状态"""
        device_id = data['device_id']

        # 设备状态逻辑
        status = "normal"
        if data['anomaly_score'] > 0.8:
            status = "warning"
        elif data['anomaly_score'] > 0.95:
            status = "critical"

        # 更新Redis设备状态
        device_status = {
            'last_update': data['timestamp'],
            'status': status,
            'anomaly_score': data['anomaly_score']
        }

        self.redis_client.hset(
            f"device_status:{device_id}",
            mapping=device_status
        )

        # 如果是告警状态，发送告警消息
        if status in ['warning', 'critical']:
            alert_data = {
                'device_id': device_id,
                'alert_type': status,
                'timestamp': data['timestamp'],
                'anomaly_score': data['anomaly_score'],
                'sensor_type': data['sensor_type'],
                'value': data['value']
            }

            self.kafka_producer.send('alerts', value=alert_data)

    async def handle_invalid_data(self, data, error_message):
        """处理无效数据"""
        error_record = {
            'original_data': data,
            'error_message': error_message,
            'timestamp': datetime.now().isoformat(),
            'error_type': 'validation_failed'
        }

        # 发送到错误处理队列
        self.kafka_producer.send('data_errors', value=error_record)

        # 记录错误统计
        error_key = f"error_count:{data.get('device_id', 'unknown')}"
        self.redis_client.incr(error_key)
        self.redis_client.expire(error_key, 86400)

-- 制造业数据仓库设计

-- 1. 时间维度表
CREATE TABLE dim_time (
    time_key INT PRIMARY KEY,
    date_value DATE,
    year_value INT,
    quarter_value INT,
    month_value INT,
    week_value INT,
    day_value INT,
    hour_value INT,
    minute_value INT,
    is_working_day BOOLEAN,
    shift_code VARCHAR(10),
    INDEX idx_date (date_value),
    INDEX idx_shift (shift_code)
);

-- 2. 设备维度表
CREATE TABLE dim_equipment (
    equipment_key INT PRIMARY KEY AUTO_INCREMENT,
    equipment_id VARCHAR(50) UNIQUE NOT NULL,
    equipment_name VARCHAR(200),
    equipment_type VARCHAR(100),
    manufacturer VARCHAR(100),
    model VARCHAR(100),
    production_line_id VARCHAR(50),
    factory_id VARCHAR(50),
    installation_date DATE,
    capacity_per_hour DECIMAL(10,2),
    status VARCHAR(20),
    effective_date DATE,
    expiry_date DATE,
    INDEX idx_equipment_id (equipment_id),
    INDEX idx_line (production_line_id),
    INDEX idx_factory (factory_id)
);

-- 3. 产品维度表
CREATE TABLE dim_product (
    product_key INT PRIMARY KEY AUTO_INCREMENT,
    product_id VARCHAR(50) UNIQUE NOT NULL,
    product_name VARCHAR(200),
    product_category VARCHAR(100),
    product_family VARCHAR(100),
    standard_cost DECIMAL(10,2),
    target_quality_score DECIMAL(5,2),
    effective_date DATE,
    expiry_date DATE,
    INDEX idx_product_id (product_id),
    INDEX idx_category (product_category)
);

-- 4. 工厂维度表
CREATE TABLE dim_factory (
    factory_key INT PRIMARY KEY AUTO_INCREMENT,
    factory_id VARCHAR(50) UNIQUE NOT NULL,
    factory_name VARCHAR(200),
    region VARCHAR(100),
    country VARCHAR(100),
    manager_name VARCHAR(100),
    capacity_rating VARCHAR(50),
    certification_level VARCHAR(50),
    INDEX idx_factory_id (factory_id),
    INDEX idx_region (region)
);

-- 5. 生产事实表
CREATE TABLE fact_production (
    production_key BIGINT PRIMARY KEY AUTO_INCREMENT,
    time_key INT,
    equipment_key INT,
    product_key INT,
    factory_key INT,
    batch_number VARCHAR(100),
    planned_quantity DECIMAL(12,2),
    actual_quantity DECIMAL(12,2),
    defect_quantity DECIMAL(12,2),
    scrap_quantity DECIMAL(12,2),
    production_time_minutes INT,
    setup_time_minutes INT,
    downtime_minutes INT,
    material_cost DECIMAL(12,2),
    labor_cost DECIMAL(12,2),
    overhead_cost DECIMAL(12,2),
    quality_score DECIMAL(5,2),
    efficiency_rate DECIMAL(5,4),
    created_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (time_key) REFERENCES dim_time(time_key),
    FOREIGN KEY (equipment_key) REFERENCES dim_equipment(equipment_key),
    FOREIGN KEY (product_key) REFERENCES dim_product(product_key),
    FOREIGN KEY (factory_key) REFERENCES dim_factory(factory_key),
    INDEX idx_time (time_key),
    INDEX idx_equipment (equipment_key),
    INDEX idx_product (product_key),
    INDEX idx_batch (batch_number)
);

-- 6. 质量检测事实表
CREATE TABLE fact_quality (
    quality_key BIGINT PRIMARY KEY AUTO_INCREMENT,
    time_key INT,
    equipment_key INT,
    product_key INT,
    batch_number VARCHAR(100),
    inspection_type VARCHAR(50),
    defect_type VARCHAR(100),
    defect_severity VARCHAR(20),
    inspector_id VARCHAR(50),
    test_parameter VARCHAR(100),
    measured_value DECIMAL(12,4),
    specification_min DECIMAL(12,4),
    specification_max DECIMAL(12,4),
    is_conforming BOOLEAN,
    corrective_action VARCHAR(500),
    FOREIGN KEY (time_key) REFERENCES dim_time(time_key),
    FOREIGN KEY (equipment_key) REFERENCES dim_equipment(equipment_key),
    FOREIGN KEY (product_key) REFERENCES dim_product(product_key),
    INDEX idx_time_quality (time_key),
    INDEX idx_batch_quality (batch_number),
    INDEX idx_defect (defect_type)
);

-- 7. 设备监控事实表
CREATE TABLE fact_equipment_monitoring (
    monitoring_key BIGINT PRIMARY KEY AUTO_INCREMENT,
    time_key INT,
    equipment_key INT,
    sensor_type VARCHAR(50),
    measured_value DECIMAL(12,4),
    normal_range_min DECIMAL(12,4),
    normal_range_max DECIMAL(12,4),
    anomaly_score DECIMAL(5,4),
    alert_level VARCHAR(20),
    maintenance_due_days INT,
    FOREIGN KEY (time_key) REFERENCES dim_time(time_key),
    FOREIGN KEY (equipment_key) REFERENCES dim_equipment(equipment_key),
    INDEX idx_time_monitoring (time_key),
    INDEX idx_equipment_monitoring (equipment_key),
    INDEX idx_sensor (sensor_type),
    INDEX idx_alert (alert_level)
);

-- 8. 创建生产效率分析视图
CREATE VIEW view_production_efficiency AS
SELECT
    f.factory_name,
    e.production_line_id,
    e.equipment_name,
    p.product_category,
    t.date_value,
    t.shift_code,
    SUM(fp.actual_quantity) as total_output,
    SUM(fp.planned_quantity) as total_planned,
    SUM(fp.actual_quantity) / SUM(fp.planned_quantity) as output_efficiency,
    SUM(fp.production_time_minutes) as total_production_time,
    SUM(fp.downtime_minutes) as total_downtime,
    (SUM(fp.production_time_minutes) - SUM(fp.downtime_minutes)) /
    SUM(fp.production_time_minutes) as availability_rate,
    AVG(fp.quality_score) as avg_quality_score,
    SUM(fp.defect_quantity) / SUM(fp.actual_quantity) as defect_rate
FROM fact_production fp
JOIN dim_time t ON fp.time_key = t.time_key
JOIN dim_equipment e ON fp.equipment_key = e.equipment_key
JOIN dim_product p ON fp.product_key = p.product_key
JOIN dim_factory f ON fp.factory_key = f.factory_key
GROUP BY
    f.factory_name, e.production_line_id, e.equipment_name,
    p.product_category, t.date_value, t.shift_code;

-- 9. 创建设备健康度分析视图
CREATE VIEW view_equipment_health AS
SELECT
    e.equipment_id,
    e.equipment_name,
    e.production_line_id,
    f.factory_name,
    AVG(fem.anomaly_score) as avg_anomaly_score,
    COUNT(CASE WHEN fem.alert_level = 'critical' THEN 1 END) as critical_alerts,
    COUNT(CASE WHEN fem.alert_level = 'warning' THEN 1 END) as warning_alerts,
    MIN(fem.maintenance_due_days) as days_to_maintenance,
    CASE
        WHEN AVG(fem.anomaly_score) > 0.8 THEN 'Poor'
        WHEN AVG(fem.anomaly_score) > 0.5 THEN 'Fair'
        WHEN AVG(fem.anomaly_score) > 0.2 THEN 'Good'
        ELSE 'Excellent'
    END as health_status
FROM dim_equipment e
JOIN fact_equipment_monitoring fem ON e.equipment_key = fem.equipment_key
JOIN dim_factory f ON e.factory_id = f.factory_id
JOIN dim_time t ON fem.time_key = t.time_key
WHERE t.date_value >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
GROUP BY e.equipment_id, e.equipment_name, e.production_line_id, f.factory_name;

import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class ManufacturingDashboard:
    def __init__(self):
        self.colors = {
            'primary': '#1f77b4',
            'success': '#2ca02c',
            'warning': '#ff7f0e',
            'danger': '#d62728',
            'info': '#17a2b8'
        }

    def create_oee_gauge(self, current_oee, target_oee=0.85):
        """创建OEE仪表盘"""
        fig = go.Figure(go.Indicator(
            mode = "gauge+number+delta",
            value = current_oee,
            domain = {'x': [0, 1], 'y': [0, 1]},
            title = {'text': "整体设备效率 (OEE)"},
            delta = {'reference': target_oee, 'valueformat': ".1%"},
            gauge = {
                'axis': {'range': [None, 1], 'tickformat': '.0%'},
                'bar': {'color': self.colors['primary']},
                'steps': [
                    {'range': [0, 0.5], 'color': self.colors['danger']},
                    {'range': [0.5, 0.75], 'color': self.colors['warning']},
                    {'range': [0.75, 0.85], 'color': self.colors['info']},
                    {'range': [0.85, 1], 'color': self.colors['success']}
                ],
                'threshold': {
                    'line': {'color': "red", 'width': 4},
                    'thickness': 0.75,
                    'value': target_oee
                }
            }
        ))

        fig.update_layout(
            height=300,
            font={'color': "darkblue", 'family': "Arial"}
        )

        return fig

    def create_production_timeline(self, production_data):
        """创建生产时间线图"""
        fig = make_subplots(
            rows=3, cols=1,
            subplot_titles=['产量趋势', '质量评分', '设备利用率'],
            vertical_spacing=0.08,
            shared_xaxes=True
        )

        # 产量趋势
        fig.add_trace(
            go.Scatter(
                x=production_data['timestamp'],
                y=production_data['hourly_output'],
                mode='lines+markers',
                name='实际产量',
                line=dict(color=self.colors['primary'], width=2)
            ),
            row=1, col=1
        )

        fig.add_trace(
            go.Scatter(
                x=production_data['timestamp'],
                y=production_data['target_output'],
                mode='lines',
                name='目标产量',
                line=dict(color=self.colors['warning'], dash='dash')
            ),
            row=1, col=1
        )

        # 质量评分
        fig.add_trace(
            go.Scatter(
                x=production_data['timestamp'],
                y=production_data['quality_score'],
                mode='lines+markers',
                name='质量评分',
                line=dict(color=self.colors['success'], width=2)
            ),
            row=2, col=1
        )

        # 设备利用率
        fig.add_trace(
            go.Scatter(
                x=production_data['timestamp'],
                y=production_data['utilization_rate'],
                mode='lines+markers',
                name='设备利用率',
                line=dict(color=self.colors['info'], width=2),
                fill='tonexty'
            ),
            row=3, col=1
        )

        fig.update_layout(
            height=600,
            title_text="生产监控时间线",
            showlegend=True
        )

        fig.update_xaxes(title_text="时间", row=3, col=1)
        fig.update_yaxes(title_text="产量 (件/小时)", row=1, col=1)
        fig.update_yaxes(title_text="质量评分", row=2, col=1)
        fig.update_yaxes(title_text="利用率 (%)", row=3, col=1)

        return fig

    def create_defect_analysis(self, defect_data):
        """创建缺陷分析图表"""
        fig = make_subplots(
            rows=1, cols=2,
            subplot_titles=['缺陷类型分布', '缺陷趋势分析'],
            specs={"type": "pie"}, {"type": "bar"}
        )

        # 缺陷类型饼图
        defect_counts = defect_data.groupby('defect_type')['count'].sum()

        fig.add_trace(
            go.Pie(
                labels=defect_counts.index,
                values=defect_counts.values,
                name="缺陷分布",
                marker_colors=px.colors.qualitative.Set3
            ),
            row=1, col=1
        )

        # 缺陷趋势柱状图
        daily_defects = defect_data.groupby(['date', 'defect_type'])['count'].sum().reset_index()

        for defect_type in daily_defects['defect_type'].unique():
            type_data = daily_defects[daily_defects['defect_type'] == defect_type]
            fig.add_trace(
                go.Bar(
                    x=type_data['date'],
                    y=type_data['count'],
                    name=defect_type
                ),
                row=1, col=2
            )

        fig.update_layout(
            height=400,
            title_text="质量缺陷分析"
        )

        return fig

    def create_equipment_heatmap(self, equipment_data):
        """创建设备状态热力图"""
        # 准备热力图数据
        pivot_data = equipment_data.pivot_table(
            index='equipment_id',
            columns='hour',
            values='efficiency',
            aggfunc='mean'
        )

        fig = go.Figure(data=go.Heatmap(
            z=pivot_data.values,
            x=pivot_data.columns,
            y=pivot_data.index,
            colorscale='RdYlGn',
            text=pivot_data.values,
            texttemplate="%{text:.1%}",
            textfont={"size": 10},
            colorbar=dict(
                title="设备效率",
                tickformat=".0%"
            )
        ))

        fig.update_layout(
            title='24小时设备效率热力图',
            xaxis_title='小时',
            yaxis_title='设备ID',
            height=500
        )

        return fig

    def create_kpi_cards(self, kpi_data):
        """创建KPI卡片"""
        kpi_cards = []

        kpi_configs = [
            {
                'title': '当日产量',
                'value': kpi_data['daily_output'],
                'unit': '件',
                'target': kpi_data['daily_target'],
                'format': '{:,.0f}',
                'color': self.colors['primary']
            },
            {
                'title': 'OEE',
                'value': kpi_data['current_oee'],
                'unit': '%',
                'target': 0.85,
                'format': '{:.1%}',
                'color': self.colors['success']
            },
            {
                'title': '缺陷率',
                'value': kpi_data['defect_rate'],
                'unit': '%',
                'target': 0.02,
                'format': '{:.2%}',
                'color': self.colors['warning'],
                'reverse': True  # 越低越好
            },
            {
                'title': '设备可用率',
                'value': kpi_data['availability'],
                'unit': '%',
                'target': 0.95,
                'format': '{:.1%}',
                'color': self.colors['info']
            }
        ]

        for config in kpi_configs:
            # 计算趋势
            is_good = (config['value'] >= config['target']) if not config.get('reverse') else (config['value'] <= config['target'])
            trend_color = self.colors['success'] if is_good else self.colors['danger']
            trend_icon = '↑' if is_good else '↓'

            card_html = f"""
            <div style="
                background-color: white;
                border-left: 4px solid {config['color']};
                padding: 20px;
                margin: 10px;
                border-radius: 8px;
                box-shadow: 0 2px 4px rgba(0,0,0,0.1);
                min-width: 200px;
            ">
                <h3 style="margin: 0; color: #666; font-size: 14px;">{config['title']}</h3>
                <div style="display: flex; align-items: center; margin: 10px 0;">
                    <span style="font-size: 28px; font-weight: bold; color: {config['color']};">
                        {config['format'].format(config['value'])}
                    </span>
                    <span style="margin-left: 10px; color: {trend_color}; font-size: 20px;">
                        {trend_icon}
                    </span>
                </div>
                <div style="font-size: 12px; color: #999;">
                    目标: {config['format'].format(config['target'])}
                </div>
            </div>
            """
            kpi_cards.append(card_html)

        return kpi_cards

    def create_alert_panel(self, alert_data):
        """创建告警面板"""
        # 按严重程度分类告警
        critical_alerts = alert_data[alert_data['severity'] == 'critical']
        warning_alerts = alert_data[alert_data['severity'] == 'warning']

        alert_html = f"""
        <div style="background-color: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
            <h3 style="margin: 0 0 15px 0; color: #333;">实时告警</h3>

            <div style="margin-bottom: 15px;">
                <span style="background-color: {self.colors['danger']}; color: white; padding: 4px 8px; border-radius: 4px; font-size: 12px;">
                    严重告警: {len(critical_alerts)}
                </span>
                <span style="background-color: {self.colors['warning']}; color: white; padding: 4px 8px; border-radius: 4px; font-size: 12px; margin-left: 10px;">
                    警告: {len(warning_alerts)}
                </span>
            </div>
        """

        # 显示最近的告警
        recent_alerts = alert_data.head(5)
        for _, alert in recent_alerts.iterrows():
            severity_color = self.colors['danger'] if alert['severity'] == 'critical' else self.colors['warning']
            alert_html += f"""
            <div style="border-left: 3px solid {severity_color}; padding: 8px 12px; margin: 8px 0; background-color: #f8f9fa;">
                <div style="font-weight: bold; color: {severity_color};">{alert['equipment_id']}</div>
                <div style="font-size: 12px; color: #666;">{alert['message']}</div>
                <div style="font-size: 10px; color: #999;">{alert['timestamp']}</div>
            </div>
            """

        alert_html += "</div>"
        return alert_html