diff --git a/src/UAV_navigation_system/drone_main.py b/src/UAV_navigation_system/drone_main.py new file mode 100644 index 0000000000..64aced148d --- /dev/null +++ b/src/UAV_navigation_system/drone_main.py @@ -0,0 +1,212 @@ +""" +无人机视觉导航系统 - 简化版 +可以立即运行测试 +""" + +import os +import sys +import time +import cv2 +import numpy as np + +# 添加项目路径，让Python能找到src模块 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +def ensure_directories(): + """确保所有目录都存在""" + dirs = ['data/images', 'data/videos', 'data/logs', 'data/config'] + for d in dirs: + os.makedirs(d, exist_ok=True) + print(f"✓ 目录已就绪: {d}") + + +class SimpleDroneCamera: + """简单的无人机摄像头类""" + + def __init__(self, camera_id=0): + self.camera_id = camera_id + self.cap = None + + def open(self): + """打开摄像头""" + print(f"尝试打开摄像头 {self.camera_id}...") + self.cap = cv2.VideoCapture(self.camera_id) + + if not self.cap.isOpened(): + print("⚠️ 无法打开物理摄像头，使用模拟模式") + return False + else: + print("✓ 摄像头已连接") + return True + + def read_frame(self): + """读取一帧""" + if self.cap and self.cap.isOpened(): + ret, frame = self.cap.read() + if ret: + return frame + + # 如果没有摄像头或读取失败，返回模拟图像 + return self.simulate_frame() + + def simulate_frame(self): + """生成模拟图像""" + width, height = 640, 480 + frame = np.zeros((height, width, 3), dtype=np.uint8) + + # 添加一些图形 + cv2.putText(frame, "无人机模拟视图", (50, 50), + cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) + cv2.rectangle(frame, (100, 100), (300, 300), (255, 0, 0), 2) + cv2.circle(frame, (400, 200), 50, (0, 0, 255), -1) + + return frame + + def release(self): + """释放摄像头""" + if self.cap: + self.cap.release() + print("摄像头已释放") + + +def analyze_scene_simple(frame): + """简单场景分析（基于颜色）""" + if frame is None: + return "未知", 0.5 + + # 转换为HSV + hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) + + # 检测绿色（植被） + green_mask = cv2.inRange(hsv, (35, 40, 40), (85, 255, 255)) + green_pct = np.sum(green_mask > 0) / green_mask.size + + # 检测蓝色（水域） + blue_mask = cv2.inRange(hsv, (100, 40, 40), (140, 255, 255)) + blue_pct = np.sum(blue_mask > 0) / blue_mask.size + + # 判断场景 + if green_pct > 0.3: + return "森林/草地", green_pct + elif blue_pct > 0.2: + return "水域", blue_pct + else: + return "城市/建筑", max(green_pct, blue_pct) + + +def get_decision(scene_type, confidence): + """根据场景类型做出决策""" + decisions = { + "森林/草地": "✓ 安全区域，继续飞行", + "水域": "⚠️ 接近水域，提高飞行高度", + "城市/建筑": "⚠️ 城市区域，降低速度并避让", + "未知": "? 无法识别，保持警戒" + } + return decisions.get(scene_type, "保持当前状态") + + +def main(): + """主函数""" + print("=" * 50) + print("无人机视觉导航系统") + print("版本: 1.0.0") + print("按 'q' 键退出，按 's' 键保存图像") + print("=" * 50) + + # 确保目录存在 + ensure_directories() + + # 创建无人机摄像头 + drone_cam = SimpleDroneCamera(camera_id=0) + drone_cam.open() + + # 初始化状态 + battery = 100 + flight_time = 0 + frame_count = 0 + start_time = time.time() + + print("\n开始飞行...") + + while True: + # 读取帧 + frame = drone_cam.read_frame() + frame_count += 1 + + # 分析场景 + scene_type, confidence = analyze_scene_simple(frame) + + # 获取决策 + decision = get_decision(scene_type, confidence) + + # 更新状态 + flight_time = time.time() - start_time + battery = max(0, battery - 0.05) # 慢慢消耗电池 + + # 在图像上显示信息 + info_lines = [ + f"场景: {scene_type} ({confidence:.1%})", + f"决策: {decision}", + f"飞行时间: {flight_time:.1f}s", + f"电池: {battery:.1f}%", + f"帧数: {frame_count}" + ] + + y_offset = 30 + for line in info_lines: + cv2.putText(frame, line, (10, y_offset), + cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 1) + y_offset += 25 + + # 显示图像 + cv2.imshow('无人机视觉导航', frame) + + # 检查按键 + key = cv2.waitKey(1) & 0xFF + if key == ord('q'): # 按 q 退出 + break + elif key == ord('s'): # 按 s 保存图像 + filename = f"data/images/capture_{frame_count}.jpg" + cv2.imwrite(filename, frame) + print(f"✓ 保存图像: {filename}") + + # 检查电池 + if battery <= 0: + print("\n⚠️ 电池耗尽！紧急降落...") + break + + # 限制运行时间（可选） + if flight_time > 60: # 60秒后自动停止 + print("\n⏰ 飞行时间到，安全降落...") + break + + # 清理 + drone_cam.release() + cv2.destroyAllWindows() + + # 显示统计信息 + print("\n" + "=" * 50) + print("飞行统计:") + print(f"- 总飞行时间: {flight_time:.1f} 秒") + print(f"- 处理帧数: {frame_count}") + print(f"- 平均帧率: {frame_count / flight_time:.1f} FPS" if flight_time > 0 else "- 平均帧率: N/A") + print(f"- 最终电池: {battery:.1f}%") + print("=" * 50) + + print("\n飞行结束！") + return 0 + + +if __name__ == "__main__": + try: + exit_code = main() + except KeyboardInterrupt: + print("\n程序被用户中断") + exit_code = 0 + except Exception as e: + print(f"\n程序出错: {e}") + exit_code = 1 + + input("\n按 Enter 键退出...") + sys.exit(exit_code) \ No newline at end of file diff --git a/src/UAV_navigation_system/drone_nav_with_vision.py b/src/UAV_navigation_system/drone_nav_with_vision.py index cfd02296cd..6e77d16970 100644 --- a/src/UAV_navigation_system/drone_nav_with_vision.py +++ b/src/UAV_navigation_system/drone_nav_with_vision.py @@ -1,63 +1,122 @@ -import cv2 -import numpy as np -from tensorflow.keras.models import load_model -from tensorflow.keras.preprocessing.image import img_to_array -import random # For simulating low battery and emergency conditions -import time # For the timeout condition - -# DroneBattery Class to manage battery +# Import necessary libraries +import cv2 # OpenCV for computer vision tasks (video capture, image processing) +import numpy as np # NumPy for numerical operations and array manipulation +from tensorflow.keras.models import load_model # Load pre-trained Keras model +from tensorflow.keras.preprocessing.image import img_to_array # Convert image to array for model input +import random # For simulating random emergency conditions (low battery, etc.) +import time # For time-related operations (timeouts, sleep) + + +# DroneBattery Class to manage battery state and operations class DroneBattery: + """Class to simulate and manage drone battery operations""" + def __init__(self, max_capacity=100, current_charge=100): + """ + Initialize battery parameters + Args: + max_capacity: Maximum battery capacity (default 100%) + current_charge: Current battery charge level (default 100%) + """ self.max_capacity = max_capacity self.current_charge = current_charge - + def display_battery_status(self): + """Display current battery percentage""" print(f"Battery Status: {self.current_charge}%") - + def charge_battery(self, charge_rate=10): + """ + Simulate battery charging process + Args: + charge_rate: Percentage to charge per iteration (default 10%) + """ while self.current_charge < self.max_capacity: self.current_charge += charge_rate if self.current_charge > self.max_capacity: self.current_charge = self.max_capacity print(f"Charging... {self.current_charge}%") - time.sleep(1) + time.sleep(1) # Simulate time delay for charging print("Battery fully charged!") - + def discharge_battery(self, discharge_rate=10): + """ + Simulate battery discharging process + Args: + discharge_rate: Percentage to discharge per iteration (default 10%) + """ while self.current_charge > 0: self.current_charge -= discharge_rate if self.current_charge < 0: self.current_charge = 0 print(f"Discharging... {self.current_charge}%") - time.sleep(1) + time.sleep(1) # Simulate time delay for discharging print("Battery completely drained!") - + def is_battery_low(self): + """ + Check if battery level is critically low + Returns: + Boolean: True if battery < 20%, False otherwise + """ return self.current_charge < 20 -# ✅ Load and compile model + +# Load and compile the pre-trained machine learning model print("📦 Loading model...") +# Load model from specified file path model = load_model(r"C:\Users\hp\DroneNavVision\data\best_model.h5") -model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) # ✅ Compile step added +# Compile model with optimizer and loss function for inference +model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) +# Define class names corresponding to model output categories class_names = ['Animal', 'City', 'Fire', 'Forest', 'Vehicle', 'Water'] + def check_emergency(): + """ + Simulate random emergency condition checks + Returns: + String: Randomly selected emergency condition + """ emergency_conditions = ['low_battery', 'emergency'] return random.choice(emergency_conditions) -# Handle low battery condition + def handle_low_battery(drone_battery): + """ + Handle low battery emergency procedure + Args: + drone_battery: DroneBattery instance to manage charging + """ print("🔋 Low battery! Returning to base.") - drone_battery.charge_battery(charge_rate=15) - exit() + drone_battery.charge_battery(charge_rate=15) # Fast charge at 15% per second + exit() # Exit program after charging + def preprocess_frame(frame): + """ + Preprocess video frame for model input + Args: + frame: Raw image frame from camera + Returns: + numpy array: Preprocessed image ready for model prediction + """ + # Resize frame to match model input dimensions (128x128) resized = cv2.resize(frame, (128, 128)) + # Normalize pixel values to [0, 1] and convert to array img_array = img_to_array(resized) / 255.0 + # Add batch dimension (1, 128, 128, 3) return np.expand_dims(img_array, axis=0) + def decide_navigation(predicted_class): + """ + Determine navigation action based on detected class + Args: + predicted_class: String of detected object/environment class + """ + # Define navigation strategies for each detected class if predicted_class == 'Fire': print("🔥 Fire detected! Navigate away.") elif predicted_class == 'Animal': @@ -73,62 +132,79 @@ def decide_navigation(predicted_class): else: print("✅ Clear path. Continue normal navigation.") + def main(): + """ + Main function to run drone vision system + Handles video capture, frame processing, prediction, and navigation decisions + """ print("🚁 Starting the drone vision process...") - start_time = time.time() - - # Use the correct video source based on your setup - VIDEO_SOURCE = "http://192.168.1.3:4747/video" # Or set your IP stream (e.g., "rtsp:///stream") + start_time = time.time() # Record start time for timeout condition - cap = cv2.VideoCapture(VIDEO_SOURCE, cv2.CAP_FFMPEG) - cap.set(cv2.CAP_PROP_BUFFERSIZE, 10) # Increase buffer size - cap.set(cv2.CAP_PROP_FPS, 30) # Set the desired FPS (adjust as needed) + # Video source configuration - adjust based on drone camera setup + # Options: RTSP stream, IP camera, or local camera + VIDEO_SOURCE = "http://192.168.1.3:4747/video" # IP camera stream URL + # Initialize video capture with FFMPEG backend for streaming + cap = cv2.VideoCapture(VIDEO_SOURCE, cv2.CAP_FFMPEG) + # Configure video capture properties + cap.set(cv2.CAP_PROP_BUFFERSIZE, 10) # Increase buffer size for smoother streaming + cap.set(cv2.CAP_PROP_FPS, 30) # Set frame rate to 30 FPS - # Check if video stream is opened successfully + # Validate video stream connection if not cap.isOpened(): print("❌ Failed to open video source. Check connection or URL.") return else: print("✅ Video source opened successfully.") - + # Initialize drone battery management drone_battery = DroneBattery() + # Main processing loop while True: print("📸 Processing frame...") - - # Check if battery is low + + # Emergency condition check: Low battery if drone_battery.is_battery_low(): handle_low_battery(drone_battery) + # Timeout condition: Stop after 5 minutes (300 seconds) elapsed_time = time.time() - start_time if elapsed_time > 300: print("⏰ Timeout reached! Stopping the drone.") break + # Capture frame from video stream ret, frame = cap.read() - if not ret: + if not ret: # Check if frame capture was successful print("❌ Failed to capture frame.") break - processed = preprocess_frame(frame) - pred = model.predict(processed) - predicted_class = class_names[np.argmax(pred)] - confidence = np.max(pred) * 100 + # Process frame and make prediction + processed = preprocess_frame(frame) # Preprocess for model + pred = model.predict(processed) # Run model inference + predicted_class = class_names[np.argmax(pred)] # Get class with highest probability + confidence = np.max(pred) * 100 # Convert confidence to percentage + # Display prediction on video feed cv2.putText(frame, f"{predicted_class} ({confidence:.2f}%)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) - cv2.imshow("Drone Vision Feed", frame) + cv2.imshow("Drone Vision Feed", frame) # Show annotated video + # Make navigation decision based on prediction decide_navigation(predicted_class) + # Exit condition: Press 'q' to quit if cv2.waitKey(1) & 0xFF == ord('q'): print("🛑 Manual stop initiated by the user.") break - cap.release() - cv2.destroyAllWindows() + # Cleanup resources + cap.release() # Release video capture device + cv2.destroyAllWindows() # Close all OpenCV windows + +# Standard Python idiom to run main function when script is executed directly if __name__ == "__main__": - main() + main() \ No newline at end of file