Update app.py

app.py

@@ -16,7 +16,6 @@ from PIL import ImageDraw
 import requests
 
 
-# 设置日志记录
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
@@ -27,13 +26,12 @@ logging.basicConfig(
 )
 logger = logging.getLogger("bambu-analysis")
 
-HOST = "mqtt.bambulab.com"  # 默认值
-PORT = 8883  # 默认值
-USERNAME = "bblp"  # 默认值
-PASSWORD = "bblp"  # 默认值
+HOST = "mqtt.bambulab.com"  
+PORT = 8883  
+USERNAME = "bblp"  
+PASSWORD = "bblp"  
 DEFAULT_SERIAL = "0309CA471800852"
 
-# 尝试从环境变量获取
 if os.environ.get("host"):
     HOST = os.environ.get("host")
 if os.environ.get("port"):
@@ -56,7 +54,6 @@ client = None
 response_topic = None  # Will be set dynamically
 
 def create_client(host, port, username, password):
-    """完全使用同事的创建客户端函数"""
     global client
     client = mqtt.Client()
     client.username_pw_set(username, password)
@@ -93,7 +90,6 @@ def get_data(serial=DEFAULT_SERIAL):
     logger.info(f"Subscribing to {response_topic}")
     client.subscribe(response_topic)
     
-    # 发送请求获取数据
     logger.info(f"Publishing request to {request_topic}")
     client.publish(request_topic, json.dumps("HI"))
     
@@ -112,7 +108,6 @@ def get_data(serial=DEFAULT_SERIAL):
     )
 
 def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
-    """发送打印参数到打印机"""
     serial = DEFAULT_SERIAL
     logger.info(f"Sending parameters to {serial}: nozzle={nozzle_temp}, bed={bed_temp}, speed={print_speed}, fan={fan_speed}")
     try:
@@ -140,17 +135,14 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
         return f"Error sending parameters: {e}"
 
 def get_image_base64(image):
-    """将图像转换为base64用于API传输"""
     if image is None:
         logger.warning("No image to encode")
         return None
     
     try:
-        # 转换为PIL图像
         if isinstance(image, np.ndarray):
             image = Image.fromarray(image)
         
-        # 转换为base64
         buffer = io.BytesIO()
         image.save(buffer, format="PNG")
         img_str = base64.b64encode(buffer.getvalue()).decode('utf-8')
@@ -161,7 +153,6 @@ def get_image_base64(image):
         return None
 
 def get_test_image(image_name=None):
-    """获取测试图片"""
     import os
     import random
     
@@ -171,7 +162,6 @@ def get_test_image(image_name=None):
         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'))]
     
@@ -179,11 +169,9 @@ def get_test_image(image_name=None):
         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}")
@@ -195,8 +183,7 @@ def get_test_image(image_name=None):
         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)
@@ -205,7 +192,7 @@ def capture_image(url=None, use_test_image=False, test_image_name=None):
         else:
             logger.warning("Failed to get specified test image, trying URL")
     
-    # 尝试从URL获取图像
+
     if url:
         try:
             logger.info(f"Capturing image from URL: {url}")
@@ -217,12 +204,10 @@ def capture_image(url=None, use_test_image=False, test_image_name=None):
         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 = {
         "app": "running",
         "time": time.strftime("%Y-%m-%d %H:%M:%S"),
@@ -232,10 +217,8 @@ def health_check():
     logger.info(f"Health check: {status}")
     return status
 
-# 创建 Gradio 应用
 demo = gr.Blocks(title="Bambu A1 Mini Print Control")
 
-# 在 Blocks 上下文中注册所有内容
 with demo:
     gr.Markdown("# Bambu A1 Mini Print Control")
     
@@ -254,7 +237,6 @@ with demo:
 
     with gr.Row():
         with gr.Column():
-            # 打印参数输入
             nozzle_temp = gr.Slider(minimum=180, maximum=250, step=1, value=200, label="Nozzle Temperature (°C)")
             bed_temp = gr.Slider(minimum=40, maximum=100, step=1, value=60, label="Bed Temperature (°C)")
             print_speed = gr.Slider(minimum=20, maximum=150, step=1, value=60, label="Print Speed (mm/s)")
@@ -262,7 +244,6 @@ with demo:
             
             send_params_btn = gr.Button("Send Print Parameters")
 
-    # 连接按钮
     refresh_btn.click(
         fn=get_data,
         outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update]
@@ -279,24 +260,19 @@ with demo:
         outputs=[current_status]
     )
     
-    # API端点
     def api_get_data():
-        """API端点：获取状态"""
         logger.info("API call: get_data")
         return get_data()
     
     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:
-                # 确保图像是 RGB 格式（去除 Alpha 通道）
                 if img.mode == 'RGBA':
                     img = img.convert('RGB')
                 
-                # 转换为base64
                 buffered = io.BytesIO()
                 img.save(buffered, format="JPEG")
                 img_str = base64.b64encode(buffered.getvalue()).decode()
@@ -318,23 +294,18 @@ with demo:
             }
     
     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(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)
             
-            # 检查输入是否是 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 获取图像
+                img = capture_image(img_data)  
             
-            # 检查输入是否是 base64 数据
             elif img_data and isinstance(img_data, str):
                 try:
                     logger.info("Lambda received base64 image data")
@@ -343,28 +314,19 @@ with demo:
                 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:
-                # 转换为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
@@ -378,10 +340,8 @@ with demo:
                     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))
                 
@@ -397,9 +357,7 @@ with demo:
                     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))
@@ -407,7 +365,6 @@ with demo:
                 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,
@@ -420,7 +377,7 @@ with demo:
                     "total_performance_score": total_performance_score
                 }
                 
-                # 将图像转换为base64并添加到结果中
+
                 if img_draw.mode == 'RGBA':
                     img_draw = img_draw.convert('RGB')
 
@@ -442,15 +399,12 @@ with demo:
             }
     
     def api_send_print_parameters(nozzle_temp=200, bed_temp=60, print_speed=60, fan_speed=100):
-        """API端点：发送打印参数"""
         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 输出的 JSON 组件
     api_json_output = gr.JSON()
     api_text_output = gr.Textbox()
     
-    # 注册 API 端点，使用 Gradio 组件作为输出
     capture_frame_api = demo.load(
         fn=api_capture_frame,
         inputs=[
@@ -458,7 +412,7 @@ with demo:
             gr.Checkbox(label="Use Test Image", value=False),
             gr.Textbox(label="Test Image Name", value="")
         ],
-        outputs=api_json_output,  # 使用 JSON 组件而不是字符串 "json"
+        outputs=api_json_output,  
         api_name="capture_frame"
     )
     
@@ -473,14 +427,14 @@ with demo:
             gr.Checkbox(label="Use Test Image", value=False),
             gr.Textbox(label="Test Image Name", value="")
         ],
-        outputs=api_json_output,  # 使用 JSON 组件而不是字符串 "json"
+        outputs=api_json_output, 
         api_name="lambda"
     )
     
     get_data_api = demo.load(
         fn=api_get_data,
         inputs=None,
-        outputs=api_json_output,  # 使用 JSON 组件而不是字符串 "json"
+        outputs=api_json_output, 
         api_name="get_data"
     )
     
@@ -492,22 +446,19 @@ with demo:
             gr.Number(label="Print Speed", value=60),
             gr.Number(label="Fan Speed", value=100)
         ],
-        outputs=api_text_output,  # 使用 Textbox 组件而不是字符串 "text"
+        outputs=api_text_output,  
         api_name="send_print_parameters"
     )
 
-# 启动应用
 if __name__ == "__main__":
     logger.info("Starting Bambu A1 Mini Print Control application")
     
-    # 尝试初始化MQTT连接
     try:
         logger.info("Initializing MQTT client")
         create_client(HOST, PORT, USERNAME, PASSWORD)
     except Exception as e:
         logger.error(f"Failed to initialize MQTT: {e}")
     
-    # 启动应用
     demo.queue().launch(
         show_error=True,
         share=False,