280 lines
9.6 KiB
Python
280 lines
9.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Mock Optimizer Server for testing
|
||
|
|
Simulates optimization service for industrial processes
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import random
|
||
|
|
import time
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from flask import Flask, jsonify, request
|
||
|
|
|
||
|
|
app = Flask(__name__)
|
||
|
|
|
||
|
|
# Mock optimization models
|
||
|
|
optimization_models = {
|
||
|
|
"energy_optimization": {
|
||
|
|
"name": "Energy Consumption Optimizer",
|
||
|
|
"description": "Optimizes energy usage while maintaining production targets",
|
||
|
|
"parameters": ["power_load", "time_of_day", "production_rate"],
|
||
|
|
"outputs": ["optimal_power_setpoint", "recommended_actions", "estimated_savings", "confidence"]
|
||
|
|
},
|
||
|
|
"production_optimization": {
|
||
|
|
"name": "Production Efficiency Optimizer",
|
||
|
|
"description": "Maximizes production efficiency and throughput",
|
||
|
|
"parameters": ["raw_material_quality", "machine_utilization", "operator_skill"],
|
||
|
|
"outputs": ["optimal_production_rate", "efficiency_gain", "recommended_adjustments"]
|
||
|
|
},
|
||
|
|
"cost_optimization": {
|
||
|
|
"name": "Cost Reduction Optimizer",
|
||
|
|
"description": "Minimizes operational costs while maintaining quality",
|
||
|
|
"parameters": ["energy_cost", "labor_cost", "maintenance_cost"],
|
||
|
|
"outputs": ["optimal_cost_structure", "cost_reduction", "implementation_plan"]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
# Optimization history
|
||
|
|
optimization_history = []
|
||
|
|
|
||
|
|
def generate_optimization_id():
|
||
|
|
"""Generate unique optimization ID"""
|
||
|
|
return f"OPT_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{random.randint(1000, 9999)}"
|
||
|
|
|
||
|
|
def run_energy_optimization(data):
|
||
|
|
"""Run energy optimization"""
|
||
|
|
power_load = data.get("power_load", 450.0)
|
||
|
|
time_of_day = data.get("time_of_day", 12)
|
||
|
|
production_rate = data.get("production_rate", 95.0)
|
||
|
|
|
||
|
|
# Mock optimization logic
|
||
|
|
if time_of_day >= 6 and time_of_day <= 18:
|
||
|
|
# Daytime optimization
|
||
|
|
optimal_power = power_load * 0.85 # Reduce power during peak hours
|
||
|
|
savings = power_load * 0.15 * 0.12 # 15% reduction at $0.12/kWh
|
||
|
|
else:
|
||
|
|
# Nighttime optimization
|
||
|
|
optimal_power = power_load * 0.95 # Slight reduction at night
|
||
|
|
savings = power_load * 0.05 * 0.08 # 5% reduction at $0.08/kWh
|
||
|
|
|
||
|
|
# Adjust based on production rate
|
||
|
|
if production_rate < 80:
|
||
|
|
optimal_power *= 0.9
|
||
|
|
savings *= 1.1
|
||
|
|
elif production_rate > 95:
|
||
|
|
optimal_power *= 1.05
|
||
|
|
savings *= 0.9
|
||
|
|
|
||
|
|
return {
|
||
|
|
"optimal_power_setpoint": round(optimal_power, 2),
|
||
|
|
"recommended_actions": [
|
||
|
|
"Adjust compressor speed to 85%",
|
||
|
|
"Optimize pump sequencing",
|
||
|
|
"Implement variable frequency drive"
|
||
|
|
],
|
||
|
|
"estimated_savings": round(savings, 2),
|
||
|
|
"confidence": random.uniform(0.85, 0.95)
|
||
|
|
}
|
||
|
|
|
||
|
|
def run_production_optimization(data):
|
||
|
|
"""Run production optimization"""
|
||
|
|
raw_material_quality = data.get("raw_material_quality", 85.0)
|
||
|
|
machine_utilization = data.get("machine_utilization", 92.0)
|
||
|
|
operator_skill = data.get("operator_skill", 88.0)
|
||
|
|
|
||
|
|
# Mock optimization logic
|
||
|
|
base_rate = 100.0
|
||
|
|
|
||
|
|
# Adjust based on factors
|
||
|
|
quality_factor = raw_material_quality / 100.0
|
||
|
|
utilization_factor = machine_utilization / 100.0
|
||
|
|
skill_factor = operator_skill / 100.0
|
||
|
|
|
||
|
|
optimal_rate = base_rate * quality_factor * utilization_factor * skill_factor
|
||
|
|
efficiency_gain = (optimal_rate - base_rate) / base_rate * 100
|
||
|
|
|
||
|
|
return {
|
||
|
|
"optimal_production_rate": round(optimal_rate, 2),
|
||
|
|
"efficiency_gain": round(efficiency_gain, 2),
|
||
|
|
"recommended_adjustments": [
|
||
|
|
"Increase raw material inspection frequency",
|
||
|
|
"Optimize machine maintenance schedule",
|
||
|
|
"Provide operator training on efficiency techniques"
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
def run_cost_optimization(data):
|
||
|
|
"""Run cost optimization"""
|
||
|
|
energy_cost = data.get("energy_cost", 55.0)
|
||
|
|
labor_cost = data.get("labor_cost", 30.0)
|
||
|
|
maintenance_cost = data.get("maintenance_cost", 15.0)
|
||
|
|
|
||
|
|
total_cost = energy_cost + labor_cost + maintenance_cost
|
||
|
|
|
||
|
|
# Mock optimization logic
|
||
|
|
optimal_energy = energy_cost * 0.9 # 10% reduction
|
||
|
|
optimal_labor = labor_cost * 0.95 # 5% reduction
|
||
|
|
optimal_maintenance = maintenance_cost * 1.05 # 5% increase for better maintenance
|
||
|
|
|
||
|
|
optimal_total = optimal_energy + optimal_labor + optimal_maintenance
|
||
|
|
cost_reduction = total_cost - optimal_total
|
||
|
|
|
||
|
|
return {
|
||
|
|
"optimal_cost_structure": {
|
||
|
|
"energy": round(optimal_energy, 2),
|
||
|
|
"labor": round(optimal_labor, 2),
|
||
|
|
"maintenance": round(optimal_maintenance, 2),
|
||
|
|
"total": round(optimal_total, 2)
|
||
|
|
},
|
||
|
|
"cost_reduction": round(cost_reduction, 2),
|
||
|
|
"implementation_plan": [
|
||
|
|
"Implement energy monitoring system",
|
||
|
|
"Optimize workforce scheduling",
|
||
|
|
"Increase preventive maintenance frequency"
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
def generate_forecast(hours=24):
|
||
|
|
"""Generate forecast data"""
|
||
|
|
forecast = []
|
||
|
|
base_time = datetime.now()
|
||
|
|
|
||
|
|
for i in range(hours):
|
||
|
|
timestamp = base_time + timedelta(hours=i)
|
||
|
|
hour = timestamp.hour
|
||
|
|
|
||
|
|
# Generate realistic forecast data
|
||
|
|
if hour >= 6 and hour <= 18:
|
||
|
|
# Daytime values
|
||
|
|
energy_consumption = random.uniform(450, 550)
|
||
|
|
production_rate = random.uniform(95, 105)
|
||
|
|
efficiency = random.uniform(85, 92)
|
||
|
|
cost = random.uniform(50, 60)
|
||
|
|
else:
|
||
|
|
# Nighttime values
|
||
|
|
energy_consumption = random.uniform(350, 450)
|
||
|
|
production_rate = random.uniform(80, 95)
|
||
|
|
efficiency = random.uniform(78, 85)
|
||
|
|
cost = random.uniform(40, 50)
|
||
|
|
|
||
|
|
forecast.append({
|
||
|
|
"timestamp": timestamp.isoformat(),
|
||
|
|
"energy_consumption": round(energy_consumption, 2),
|
||
|
|
"production_rate": round(production_rate, 2),
|
||
|
|
"efficiency": round(efficiency, 2),
|
||
|
|
"cost": round(cost, 2)
|
||
|
|
})
|
||
|
|
|
||
|
|
return forecast
|
||
|
|
|
||
|
|
@app.route('/health', methods=['GET'])
|
||
|
|
def health():
|
||
|
|
"""Health check endpoint"""
|
||
|
|
return jsonify({
|
||
|
|
"status": "healthy",
|
||
|
|
"service": "mock-optimizer",
|
||
|
|
"timestamp": datetime.now().isoformat(),
|
||
|
|
"models_available": len(optimization_models)
|
||
|
|
})
|
||
|
|
|
||
|
|
@app.route('/api/v1/models', methods=['GET'])
|
||
|
|
def get_models():
|
||
|
|
"""Get available optimization models"""
|
||
|
|
return jsonify({
|
||
|
|
"models": optimization_models,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
})
|
||
|
|
|
||
|
|
@app.route('/api/v1/optimize/<model_name>', methods=['POST'])
|
||
|
|
def run_optimization(model_name):
|
||
|
|
"""Run optimization with specified model"""
|
||
|
|
if model_name not in optimization_models:
|
||
|
|
return jsonify({"error": f"Model '{model_name}' not found"}), 404
|
||
|
|
|
||
|
|
data = request.json
|
||
|
|
if not data:
|
||
|
|
return jsonify({"error": "No data provided"}), 400
|
||
|
|
|
||
|
|
start_time = time.time()
|
||
|
|
|
||
|
|
# Run appropriate optimization
|
||
|
|
if model_name == "energy_optimization":
|
||
|
|
result = run_energy_optimization(data)
|
||
|
|
elif model_name == "production_optimization":
|
||
|
|
result = run_production_optimization(data)
|
||
|
|
elif model_name == "cost_optimization":
|
||
|
|
result = run_cost_optimization(data)
|
||
|
|
else:
|
||
|
|
return jsonify({"error": "Model not implemented"}), 501
|
||
|
|
|
||
|
|
processing_time = round(time.time() - start_time, 3)
|
||
|
|
optimization_id = generate_optimization_id()
|
||
|
|
|
||
|
|
# Add to history
|
||
|
|
optimization_history.append({
|
||
|
|
"optimization_id": optimization_id,
|
||
|
|
"model": model_name,
|
||
|
|
"timestamp": datetime.now().isoformat(),
|
||
|
|
"processing_time": processing_time
|
||
|
|
})
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
"optimization_id": optimization_id,
|
||
|
|
"model": model_name,
|
||
|
|
"result": result,
|
||
|
|
"processing_time": processing_time,
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
})
|
||
|
|
|
||
|
|
@app.route('/api/v1/history', methods=['GET'])
|
||
|
|
def get_history():
|
||
|
|
"""Get optimization history"""
|
||
|
|
return jsonify({
|
||
|
|
"history": optimization_history[-10:], # Last 10 optimizations
|
||
|
|
"total_optimizations": len(optimization_history),
|
||
|
|
"timestamp": datetime.now().isoformat()
|
||
|
|
})
|
||
|
|
|
||
|
|
@app.route('/api/v1/forecast', methods=['POST'])
|
||
|
|
def generate_optimization_forecast():
|
||
|
|
"""Generate optimization forecast"""
|
||
|
|
data = request.json or {}
|
||
|
|
hours = data.get("hours", 24)
|
||
|
|
|
||
|
|
if hours > 168: # Max 1 week
|
||
|
|
return jsonify({"error": "Forecast horizon too long. Maximum 168 hours (1 week)"}), 400
|
||
|
|
|
||
|
|
forecast = generate_forecast(hours)
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
"forecast": forecast,
|
||
|
|
"generated_at": datetime.now().isoformat(),
|
||
|
|
"horizon_hours": hours
|
||
|
|
})
|
||
|
|
|
||
|
|
@app.route('/api/v1/status/<optimization_id>', methods=['GET'])
|
||
|
|
def get_optimization_status(optimization_id):
|
||
|
|
"""Get status of specific optimization"""
|
||
|
|
for opt in optimization_history:
|
||
|
|
if opt["optimization_id"] == optimization_id:
|
||
|
|
return jsonify({
|
||
|
|
"optimization_id": optimization_id,
|
||
|
|
"status": "completed",
|
||
|
|
"model": opt["model"],
|
||
|
|
"completed_at": opt["timestamp"],
|
||
|
|
"processing_time": opt["processing_time"]
|
||
|
|
})
|
||
|
|
|
||
|
|
return jsonify({"error": f"Optimization '{optimization_id}' not found"}), 404
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
print("🧠 Starting Mock Optimizer Server on port 8082...")
|
||
|
|
print("📊 Available endpoints:")
|
||
|
|
print(" GET /health")
|
||
|
|
print(" GET /api/v1/models")
|
||
|
|
print(" POST /api/v1/optimize/<model_name>")
|
||
|
|
print(" GET /api/v1/history")
|
||
|
|
print(" POST /api/v1/forecast")
|
||
|
|
print(" GET /api/v1/status/<optimization_id>")
|
||
|
|
|
||
|
|
app.run(host='0.0.0.0', port=8082, debug=False)
|