Manufacturing IoT Industrial Pipeline

-- Table pour les lectures de capteurs avec compression temporelle
CREATE TABLE FACT_SENSOR_READING (
    reading_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    sensor_sk INT NOT NULL,
    equipment_sk INT NOT NULL,
    timestamp TIMESTAMP(3) NOT NULL,
    reading_value DECIMAL(15,6),
    reading_unit VARCHAR(20),
    quality_flag BOOLEAN DEFAULT TRUE,
    compression_level VARCHAR(20), -- raw, 1s, 1min, 5min
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_sensor_timestamp (sensor_sk, timestamp),
    INDEX idx_equipment_timestamp (equipment_sk, timestamp),
    INDEX idx_compression (compression_level)
);

-- Table pour les événements de production discrets
CREATE TABLE FACT_PRODUCTION_EVENT (
    event_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    equipment_sk INT NOT NULL,
    event_type VARCHAR(100) NOT NULL, -- start_batch, end_batch, change_parameters
    event_timestamp TIMESTAMP NOT NULL,
    operator_sk INT,
    parameters_json JSON,
    duration_minutes INT,
    
    INDEX idx_equipment_event (equipment_sk, event_timestamp),
    INDEX idx_event_type (event_type)
);

-- Table pour les mesures de qualité
CREATE TABLE FACT_QUALITY_MEASUREMENT (
    measurement_sk BIGINT PRIMARY KEY AUTO_INCREMENT,
    product_sk INT NOT NULL,
    equipment_sk INT NOT NULL,
    measurement_timestamp TIMESTAMP NOT NULL,
    measurement_type VARCHAR(100), -- dimensional, destructive, visual
    measurement_value DECIMAL(15,6),
    tolerance_upper DECIMAL(15,6),
    tolerance_lower DECIMAL(15,6),
    is_within_tolerance BOOLEAN,
    inspector_sk INT,
    
    INDEX idx_product_timestamp (product_sk, measurement_timestamp),
    INDEX idx_equipment_timestamp (equipment_sk, measurement_timestamp)
);

-- Dimension équipement avec hiérarchie flexible
CREATE TABLE DIM_EQUIPMENT (
    equipment_sk INT PRIMARY KEY AUTO_INCREMENT,
    equipment_id VARCHAR(100) NOT NULL,
    equipment_name VARCHAR(255) NOT NULL,
    equipment_type VARCHAR(100),
    parent_equipment_sk INT NULL,
    plant_sk INT NOT NULL,
    line_sk INT NOT NULL,
    cell_sk INT NOT NULL,
    nominal_capacity DECIMAL(15,6),
    mtbf_hours DECIMAL(10,2),
    operational_cost_per_hour DECIMAL(10,2),
    installation_date DATE,
    last_maintenance_date DATE,
    
    INDEX idx_equipment_id (equipment_id),
    INDEX idx_parent (parent_equipment_sk),
    INDEX idx_plant_line (plant_sk, line_sk)
);

-- Dimension produit avec spécifications et tolérances
CREATE TABLE DIM_PRODUCT_SPEC (
    spec_sk INT PRIMARY KEY AUTO_INCREMENT,
    product_sk INT NOT NULL,
    spec_name VARCHAR(100) NOT NULL,
    spec_value DECIMAL(15,6),
    upper_limit DECIMAL(15,6),
    lower_limit DECIMAL(15,6),
    unit VARCHAR(20),
    is_critical BOOLEAN DEFAULT FALSE,
    measurement_method VARCHAR(100),
    
    INDEX idx_product_spec (product_sk, spec_name),
    INDEX idx_critical (is_critical) WHERE is_critical = TRUE
);

// Configuration Azure IoT Edge pour le processing local
@Configuration
public class IoTCoreConfig {
    
    @Bean
    public IoTCoreClient ioTCoreClient() {
        IoTCoreClientBuilder builder = IoTCoreClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
    
    @Bean
    public TimeSeriesClient timeSeriesClient() {
        TimeSeriesClientBuilder builder = TimeSeriesClient.builder();
        builder.region(Region.US_EAST_1);
        builder.credentialsProvider(DefaultCredentialsProvider.create());
        
        return builder.build();
    }
}

// Service de processing edge pour la détection d'anomalies
@Service
public class EdgeProcessingService {
    
    private final AnomalyDetectionModel anomalyModel;
    private final DataCompressionService compressionService;
    
    public ProcessedData processSensorData(SensorData rawData) {
        try {
            // Validation des données en edge
            if (!validateSensorData(rawData)) {
                throw new InvalidDataException("Données capteur invalides");
            }
            
            // Détection d'anomalies locale
            AnomalyResult anomaly = anomalyModel.detectAnomaly(rawData);
            
            // Compression intelligente selon l'importance
            CompressedData compressed = compressionService.compressData(rawData, anomaly.getSeverity());
            
            // Enrichissement avec métadonnées edge
            return ProcessedData.builder()
                .originalData(rawData)
                .anomalyResult(anomaly)
                .compressedData(compressed)
                .edgeProcessingTimestamp(Instant.now())
                .edgeNodeId(getEdgeNodeId())
                .build();
                
        } catch (Exception e) {
            handleEdgeProcessingError(rawData, e);
            throw new EdgeProcessingException("Échec du processing edge", e);
        }
    }
}