import os import sqlite3 import json import requests import random import time from flask import Flask, render_template, request, jsonify, g from flask_cors import CORS from werkzeug.utils import secure_filename app = Flask(__name__) CORS(app) # Configuration app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB app.config['UPLOAD_FOLDER'] = os.path.join(app.instance_path, 'uploads') SILICONFLOW_API_KEY = "sk-vimuseiptfbomzegyuvmebjzooncsqbyjtlddrfodzcdskgi" SILICONFLOW_API_URL = "https://api.siliconflow.cn/v1/chat/completions" # Ensure upload directory exists os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) # Error Handlers @app.errorhandler(413) def request_entity_too_large(error): return jsonify({'error': 'File too large. Maximum size is 16MB.'}), 413 @app.errorhandler(404) def not_found_error(error): return render_template('index.html'), 200 # SPA fallback or just ignore @app.errorhandler(500) def internal_error(error): return jsonify({'error': 'Internal Server Error'}), 500 # Database Setup def get_db(): db = getattr(g, '_database', None) if db is None: db_path = os.path.join(app.instance_path, 'orbit_ops.db') if not os.path.exists(app.instance_path): os.makedirs(app.instance_path) db = g._database = sqlite3.connect(db_path) db.row_factory = sqlite3.Row return db @app.teardown_appcontext def close_connection(exception): db = getattr(g, '_database', None) if db is not None: db.close() def init_db(): with app.app_context(): db = get_db() # Satellites Table db.execute(''' CREATE TABLE IF NOT EXISTS satellites ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, type TEXT NOT NULL, status TEXT NOT NULL, orbit_altitude INTEGER, inclination REAL, launch_date TEXT ) ''') # Missions Table db.execute(''' CREATE TABLE IF NOT EXISTS missions ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, description TEXT, status TEXT NOT NULL, target_date TEXT ) ''') # Seed Data if empty cur = db.execute('SELECT count(*) FROM satellites') if cur.fetchone()[0] == 0: satellites = [ ('StarLink-X1', 'Comm', 'Active', 550, 53.0, '2024-01-15'), ('Sentinel-Prime', 'EarthObs', 'Active', 700, 98.2, '2023-11-20'), ('Quantum-Relay', 'Comm', 'Testing', 1200, 45.0, '2025-02-01'), ('Debris-Hunter', 'Cleanup', 'Planned', 800, 85.0, '2026-06-10') ] db.executemany('INSERT INTO satellites (name, type, status, orbit_altitude, inclination, launch_date) VALUES (?, ?, ?, ?, ?, ?)', satellites) missions = [ ('Polar Orbit Insertion', 'Deploy Sentinel-Prime to polar orbit for ice monitoring.', 'Completed', '2023-11-20'), ('Constellation Expansion', 'Launch batch of 60 StarLink-X satellites.', 'Pending', '2026-03-15'), ('Debris Removal Demo', 'Test capture mechanism on defunct satellite.', 'Planning', '2026-08-01') ] db.executemany('INSERT INTO missions (name, description, status, target_date) VALUES (?, ?, ?, ?)', missions) db.commit() # Routes @app.route('/') def index(): return render_template('index.html') @app.route('/api/satellites', methods=['GET', 'POST']) def handle_satellites(): db = get_db() if request.method == 'POST': data = request.json db.execute('INSERT INTO satellites (name, type, status, orbit_altitude, inclination, launch_date) VALUES (?, ?, ?, ?, ?, ?)', (data['name'], data['type'], data['status'], data['orbit_altitude'], data['inclination'], data['launch_date'])) db.commit() return jsonify({'status': 'success'}) else: cur = db.execute('SELECT * FROM satellites') return jsonify([dict(row) for row in cur.fetchall()]) @app.route('/api/missions', methods=['GET', 'POST']) def handle_missions(): db = get_db() if request.method == 'POST': data = request.json db.execute('INSERT INTO missions (name, description, status, target_date) VALUES (?, ?, ?, ?)', (data['name'], data['description'], data['status'], data['target_date'])) db.commit() return jsonify({'status': 'success'}) else: cur = db.execute('SELECT * FROM missions') return jsonify([dict(row) for row in cur.fetchall()]) @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No selected file'}), 400 if file: filename = secure_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) # If it's a JSON file, try to import satellites or missions if filename.endswith('.json'): try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) db = get_db() if 'satellites' in data: for sat in data['satellites']: db.execute('INSERT INTO satellites (name, type, status, orbit_altitude, inclination, launch_date) VALUES (?, ?, ?, ?, ?, ?)', (sat.get('name'), sat.get('type'), sat.get('status'), sat.get('orbit_altitude'), sat.get('inclination'), sat.get('launch_date'))) if 'missions' in data: for mission in data['missions']: db.execute('INSERT INTO missions (name, description, status, target_date) VALUES (?, ?, ?, ?)', (mission.get('name'), mission.get('description'), mission.get('status'), mission.get('target_date'))) db.commit() return jsonify({'status': 'success', 'message': 'File uploaded and data imported successfully'}) except Exception as e: return jsonify({'status': 'warning', 'message': f'File uploaded but import failed: {str(e)}'}) return jsonify({'status': 'success', 'message': 'File uploaded successfully'}) @app.route('/api/telemetry') def get_telemetry(): # Mock real-time telemetry return jsonify({ 'timestamp': time.time(), 'signal_strength': random.uniform(80, 100), 'battery_level': random.uniform(90, 100), 'cpu_load': random.uniform(10, 40), 'temperature': random.uniform(20, 35) }) # Mock Response Generator def generate_mock_response(user_message): msg = user_message.lower() if "变轨" in msg or "maneuver" in msg or "规划" in msg: return """### 任务规划建议 (Mock Mode) 根据当前轨道参数，建议执行霍曼转移： 1. **Delta-V**: 12.5 m/s 2. **点火时间**: T+2h 30m 3. **目标轨道**: 550km -> 560km 请在任务规划页面确认执行。""" elif "状态" in msg or "status" in msg or "遥测" in msg: return """### 卫星状态报告 (Mock Mode) - **StarLink-X1**: 正常 (信号 98%) - **Sentinel-Prime**: 正常 (电池 95%) - **Quantum-Relay**: 警告 (温度偏高 45°C) 建议检查散热系统。""" elif "发射" in msg or "launch" in msg: return """### 发射窗口分析 (Mock Mode) 下一次最佳发射窗口： - **日期**: 2026-03-15 - **时间**: 14:30 UTC - **倾角**: 53.0° 天气状况良好，满足发射条件。""" else: return f"""### AI 助手 (Mock Mode) 我收到了你的指令："{user_message}"。 由于连接外部大模型超时，我正在使用本地应急模式。 我可以帮你： - 查询卫星状态 - 规划变轨任务 - 分析发射窗口 请尝试询问具体的技术参数。""" @app.route('/api/chat', methods=['POST']) def chat(): data = request.json user_message = data.get('message', '') headers = { "Authorization": f"Bearer {SILICONFLOW_API_KEY}", "Content-Type": "application/json" } payload = { "model": "Qwen/Qwen2.5-7B-Instruct", "messages": [ {"role": "system", "content": "你是 Orbit Ops 的 AI 任务规划助手。你是一个专业的航天工程师，负责协助用户进行卫星星座管理、轨道计算、发射任务规划和故障排查。请用中文回答，回答要专业、严谨，并在适当时提供 Markdown 格式的表格或列表。如果涉及数据分析，可以建议用户查看仪表盘。"}, {"role": "user", "content": user_message} ], "stream": False } try: # Reduced timeout to 5 seconds for better UX response = requests.post(SILICONFLOW_API_URL, json=payload, headers=headers, timeout=5) response.raise_for_status() result = response.json() content = result['choices'][0]['message']['content'] return jsonify({'response': content}) except Exception as e: print(f"API Error (using fallback): {e}") # Robust Mock Fallback mock_content = generate_mock_response(user_message) return jsonify({'response': mock_content}) if __name__ == '__main__': init_db() app.run(host='0.0.0.0', port=7866, debug=True)