#!/usr/bin/env python3 """ Mock Optimization Server for Testing Simulates an optimization system that provides setpoints and control strategies """ import asyncio import logging import random import time import json from datetime import datetime from typing import Dict, Any, List from dataclasses import dataclass from enum import Enum logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class OptimizationStrategy(Enum): """Optimization strategies""" ENERGY_EFFICIENCY = "energy_efficiency" COST_OPTIMIZATION = "cost_optimization" PRODUCTION_MAXIMIZATION = "production_maximization" QUALITY_OPTIMIZATION = "quality_optimization" @dataclass class OptimizationResult: """Result of optimization calculation""" strategy: OptimizationStrategy setpoints: Dict[str, float] cost_savings: float energy_savings: float production_increase: float quality_improvement: float confidence: float timestamp: datetime class MockOptimizationServer: """Mock optimization server that simulates optimization calculations""" def __init__(self, port: int = 8081): self.port = port self.running = False self.current_strategy = OptimizationStrategy.ENERGY_EFFICIENCY # Historical optimization results self.optimization_history: List[OptimizationResult] = [] # Current process state (would come from SCADA) self.current_state = { 'temperature': 25.0, 'pressure': 101.3, 'flow_rate': 100.0, 'level': 50.0, 'energy_consumption': 150.0, 'production_rate': 85.0, 'quality_metric': 92.0, 'operating_cost': 1250.0 } # Market and external factors self.market_data = { 'energy_price': 0.12, # $/kWh 'raw_material_cost': 45.0, # $/ton 'product_price': 150.0, # $/unit 'demand_factor': 0.95 } def calculate_optimization(self, strategy: OptimizationStrategy) -> OptimizationResult: """Calculate optimization based on current state and strategy""" # Simulate optimization calculation base_setpoints = { 'temperature_setpoint': 75.0, 'pressure_setpoint': 105.0, 'flow_setpoint': 120.0, 'level_setpoint': 60.0 } # Adjust setpoints based on strategy if strategy == OptimizationStrategy.ENERGY_EFFICIENCY: setpoints = { 'temperature_setpoint': base_setpoints['temperature_setpoint'] - 2.0, 'pressure_setpoint': base_setpoints['pressure_setpoint'] - 1.0, 'flow_setpoint': base_setpoints['flow_setpoint'] - 5.0, 'level_setpoint': base_setpoints['level_setpoint'] } cost_savings = random.uniform(5.0, 15.0) energy_savings = random.uniform(8.0, 20.0) production_increase = random.uniform(-2.0, 2.0) quality_improvement = random.uniform(-1.0, 1.0) elif strategy == OptimizationStrategy.COST_OPTIMIZATION: setpoints = { 'temperature_setpoint': base_setpoints['temperature_setpoint'] - 1.0, 'pressure_setpoint': base_setpoints['pressure_setpoint'] - 0.5, 'flow_setpoint': base_setpoints['flow_setpoint'] - 3.0, 'level_setpoint': base_setpoints['level_setpoint'] + 5.0 } cost_savings = random.uniform(10.0, 25.0) energy_savings = random.uniform(5.0, 12.0) production_increase = random.uniform(1.0, 5.0) quality_improvement = random.uniform(0.0, 2.0) elif strategy == OptimizationStrategy.PRODUCTION_MAXIMIZATION: setpoints = { 'temperature_setpoint': base_setpoints['temperature_setpoint'] + 3.0, 'pressure_setpoint': base_setpoints['pressure_setpoint'] + 2.0, 'flow_setpoint': base_setpoints['flow_setpoint'] + 10.0, 'level_setpoint': base_setpoints['level_setpoint'] - 5.0 } cost_savings = random.uniform(-5.0, 5.0) energy_savings = random.uniform(-10.0, 0.0) production_increase = random.uniform(8.0, 15.0) quality_improvement = random.uniform(-3.0, 0.0) else: # QUALITY_OPTIMIZATION setpoints = { 'temperature_setpoint': base_setpoints['temperature_setpoint'] + 1.0, 'pressure_setpoint': base_setpoints['pressure_setpoint'] + 0.5, 'flow_setpoint': base_setpoints['flow_setpoint'] - 2.0, 'level_setpoint': base_setpoints['level_setpoint'] + 2.0 } cost_savings = random.uniform(2.0, 8.0) energy_savings = random.uniform(3.0, 8.0) production_increase = random.uniform(-1.0, 1.0) quality_improvement = random.uniform(5.0, 12.0) # Add some randomness to simulate real optimization for key in setpoints: setpoints[key] += random.uniform(-1.0, 1.0) # Calculate confidence based on strategy and current conditions confidence = random.uniform(0.7, 0.95) result = OptimizationResult( strategy=strategy, setpoints=setpoints, cost_savings=cost_savings, energy_savings=energy_savings, production_increase=production_increase, quality_improvement=quality_improvement, confidence=confidence, timestamp=datetime.now() ) return result def get_optimal_strategy(self) -> OptimizationStrategy: """Determine the best strategy based on current conditions""" # Simple heuristic based on current state and market conditions if self.market_data['energy_price'] > 0.15: return OptimizationStrategy.ENERGY_EFFICIENCY elif self.market_data['demand_factor'] > 1.1: return OptimizationStrategy.PRODUCTION_MAXIMIZATION elif self.current_state['quality_metric'] < 90.0: return OptimizationStrategy.QUALITY_OPTIMIZATION else: return OptimizationStrategy.COST_OPTIMIZATION async def update_market_data(self): """Simulate changing market conditions""" while self.running: try: # Simulate market fluctuations self.market_data['energy_price'] += random.uniform(-0.01, 0.01) self.market_data['energy_price'] = max(0.08, min(0.20, self.market_data['energy_price'])) self.market_data['raw_material_cost'] += random.uniform(-1.0, 1.0) self.market_data['raw_material_cost'] = max(40.0, min(60.0, self.market_data['raw_material_cost'])) self.market_data['demand_factor'] += random.uniform(-0.05, 0.05) self.market_data['demand_factor'] = max(0.8, min(1.3, self.market_data['demand_factor'])) await asyncio.sleep(30) # Update every 30 seconds except Exception as e: logger.error(f"Error updating market data: {e}") await asyncio.sleep(10) async def run_optimization_cycle(self): """Run optimization cycles periodically""" while self.running: try: # Get optimal strategy strategy = self.get_optimal_strategy() self.current_strategy = strategy # Calculate optimization result = self.calculate_optimization(strategy) # Store in history self.optimization_history.append(result) # Keep only last 100 optimizations if len(self.optimization_history) > 100: self.optimization_history = self.optimization_history[-100:] logger.info(f"Optimization completed: {strategy.value} - Confidence: {result.confidence:.2f}") await asyncio.sleep(60) # Run optimization every minute except Exception as e: logger.error(f"Error in optimization cycle: {e}") await asyncio.sleep(10) def get_status(self) -> Dict[str, Any]: """Get server status""" latest_result = self.optimization_history[-1] if self.optimization_history else None return { 'running': self.running, 'current_strategy': self.current_strategy.value if self.current_strategy else None, 'market_data': self.market_data, 'optimization_count': len(self.optimization_history), 'latest_optimization': { 'strategy': latest_result.strategy.value if latest_result else None, 'setpoints': latest_result.setpoints if latest_result else {}, 'confidence': latest_result.confidence if latest_result else 0.0, 'timestamp': latest_result.timestamp.isoformat() if latest_result else None } if latest_result else None } async def start(self): """Start the mock optimization server""" if self.running: return self.running = True # Start background tasks self.market_task = asyncio.create_task(self.update_market_data()) self.optimization_task = asyncio.create_task(self.run_optimization_cycle()) logger.info(f"Mock optimization server started") async def stop(self): """Stop the mock optimization server""" self.running = False if hasattr(self, 'market_task'): self.market_task.cancel() if hasattr(self, 'optimization_task'): self.optimization_task.cancel() logger.info("Mock optimization server stopped") async def main(): """Main function to run the mock optimization server""" server = MockOptimizationServer() try: await server.start() # Keep server running while True: await asyncio.sleep(1) except KeyboardInterrupt: print("\nShutting down mock optimization server...") await server.stop() if __name__ == "__main__": asyncio.run(main())