update

app.py

@@ -10,8 +10,11 @@ import io
 import numpy as np
 import logging
 import sys
+import random
+import cv2
+from PIL import ImageDraw
 
-# 设置详细的日志记录
+# 
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
@@ -22,28 +25,37 @@ logging.basicConfig(
 )
 logger = logging.getLogger("bambu-analysis")
 
-# Environment Variables - 使用同事的配置方式
+# Environment Variables
 HOST = "mqtt.bambulab.com"
 PORT = 8883
 USERNAME = "bblp"
 PASSWORD = "bblp"
 DEFAULT_SERIAL = "0309CA471800852"
 
-# Global variables to store received data - 完全保持同事的数据结构
+# 
+SIMULATION_MODE = True  # 默认启用模拟模式
+simulated_data = {
+    "bed_temperature": "60",
+    "nozzle_temperature": "200",
+    "status": "Printing",
+    "update_time": time.strftime("%Y-%m-%d %H:%M:%S")
+}
+
+# Global variables
 latest_data = {
     "bed_temperature": "N/A",
     "nozzle_temperature": "N/A",
     "status": "N/A",
-    "update_time": "Waiting for data...",
+    "update_time": "Waiting for data..."
 }
 
-# MQTT Client setup - 完全保持同事的客户端设置
+# MQTT Client setup
 client = None
-response_topic = None  # Will be set dynamically
+response_topic = None
 
 def create_client(host, port, username, password):
-    """创建并配置MQTT客户端 - 使用同事的代码"""
-    global client
+    """创建并配置MQTT客户端"""
+    global client, SIMULATION_MODE
     try:
         logger.info(f"Creating MQTT client for {host}:{port}")
         client = mqtt.Client()
@@ -53,19 +65,27 @@ def create_client(host, port, username, password):
         client.on_message = on_message
         
         logger.info("Connecting to MQTT broker...")
-        client.connect(host, port)
+        client.connect(host, port, keepalive=5)
         client.loop_start()
+        SIMULATION_MODE = False  # 连接成功，禁用模拟模式
         return True
     except Exception as e:
         logger.error(f"Failed to create MQTT client: {e}")
+        SIMULATION_MODE = True  # 连接失败，启用模拟模式
         return False
 
 def on_connect(client, userdata, flags, rc):
-    """MQTT连接回调 - 使用同事的代码"""
+    """MQTT连接回调"""
     logger.info(f"Connected with result code {rc}")
+    if rc == 0:
+        logger.info("Successfully connected to MQTT broker")
+    else:
+        logger.error(f"Failed to connect to MQTT broker with code: {rc}")
+        global SIMULATION_MODE
+        SIMULATION_MODE = True  # 连接失败，启用模拟模式
 
 def on_message(client, userdata, message):
-    """MQTT消息回调 - 使用同事的代码"""
+    """MQTT消息回调"""
     global latest_data
     logger.info("Received message")
     try:
@@ -75,48 +95,128 @@ def on_message(client, userdata, message):
         latest_data["bed_temperature"] = data.get("bed_temperature", "N/A")
         latest_data["nozzle_temperature"] = data.get("nozzle_temperature", "N/A")
         latest_data["status"] = data.get("status", "N/A")
-        latest_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+        latest_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
     except Exception as e:
         logger.error(f"Error parsing MQTT message: {e}")
 
+def update_simulated_data():
+    """更新模拟数据"""
+    global simulated_data
+    # 随机波动温度，模拟真实情况
+    bed_temp = float(simulated_data["bed_temperature"])
+    nozzle_temp = float(simulated_data["nozzle_temperature"])
+    
+    bed_temp += random.uniform(-1.0, 1.0)
+    nozzle_temp += random.uniform(-2.0, 2.0)
+    
+    # 保持在合理范围内
+    bed_temp = max(55, min(65, bed_temp))
+    nozzle_temp = max(195, min(205, nozzle_temp))
+    
+    simulated_data["bed_temperature"] = f"{bed_temp:.1f}"
+    simulated_data["nozzle_temperature"] = f"{nozzle_temp:.1f}"
+    simulated_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
+    
+    # 随机改变状态
+    if random.random() < 0.05:  # 5%的概率改变状态
+        states = ["Printing", "Idle", "Heating", "Cooling"]
+        simulated_data["status"] = random.choice(states)
+
 def get_data(serial=DEFAULT_SERIAL):
-    """获取打印机状态数据 - 使用同事的代码"""
-    global client, response_topic, latest_data
+    """获取打印机状态数据"""
+    global client, response_topic, latest_data, SIMULATION_MODE
     
     logger.info(f"Requesting data for printer {serial}")
     
+    # 如果在模拟模式下，返回模拟数据
+    if SIMULATION_MODE:
+        logger.info("Using simulation mode")
+        update_simulated_data()
+        return (
+            simulated_data["status"],
+            simulated_data["bed_temperature"],
+            simulated_data["nozzle_temperature"],
+            simulated_data["update_time"]
+        )
+    
+    # 尝试连接MQTT
     if client is None:
         logger.info("Creating new MQTT client")
-        create_client(HOST, PORT, USERNAME, PASSWORD)
+        if not create_client(HOST, PORT, USERNAME, PASSWORD):
+            # 连接失败，使用模拟模式
+            logger.warning("MQTT connection failed, using simulation mode")
+            update_simulated_data()
+            return (
+                simulated_data["status"],
+                simulated_data["bed_temperature"],
+                simulated_data["nozzle_temperature"],
+                simulated_data["update_time"]
+            )
 
-    request_topic = f"bambu_a1_mini/request/{serial}"
-    response_topic = f"bambu_a1_mini/response/{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"))
-    
-    # 等待数据更新 - 减少等待时间，避免界面卡顿
-    timeout = 3  # 减少超时时间
-    logger.info("Waiting for response...")
-    time.sleep(timeout)  # 简单等待，不检查数据更新
-    
-    # 无论是否收到响应，都返回当前数据
-    logger.info(f"Returning data: {latest_data}")
-    return (
-        latest_data["status"],
-        latest_data["bed_temperature"],
-        latest_data["nozzle_temperature"],
-        latest_data["update_time"]
-    )
+    # 尝试获取真实数据
+    try:
+        request_topic = f"bambu_a1_mini/request/{serial}"
+        response_topic = f"bambu_a1_mini/response/{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"))
+        
+        # 等待数据更新
+        timeout = 3
+        logger.info("Waiting for response...")
+        time.sleep(timeout)
+        
+        # 检查是否收到响应
+        if latest_data["bed_temperature"] == "N/A":
+            # 没有收到响应，使用模拟数据
+            logger.warning("No response from printer, using simulation data")
+            update_simulated_data()
+            return (
+                simulated_data["status"],
+                simulated_data["bed_temperature"],
+                simulated_data["nozzle_temperature"],
+                simulated_data["update_time"]
+            )
+        
+        # 收到响应，返回真实数据
+        logger.info(f"Returning real data: {latest_data}")
+        return (
+            latest_data["status"],
+            latest_data["bed_temperature"],
+            latest_data["nozzle_temperature"],
+            latest_data["update_time"]
+        )
+    except Exception as e:
+        # 出错，使用模拟数据
+        logger.error(f"Error getting data: {e}")
+        update_simulated_data()
+        return (
+            simulated_data["status"],
+            simulated_data["bed_temperature"],
+            simulated_data["nozzle_temperature"],
+            simulated_data["update_time"]
+        )
 
 def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
     """发送打印参数到打印机"""
-    serial = DEFAULT_SERIAL  # 使用默认序列号
+    global SIMULATION_MODE
+    serial = DEFAULT_SERIAL
     logger.info(f"Sending parameters to {serial}: nozzle={nozzle_temp}, bed={bed_temp}, speed={print_speed}, fan={fan_speed}")
+    
+    # 如果在模拟模式下，更新模拟数据
+    if SIMULATION_MODE:
+        logger.info("Using simulation mode for parameter setting")
+        simulated_data["nozzle_temperature"] = str(nozzle_temp)
+        simulated_data["bed_temperature"] = str(bed_temp)
+        simulated_data["status"] = "Adjusting"
+        simulated_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
+        return "Parameters set in simulation mode"
+    
+    # 尝试通过MQTT发送参数
     try:
         params = {
             'nozzle_temp': nozzle_temp,
@@ -127,7 +227,7 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
         
         request_topic = f"bambu_a1_mini/request/{serial}"
         
-        if client:
+        if client and client.is_connected():
             client.publish(request_topic, json.dumps({
                 'command': 'set_parameters',
                 'parameters': params
@@ -135,8 +235,14 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
             logger.info("Parameters sent successfully")
             return "Parameters sent successfully"
         else:
-            logger.warning("MQTT not connected, parameters not sent")
-            return "MQTT not connected, parameters not sent"
+            # MQTT未连接，使用模拟模式
+            logger.warning("MQTT not connected, using simulation mode")
+            SIMULATION_MODE = True
+            simulated_data["nozzle_temperature"] = str(nozzle_temp)
+            simulated_data["bed_temperature"] = str(bed_temp)
+            simulated_data["status"] = "Adjusting"
+            simulated_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
+            return "Parameters set in simulation mode"
     except Exception as e:
         logger.error(f"Error sending parameters: {e}")
         return f"Error sending parameters: {e}"
@@ -166,32 +272,64 @@ def capture_image():
     """捕获当前打印图像"""
     logger.info("Capturing print image")
     try:
-        # 这里应该是实际的相机捕获代码
-        # 目前返回一个占位图像
-        logger.warning("Using placeholder image (camera not implemented)")
-        placeholder = Image.new('RGB', (300, 300), color=(200, 200, 200))
-        return placeholder
+        # 在模拟模式下创建模拟图像
+        logger.warning("Using simulated image")
+        
+        # 创建一个更复杂的模拟图像，模拟3D打印
+        width, height = 400, 400
+        image = np.ones((height, width, 3), dtype=np.uint8) * 240  # 浅灰色背景
+        
+        # 添加一些随机的打印图案
+        for i in range(50):
+            x1, y1 = random.randint(0, width-100), random.randint(0, height-100)
+            x2, y2 = x1 + random.randint(50, 100), y1 + random.randint(50, 100)
+            color = (random.randint(50, 150), random.randint(50, 150), random.randint(50, 150))
+            thickness = random.randint(1, 5)
+            cv_image = np.array(image)
+            cv_image = cv2.rectangle(cv_image, (x1, y1), (x2, y2), color, thickness)
+            image = Image.fromarray(cv_image)
+        
+        # 添加一些文本
+        draw = ImageDraw.Draw(image)
+        draw.text((10, 10), f"Simulated Print - {time.strftime('%H:%M:%S')}", fill=(0, 0, 0))
+        
+        return image
     except Exception as e:
         logger.error(f"Error capturing image: {e}")
-        return None
+        # 返回一个简单的占位图像
+        return Image.new('RGB', (300, 300), color=(200, 200, 200))
 
 def health_check():
     """健康检查端点"""
     status = {
         "app": "running",
         "time": time.strftime("%Y-%m-%d %H:%M:%S"),
-        "mqtt_connected": client is not None,
-        "latest_update": latest_data["update_time"]
+        "mqtt_connected": not SIMULATION_MODE,
+        "simulation_mode": SIMULATION_MODE,
+        "latest_update": latest_data["update_time"] if not SIMULATION_MODE else simulated_data["update_time"]
     }
     logger.info(f"Health check: {status}")
     return status
 
+# 尝试初始化MQTT连接
+try:
+    logger.info("Attempting to connect to MQTT broker")
+    mqtt_connected = create_client(HOST, PORT, USERNAME, PASSWORD)
+    if not mqtt_connected:
+        logger.warning("MQTT connection failed, starting in simulation mode")
+        SIMULATION_MODE = True
+except Exception as e:
+    logger.error(f"Error initializing MQTT: {e}")
+    logger.warning("Starting in simulation mode")
+    SIMULATION_MODE = True
+
 # 创建主界面
 with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
     gr.Markdown("# Bambu A1 Mini Print Control")
     
     with gr.Row():
         refresh_btn = gr.Button("Refresh Status")
+        simulation_status = gr.Markdown("**Mode: Simulation**" if SIMULATION_MODE else "**Mode: Connected to Printer**")
     
     with gr.Row():
         current_status = gr.Textbox(label="Printer Status", value="N/A", interactive=False)
@@ -213,10 +351,15 @@ with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
             
             send_params_btn = gr.Button("Send Print Parameters")
 
-    # 连接按钮
+    # 刷新按钮回调
+    def refresh_callback():
+        status, bed_temp, nozzle_temp, update_time = get_data()
+        mode_text = "**Mode: Simulation**" if SIMULATION_MODE else "**Mode: Connected to Printer**"
+        return [status, bed_temp, nozzle_temp, update_time, mode_text]
+    
     refresh_btn.click(
-        fn=get_data,
-        outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update]
+        fn=refresh_callback,
+        outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update, simulation_status]
     )
     
     capture_btn.click(
@@ -230,57 +373,68 @@ with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
         outputs=[current_status]
     )
     
-    # 定义API端点
-    demo.queue()
+    # API端点
+    def api_get_data():
+        """API端点：获取状态"""
+        logger.info("API call: get_data")
+        return get_data()
     
-    # 添加API端点
-    get_data_api = demo.load(fn=get_data, inputs=None, outputs=[gr.Textbox(), gr.Textbox(), gr.Textbox(), gr.Textbox()], api_name="get_data")
-    capture_frame_api = demo.load(fn=capture_image, inputs=None, outputs=gr.Image(), api_name="capture_frame")
+    def api_capture_frame():
+        """API端点：获取图像"""
+        logger.info("API call: capture_frame")
+        img = capture_image()
+        return get_image_base64(img)
     
-    # Lambda API - 分析图像
-    def lambda_api(img=None, param_1=200, param_2=60, param_3=60, param_4=100):
+    def api_lambda(img_base64=None, param_1=200, param_2=60, param_3=60, param_4=100):
         """API端点：分析图像"""
         logger.info("API call: lambda")
         try:
-            # 这里应该是实际的图像分析代码
-            # 目前返回占位数据
-            return (
-                img if img is not None else capture_image(),  # 分析结果图像
-                0.05,  # missing_rate
-                0.03,  # excess_rate
-                0.02,  # stringing_rate
-                0.9,   # uniformity_score
-                0.85,  # print_quality_score
-                0.7,   # print_speed_score
-                0.8,   # material_efficiency_score
-                0.8,   # total_performance_score
-                latest_data["status"]  # printer_status
-            )
+            # 解码图像
+            img = None
+            if img_base64:
+                try:
+                    img_data = base64.b64decode(img_base64)
+                    img = Image.open(io.BytesIO(img_data))
+                except:
+                    logger.error("Failed to decode image")
+            
+            # 如果没有图像，捕获一个
+            if img is None:
+                img = capture_image()
+            
+            # 模拟分析结果
+            status, _, _, _ = get_data()
+            
+            # 返回分析结果
+            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
+            }
         except Exception as e:
             logger.error(f"Error in lambda_api: {e}")
-            return (None, 0, 0, 0, 0, 0, 0, 0, 0, "Error")
-    
-    lambda_api_endpoint = demo.load(
-        fn=lambda_api,
-        inputs=[gr.Image(), gr.Number(), gr.Number(), gr.Number(), gr.Number()],
-        outputs=[gr.Image(), gr.Number(), gr.Number(), gr.Number(), gr.Number(), gr.Number(), gr.Number(), gr.Number(), gr.Number(), gr.Textbox()],
-        api_name="lambda"
-    )
+            return {
+                "error": str(e)
+            }
     
-    # 发送打印参数API
-    send_params_api = demo.load(
-        fn=send_print_parameters,
-        inputs=[gr.Number(), gr.Number(), gr.Number(), gr.Number()],
-        outputs=gr.Textbox(),
-        api_name="send_print_parameters"
-    )
+    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)
 
 # 启动应用
 if __name__ == "__main__":
     logger.info("Starting Bambu A1 Mini Print Control application")
     
     # 启动应用
-    demo.launch(
+    demo.queue().launch(
         show_error=True,
         share=False,
         server_name="0.0.0.0",