Update app.py

app.py

@@ -254,57 +254,84 @@ def get_test_image(image_name=None):
         return None
 
 
-def capture_image(url=None, use_test_image=False, test_image_name=None):
-    global rpi_client, latest_data
-
-    if rpi_client is None:
-        create_client("rpi")
-
-    serial = DEFAULT_SERIAL
-    request_topic = f"bambu_a1_mini/request/{serial}"
-    response_topic = f"bambu_a1_mini/response/{serial}"
-
-    logger.info(f"Subscribing to {response_topic}")
-    rpi_client.subscribe(response_topic)
-
-    rpi_client.publish(
-        request_topic,
-        json.dumps(
-            {
-                "command": "capture_image",
-            }
-        ),
-    )
-
-    latest_data["image_url"] = "N/A"
-    timeout = 45
-    while latest_data["image_url"] == "N/A" and timeout > 0:
-        # print("timeout", timeout)
-        time.sleep(1)
-        timeout -= 1
-
-    url = latest_data["image_url"]
+def capture_image():
+    """捕获图像"""
+    try:
+        logger.info("Capturing image...")
+        
+        # 尝试使用摄像头捕获图像
+        try:
+            cap = cv2.VideoCapture(0)  # 使用默认摄像头
+            ret, frame = cap.read()
+            cap.release()
+            
+            if ret:
+                # 转换为RGB（OpenCV使用BGR）
+                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                logger.info(f"Image captured successfully: {frame_rgb.shape}")
+                return frame_rgb
+            else:
+                logger.error("Failed to capture image from camera")
+        except Exception as e:
+            logger.error(f"Error accessing camera: {e}")
+        
+        # 如果摄像头捕获失败，使用测试图像
+        test_image = get_test_image()
+        if test_image is not None:
+            logger.info("Using test image instead")
+            return np.array(test_image)
+        
+        # 如果测试图像也不可用，创建一个错误图像
+        error_img = np.zeros((480, 640, 3), dtype=np.uint8)
+        cv2.putText(error_img, "Camera Error", (50, 240), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
+        return error_img
+        
+    except Exception as e:
+        logger.error(f"Error in capture_image: {e}")
+        # 返回错误图像
+        error_img = np.zeros((480, 640, 3), dtype=np.uint8)
+        cv2.putText(error_img, f"Error: {str(e)}", (50, 240), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
+        return error_img
 
-    if use_test_image:
-        logger.info("Using test image instead of URL")
-        test_img = get_test_image(test_image_name)
-        if test_img:
-            return test_img
-        else:
-            logger.warning("Failed to get specified test image, trying URL")
 
-    if url != "N/A":
+def handle_auto_capture(message):
+    """处理自动拍照请求"""
+    try:
+        logger.info(f"收到自动拍照请求: {message}")
+        
+        # 提取信息
+        print_job = message.get("print_job", "unknown_job")
+        timestamp = message.get("timestamp", time.strftime("%Y-%m-%d %H:%M:%S"))
+        
+        # 记录到最新数据中
+        latest_data["auto_capture_requested"] = True
+        latest_data["last_capture_job"] = print_job
+        latest_data["last_capture_time"] = timestamp
+        
+        # 执行拍照操作
+        image = capture_image()
+        
+        # 保存图像
         try:
-            logger.info(f"Capturing image from URL: {url}")
-            response = requests.get(url, timeout=10)
-            if response.status_code == 200:
-                return Image.open(io.BytesIO(response.content))
+            save_dir = os.path.join(os.path.dirname(__file__), "captured_images")
+            os.makedirs(save_dir, exist_ok=True)
+            
+            filename = f"{print_job}_{timestamp.replace(' ', '_').replace(':', '-')}.jpg"
+            filepath = os.path.join(save_dir, filename)
+            
+            if isinstance(image, np.ndarray):
+                cv2.imwrite(filepath, cv2.cvtColor(image, cv2.COLOR_RGB2BGR))
             else:
-                logger.error(f"Failed to get image from URL: {response.status_code}")
+                image.save(filepath)
+                
+            logger.info(f"Saved captured image to {filepath}")
         except Exception as e:
-            logger.error(f"Error capturing image from URL: {e}")
-    else:
-        raise Exception("url is 'N/A'")
+            logger.error(f"Error saving captured image: {e}")
+        
+        return image
+    except Exception as e:
+        logger.error(f"处理自动拍照请求时出错: {e}")
+        return None
 
 
 def health_check():
@@ -341,32 +368,6 @@ def update_print_status():
     )
 
 
-def handle_auto_capture(message):
-    """处理自动拍照请求"""
-    try:
-        logger.info(f"收到自动拍照请求: {message}")
-        
-        # 提取信息
-        print_job = message.get("print_job", "unknown_job")
-        timestamp = message.get("timestamp", time.strftime("%Y-%m-%d %H:%M:%S"))
-        
-        # 记录到最新数据���
-        latest_data["auto_capture_requested"] = True
-        latest_data["last_capture_job"] = print_job
-        latest_data["last_capture_time"] = timestamp
-        
-        # 执行拍照操作 - 假设capture_image函数已存在
-        image, status = capture_image()
-        
-        # 记录拍照结果
-        logger.info(f"自动拍照结果: {status}")
-        
-        return image, status
-    except Exception as e:
-        logger.error(f"处理自动拍照请求时出错: {e}")
-        return None, f"自动拍照失败: {str(e)}"
-
-
 demo = gr.Blocks(title="Bambu A1 Mini Print Control")
 
 with demo:
@@ -672,20 +673,46 @@ with demo:
         # 添加自动拍照状态显示
         with gr.Row():
             auto_capture_status = gr.Textbox(label="Auto Capture Status", value="Waiting")
+            refresh_status_btn = gr.Button("Refresh Status")
         
-        # 添加自动刷新，更新自动拍照状态
+        # 添加手动拍照按钮和显示区域
+        with gr.Row():
+            capture_btn = gr.Button("Capture Image")
+        
+        with gr.Row():
+            captured_image = gr.Image(label="Captured Image", type="numpy")
+        
+        # 定义更新状态函数
         def update_capture_status():
             auto_status = "Triggered" if latest_data.get("auto_capture_requested", False) else "Waiting"
             last_time = latest_data.get("last_capture_time", "None")
             last_job = latest_data.get("last_capture_job", "None")
             return f"{auto_status} (Last: {last_job} at {last_time})"
         
-        demo.load(
+        # 连接刷新按钮
+        refresh_status_btn.click(
             fn=update_capture_status,
             inputs=[],
-            outputs=[auto_capture_status],
-            every=5  # 每5秒刷新一次
+            outputs=[auto_capture_status]
         )
+        
+        # 连接拍照按钮
+        capture_btn.click(
+            fn=capture_image,
+            inputs=[],
+            outputs=[captured_image]
+        )
+        
+        # 使用JavaScript实现自动刷新
+        auto_refresh_js = """
+        function autoRefresh() {
+            document.querySelector('#refresh-status-btn').click();
+            setTimeout(autoRefresh, 5000);  // 每5秒刷新一次
+        }
+        setTimeout(autoRefresh, 5000);  // 启动自动刷新
+        """
+        
+        gr.HTML(f"<script>{auto_refresh_js}</script>")
 
 if __name__ == "__main__":
     logger.info("Starting Bambu A1 Mini Print Control application")