update

app.py

@@ -13,6 +13,7 @@ import sys
 import random
 import cv2
 from PIL import ImageDraw
+import requests
 
 
 # 设置日志记录
@@ -159,19 +160,67 @@ def get_image_base64(image):
         logger.error(f"Error encoding image: {e}")
         return None
 
-def capture_image():
-    """捕获当前打印图像"""
-    logger.info("Capturing print image")
+def get_test_image(image_name=None):
+    """获取测试图片"""
+    import os
+    import random
+    
+    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:
-        # 这里应该是实际的相机捕获代码
-        # 目前返回一个占位图像
-        logger.warning("Using placeholder image (camera not implemented)")
-        placeholder = Image.new('RGB', (300, 300), color=(200, 200, 200))
-        return placeholder
+        return Image.open(image_path)
     except Exception as e:
-        logger.error(f"Error capturing image: {e}")
+        logger.error(f"Failed to open test image: {e}")
         return None
 
+def capture_image(url=None, use_test_image=False, test_image_name=None):
+    """从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")
+    
+    # 尝试从URL获取图像
+    if url:
+        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))
+            else:
+                logger.error(f"Failed to get image from URL: {response.status_code}")
+        except Exception as e:
+            logger.error(f"Error capturing image from URL: {e}")
+    
+    # 如果URL获取失败，尝试使用随机测试图片
+    logger.info("URL capture failed or not provided, using random test image")
+    return get_test_image()
+
 def health_check():
     """健康检查端点"""
     status = {
@@ -233,47 +282,151 @@ with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
         logger.info("API call: get_data")
         return get_data()
     
-    def api_capture_frame():
-        """API端点：获取图像"""
-        logger.info("API call: capture_frame")
-        img = capture_image()
-        return get_image_base64(img)
+    def api_capture_frame(url=None, use_test_image=False, test_image_name=None):
+        """API端点：捕获图像帧"""
+        logger.info(f"API call: capture_frame with URL: {url}, use_test_image: {use_test_image}")
+        
+        try:
+            img = capture_image(url, use_test_image, test_image_name)
+            if img:
+                # 转换为base64
+                buffered = io.BytesIO()
+                img.save(buffered, format="JPEG")
+                img_str = base64.b64encode(buffered.getvalue()).decode()
+                
+                return {
+                    "success": True,
+                    "image": img_str
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Failed to capture image"
+                }
+        except Exception as e:
+            logger.error(f"Error in capture_frame: {e}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
     
-    def api_lambda(img_base64=None, param_1=200, param_2=60, param_3=60, param_4=100):
+    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):
         """API端点：分析图像"""
-        logger.info("API call: lambda")
+        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 img_base64:
+            
+            # 使用测试图片
+            if use_test_image:
+                logger.info(f"Lambda using test image: {test_image_name}")
+                img = get_test_image(test_image_name)
+            
+            # 检查输入是否是 URL
+            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)  # 从 URL 获取图像
+            
+            # 检查输入是否是 base64 数据
+            elif img_data and isinstance(img_data, str):
                 try:
-                    img_data = base64.b64decode(img_base64)
-                    img = Image.open(io.BytesIO(img_data))
-                except:
-                    logger.error("Failed to decode image")
+                    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:
-                img = capture_image()
-            
-            # 模拟分析结果
-            status, _, _, _ = get_data()
+                logger.info("No valid image data received, using default test image")
+                img = get_test_image()
             
-            # 返回分析结果
-            return {
-                "image": get_image_base64(img),
-                "missing_rate": 0.05,
-                "excess_rate": 0.03,
-                "stringing_rate": 0.02,
-                "uniformity_score": 0.9,
-                "print_quality_score": 0.85,
-                "print_speed_score": 0.7,
-                "material_efficiency_score": 0.8,
-                "total_performance_score": 0.8,
-                "printer_status": status
-            }
+            # 分析图像
+            if img:
+                # 转换为numpy数组
+                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
+                }
+                
+                # 将图像转换为base64并添加到结果中
+                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_api: {e}")
+            logger.error(f"Error in lambda: {e}")
             return {
                 "error": str(e)
             }
@@ -283,6 +436,55 @@ with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
         logger.info(f"API call: send_print_parameters with nozzle={nozzle_temp}, bed={bed_temp}, speed={print_speed}, fan={fan_speed}")
         return send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed)
 
+# 修改 API 端点注册部分
+demo.queue()
+
+# 正确注册 API 端点
+capture_frame_api = demo.load(
+    fn=api_capture_frame,
+    inputs=[
+        gr.Textbox(label="Image URL"),  # 接收图像 URL
+        gr.Checkbox(label="Use Test Image", value=False),
+        gr.Textbox(label="Test Image Name", value="")
+    ],
+    outputs="json",
+    api_name="capture_frame"
+)
+
+lambda_api = demo.load(
+    fn=api_lambda,
+    inputs=[
+        gr.Textbox(label="Image Base64 or URL"),  # 接收图像数据或 URL
+        gr.Number(label="Nozzle Temperature", value=200),
+        gr.Number(label="Bed Temperature", value=60),
+        gr.Number(label="Print Speed", value=60),
+        gr.Number(label="Fan Speed", value=100),
+        gr.Checkbox(label="Use Test Image", value=False),
+        gr.Textbox(label="Test Image Name", value="")
+    ],
+    outputs="json",
+    api_name="lambda"
+)
+
+get_data_api = demo.load(
+    fn=api_get_data,
+    inputs=None,
+    outputs="json",
+    api_name="get_data"
+)
+
+send_params_api = demo.load(
+    fn=api_send_print_parameters,
+    inputs=[
+        gr.Number(label="Nozzle Temperature", value=200),
+        gr.Number(label="Bed Temperature", value=60),
+        gr.Number(label="Print Speed", value=60),
+        gr.Number(label="Fan Speed", value=100)
+    ],
+    outputs="text",
+    api_name="send_print_parameters"
+)
+
 # 启动应用
 if __name__ == "__main__":
     logger.info("Starting Bambu A1 Mini Print Control application")