import os import json import time import random import threading from datetime import datetime, timedelta from flask import Flask, render_template, jsonify, request, send_from_directory from werkzeug.utils import secure_filename from werkzeug.exceptions import RequestEntityTooLarge app = Flask(__name__) # --- Configuration --- app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 # 100MB limit app.config['UPLOAD_FOLDER'] = 'uploads' os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) # --- Core Memory / Storage --- DATA_FILE = 'maintenance_data.json' # --- Simulation Logic --- class EquipmentSimulator: def __init__(self, eq_id, name, type_, **kwargs): self.id = eq_id self.name = name self.type = type_ self.status = kwargs.get('status', 'Running') self.health_score = kwargs.get('health_score', 100.0) self.run_hours = kwargs.get('run_hours', 0.0) self.temperature = kwargs.get('temperature', 45.0) self.vibration = kwargs.get('vibration', 0.5) self.pressure = kwargs.get('pressure', 100.0) self.rpm = kwargs.get('rpm', 3000) self.last_update = time.time() self.maintenance_history = kwargs.get('maintenance_history', []) self.rul_prediction = kwargs.get('rul_prediction', 1000) def to_dict(self): return { 'id': self.id, 'name': self.name, 'type': self.type, 'status': self.status, 'health_score': self.health_score, 'run_hours': self.run_hours, 'temperature': self.temperature, 'vibration': self.vibration, 'pressure': self.pressure, 'rpm': self.rpm, 'maintenance_history': self.maintenance_history, 'rul_prediction': self.rul_prediction } def update(self): now = time.time() delta = (now - self.last_update) self.last_update = now if self.status == 'Stopped': self.temperature = max(25.0, self.temperature - 0.5) self.vibration = 0.0 self.rpm = 0 return # Simulate running accumulation sim_hours = delta * 1.0 self.run_hours += sim_hours # Degradation logic decay_rate = 0.01 if self.temperature > 80: decay_rate *= 2 if self.vibration > 2.0: decay_rate *= 3 self.health_score = max(0, self.health_score - (decay_rate * sim_hours)) # Sensor Simulation with Random Walk self.temperature += random.uniform(-0.5, 0.6) + (0.05 if self.health_score < 80 else 0) self.vibration += random.uniform(-0.05, 0.06) + (0.01 if self.health_score < 60 else 0) self.pressure += random.uniform(-1, 1) self.rpm += random.uniform(-10, 10) # Anomaly Injection (Random spikes) if random.random() < 0.05: self.vibration += 0.5 # Status Logic if self.health_score < 30: self.status = 'Critical' elif self.health_score < 70: self.status = 'Warning' else: self.status = 'Running' # RUL Calculation self.rul_prediction = self.health_score * 10 def perform_maintenance(self): self.health_score = 100.0 self.status = 'Running' self.temperature = 45.0 self.vibration = 0.5 self.maintenance_history.append({ 'date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'type': 'Preventive', 'cost': 500 }) # --- Persistence Helpers --- def load_data(): if os.path.exists(DATA_FILE): try: with open(DATA_FILE, 'r') as f: data = json.load(f) return { k: EquipmentSimulator(k, v['name'], v['type'], **v) for k, v in data.items() } except Exception as e: print(f"Error loading data: {e}") # Default data return { 'EQ-001': EquipmentSimulator('EQ-001', 'CNC-Alpha', 'CNC Machine'), 'EQ-002': EquipmentSimulator('EQ-002', 'Pump-Beta', 'Hydraulic Pump'), 'EQ-003': EquipmentSimulator('EQ-003', 'Turbine-Gamma', 'Gas Turbine'), 'EQ-004': EquipmentSimulator('EQ-004', 'Conveyor-Delta', 'Belt System'), 'EQ-005': EquipmentSimulator('EQ-005', 'Robot-Arm-X', 'Robotic Arm'), 'EQ-006': EquipmentSimulator('EQ-006', 'Generator-Z', 'Backup Generator') } def save_data(): try: data = {k: v.to_dict() for k, v in fleet.items()} with open(DATA_FILE, 'w') as f: json.dump(data, f, indent=2) except Exception as e: print(f"Error saving data: {e}") # Initialize Fleet fleet = load_data() # --- Routes --- @app.route('/') def index(): return render_template('index.html') @app.route('/api/status') def get_status(): data = [] total_cost = 0 risk_score = 0 for eq in fleet.values(): eq.update() data.append({ 'id': eq.id, 'name': eq.name, 'type': eq.type, 'status': eq.status, 'health': round(eq.health_score, 1), 'sensors': { 'temp': round(eq.temperature, 1), 'vibration': round(eq.vibration, 2), 'pressure': round(eq.pressure, 0), 'rpm': int(eq.rpm) }, 'rul': int(eq.rul_prediction), 'run_hours': round(eq.run_hours, 1) }) total_cost += sum(m['cost'] for m in eq.maintenance_history) if eq.health_score < 50: risk_score += 1 return jsonify({ 'fleet': data, 'timestamp': datetime.now().strftime('%H:%M:%S'), 'metrics': { 'total_maintenance_cost': total_cost, 'fleet_risk_level': 'High' if risk_score > 1 else 'Low' if risk_score == 0 else 'Medium' } }) @app.route('/api/maintain', methods=['POST']) def maintain_equipment(): eq_id = request.json.get('id') if eq_id in fleet: fleet[eq_id].perform_maintenance() save_data() # Save after maintenance return jsonify({'success': True, 'message': f'Maintenance performed on {eq_id}'}) return jsonify({'success': False, 'message': 'Equipment not found'}), 404 @app.route('/api/export', methods=['GET']) def export_data(): export_data = { 'generated_at': datetime.now().isoformat(), 'fleet_status': [ { 'id': eq.id, 'name': eq.name, 'history': eq.maintenance_history } for eq in fleet.values() ] } return jsonify(export_data) @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({'success': False, 'message': 'No file part'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'success': False, 'message': 'No selected file'}), 400 if file: filename = secure_filename(file.filename) file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename)) return jsonify({'success': True, 'message': f'File {filename} uploaded successfully'}) return jsonify({'success': False, 'message': 'Upload failed'}), 500 # --- Error Handlers --- @app.errorhandler(404) def not_found(e): if request.path.startswith('/api/'): return jsonify({'error': 'Not found'}), 404 return "Page Not Found", 404 @app.errorhandler(500) def server_error(e): if request.path.startswith('/api/'): return jsonify({'error': 'Internal Server Error'}), 500 return "Internal Server Error", 500 @app.errorhandler(RequestEntityTooLarge) def handle_file_too_large(e): return jsonify({'success': False, 'message': 'File is too large'}), 413 if __name__ == '__main__': port = int(os.environ.get('PORT', 7860)) app.run(host='0.0.0.0', port=port, debug=True)