Update app.py

app.py

@@ -47,9 +47,8 @@ latest_data = {
     "status": "N/A",
     "update_time": "Waiting for data...",
     "image_url": "N/A",
-    "auto_capture_requested": False,
-    "last_capture_job": "None",
-    "last_capture_time": "None"
+    "progress": 0,  
+    "message": "",  
 }
 
 bambu_client = None
@@ -104,28 +103,41 @@ def rpi_on_message(client, userdata, msg):
         payload = msg.payload.decode("utf-8")
         logger.info(f"Received message from RPI: {payload}")
         
+        data = json.loads(payload)
+        status = data.get("status", "Unknown")
+        
+        latest_data["status"] = status
+        latest_data["update_time"] = data.get("update_time", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
+        
+        if "nozzle_temperature" in data:
+            latest_data["nozzle_temperature"] = data["nozzle_temperature"]
+        
+        if "bed_temperature" in data:
+            latest_data["bed_temperature"] = data["bed_temperature"]
+        
+        if "error" in data:
+            logger.error(f"Error from RPI: {data['error']}")
+       
+        if status == "Ready":
+            latest_data["progress"] = 0
+            latest_data["message"] = "Printer ready"
+        elif status == "Processing":
+            latest_data["progress"] = 25
+            latest_data["message"] = "Processing G-code..."
+        
+    except Exception as e:
+        logger.error(f"Error processing message from RPI: {e}")
+        
         try:
-            data = json.loads(payload)
+            result = json.loads(payload)
             
-            # 处理状态更新
-            if "status" in data:
-                latest_data["status"] = data["status"]
-                latest_data["update_time"] = data.get("update_time", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
-                
-                if "nozzle_temperature" in data:
-                    latest_data["nozzle_temperature"] = data["nozzle_temperature"]
-                
-                if "bed_temperature" in data:
-                    latest_data["bed_temperature"] = data["bed_temperature"]
-                
-                if "error" in data:
-                    logger.error(f"Error from RPI: {data['error']}")
-            
-            # 添加对自动拍照命令的处理
-            if "command" in data and data["command"] == "capture_image" and data.get("auto_triggered", False):
-                logger.info("收到自动拍照命令")
-                # 异步处理拍照请求，避免阻塞MQTT回调
-                threading.Thread(target=handle_auto_capture, args=(data,), daemon=True).start()
+            if "status" in result:
+                status = result["status"]
+                latest_data["status"] = status
+ 
+            if "command" in result and result["command"] == "capture_image" and result.get("auto_triggered", False):
+                logger.info("receive capture command")
+                threading.Thread(target=handle_auto_capture, args=(result,), daemon=True).start()
                 
         except json.JSONDecodeError:
             logger.error(f"Invalid JSON in message: {payload}")
@@ -203,6 +215,12 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
         logger.error(f"Error sending parameters: {e}")
         return f"Error sending parameters: {e}"
 
+      
+    latest_data["status"] = "Sending"
+    latest_data["progress"] = 10
+    latest_data["message"] = "Sending parameters to printer..."
+
+
 
 def get_image_base64(image):
     if image is None:
@@ -223,115 +241,85 @@ def get_image_base64(image):
         return None
 
 
-def get_test_image(image_name=None):
-    test_dir = os.path.join(os.path.dirname(__file__), "test_images")
-
-    if not os.path.exists(test_dir):
-        logger.error(f"Test images directory not found: {test_dir}")
-        return None
-
-    image_files = [
-        f
-        for f in os.listdir(test_dir)
-        if f.lower().endswith((".png", ".jpg", ".jpeg", ".bmp"))
-    ]
-
-    if not image_files:
-        logger.error("No test images found")
-        return None
-
-    if image_name and image_name in image_files:
-        image_path = os.path.join(test_dir, image_name)
-    else:
-        image_path = os.path.join(test_dir, random.choice(image_files))
-
-    logger.info(f"Using test image: {image_path}")
-
-    try:
-        return Image.open(image_path)
-    except Exception as e:
-        logger.error(f"Failed to open test image: {e}")
-        return None
-
-
-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
-
-
 def handle_auto_capture(message):
-    """处理自动拍照请求"""
     try:
-        logger.info(f"收到自动拍照请求: {message}")
+        logger.info(f"receive capture command: {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()
-        
-        # 保存图像
+        return capture_image()
+    except Exception as e:
+        logger.error(f"error: {e}")
+        return None, f"capture failed: {str(e)}"
+
+
+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"]
+
+    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":
         try:
-            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))
+            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))
             else:
-                image.save(filepath)
-                
-            logger.info(f"Saved captured image to {filepath}")
+                logger.error(f"Failed to get image from URL: {response.status_code}")
         except Exception as e:
-            logger.error(f"Error saving captured image: {e}")
-        
-        return image
-    except Exception as e:
-        logger.error(f"处理自动拍照请求时出错: {e}")
-        return None
+            logger.error(f"Error capturing image from URL: {e}")
+    else:
+        raise Exception("url is 'N/A'")
+
+def update_print_status():
+   global latest_data
+   return (
+       latest_data["status"],
+       latest_data["nozzle_temperature"],
+       latest_data["bed_temperature"],
+       latest_data["update_time"],
+       latest_data.get("progress", 0),
+       latest_data.get("message", ""),
+   )
 
 
 def health_check():
@@ -343,39 +331,17 @@ def health_check():
         "latest_update": latest_data["update_time"],
     }
     logger.info(f"Health check: {status}")
-    
-    # 修改返回值，确保返回6个值
-    return (
-        status.get("app", "unknown"),
-        "N/A",  # 作为nozzle_temperature
-        "N/A",  # 作为bed_temperature
-        status.get("time", "unknown"),
-        0,      # 作为progress
-        f"MQTT: Bambu={status.get('bambu_mqtt_connected')}, RPI={status.get('rpi_mqtt_connected')}"  # 作为message
-    )
-
-
-def update_print_status():
-    """更新打印状态显示"""
-    global latest_data
-    return (
-        latest_data["status"],
-        latest_data["nozzle_temperature"],
-        latest_data["bed_temperature"],
-        latest_data["update_time"],
-        latest_data.get("progress", 0),
-        latest_data.get("message", ""),
-    )
+    return status
 
 
 demo = gr.Blocks(title="Bambu A1 Mini Print Control")
 
 with demo:
     gr.Markdown("# Bambu A1 Mini Print Control")
-    
+
     with gr.Row():
         refresh_btn = gr.Button("Refresh Status")
-    
+
     with gr.Row():
         current_status = gr.Textbox(
             label="Printer Status", value="N/A", interactive=False
@@ -387,7 +353,7 @@ with demo:
             label="Current Nozzle Temperature", value="N/A", interactive=False
         )
         last_update = gr.Textbox(label="Last Update", value="N/A", interactive=False)
-    
+
     with gr.Row():
         # Left column for image and capture button
         with gr.Column(scale=2):
@@ -437,7 +403,7 @@ with demo:
         fn=get_data,
         outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update],
     )
-    
+
     capture_btn.click(
         fn=lambda: gr.update(interactive=False), outputs=capture_btn
     ).then(fn=capture_image, outputs=[captured_image]).then(
@@ -476,123 +442,6 @@ with demo:
             logger.error(f"Error in capture_frame: {e}")
             return {"success": False, "error": str(e)}
 
-    def api_lambda(
-        img_data=None,
-        param_1=200,
-        param_2=60,
-        param_3=60,
-        param_4=100,
-        use_test_image=False,
-        test_image_name=None,
-    ):
-        logger.info(
-            f"API call: lambda with params: {param_1}, {param_2}, {param_3}, {param_4}, use_test_image: {use_test_image}, test_image_name: {test_image_name}"
-        )
-        try:
-            img = None
-
-            if use_test_image:
-                logger.info(f"Lambda using test image: {test_image_name}")
-                img = get_test_image(test_image_name)
-
-            elif (
-                img_data
-                and isinstance(img_data, str)
-                and (img_data.startswith("http://") or img_data.startswith("https://"))
-            ):
-                logger.info(f"Lambda received image URL: {img_data}")
-                img = capture_image(img_data)
-
-            elif img_data and isinstance(img_data, str):
-                try:
-                    logger.info("Lambda received base64 image data")
-                    img_bytes = base64.b64decode(img_data)
-                    img = Image.open(io.BytesIO(img_bytes))
-                except Exception as e:
-                    logger.error(f"Failed to decode base64 image: {e}")
-
-            if img is None:
-                logger.info("No valid image data received, using default test image")
-                img = get_test_image()
-
-            if img:
-                img_array = np.array(img)
-
-                quality_level = "low"
-                if 190 <= param_1 <= 210 and param_3 <= 50 and param_4 >= 80:
-                    quality_level = "high"
-                elif 185 <= param_1 <= 215 and param_3 <= 70 and param_4 >= 60:
-                    quality_level = "medium"
-
-                if quality_level == "high":
-                    missing_rate = 0.02
-                    excess_rate = 0.01
-                    stringing_rate = 0.01
-                elif quality_level == "medium":
-                    missing_rate = 0.05
-                    excess_rate = 0.03
-                    stringing_rate = 0.02
-                else:  # low
-                    missing_rate = 0.10
-                    excess_rate = 0.07
-                    stringing_rate = 0.05
-
-                uniformity_score = 1.0 - (missing_rate + excess_rate + stringing_rate)
-
-                print_quality_score = 1.0 - (
-                    missing_rate * 2.0 + excess_rate * 1.5 + stringing_rate * 1.0
-                )
-                print_quality_score = max(0, min(1, print_quality_score))
-
-                print_speed_score = param_3 / 150.0
-                print_speed_score = max(0, min(1, print_speed_score))
-
-                material_efficiency_score = 1.0 - excess_rate * 3.0
-                material_efficiency_score = max(0, min(1, material_efficiency_score))
-
-                total_performance_score = (
-                    0.5 * print_quality_score
-                    + 0.3 * print_speed_score
-                    + 0.2 * material_efficiency_score
-                )
-
-                img_draw = img.copy()
-                draw = ImageDraw.Draw(img_draw)
-                draw.text(
-                    (10, 10), f"Quality: {quality_level.upper()}", fill=(255, 0, 0)
-                )
-                draw.text((10, 30), f"Missing: {missing_rate:.2f}", fill=(255, 0, 0))
-                draw.text((10, 50), f"Excess: {excess_rate:.2f}", fill=(255, 0, 0))
-                draw.text(
-                    (10, 70), f"Stringing: {stringing_rate:.2f}", fill=(255, 0, 0)
-                )
-
-                result = {
-                    "success": True,
-                    "missing_rate": missing_rate,
-                    "excess_rate": excess_rate,
-                    "stringing_rate": stringing_rate,
-                    "uniformity_score": uniformity_score,
-                    "print_quality_score": print_quality_score,
-                    "print_speed_score": print_speed_score,
-                    "material_efficiency_score": material_efficiency_score,
-                    "total_performance_score": total_performance_score,
-                }
-
-                if img_draw.mode == "RGBA":
-                    img_draw = img_draw.convert("RGB")
-
-                buffered = io.BytesIO()
-                img_draw.save(buffered, format="JPEG")
-                img_str = base64.b64encode(buffered.getvalue()).decode()
-                result["image"] = img_str
-
-                return result
-            else:
-                return {"success": False, "error": "Failed to get image"}
-        except Exception as e:
-            logger.error(f"Error in lambda: {e}")
-            return {"error": str(e)}
 
     def api_send_print_parameters(
         nozzle_temp=200, bed_temp=60, print_speed=60, fan_speed=100
@@ -668,52 +517,6 @@ with demo:
         api_name="health_check",
     )
 
-    # 在Camera Control标签页中添加自动拍照状态显示
-    with gr.Tab("Camera Control"):
-        # 添加自动拍照状态显示
-        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})"
-        
-        # 连接刷新按钮
-        refresh_status_btn.click(
-            fn=update_capture_status,
-            inputs=[],
-            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")