update

app.py

@@ -14,7 +14,8 @@ import random
 import cv2
 from PIL import ImageDraw
 
-# 
+
+# 设置日志记录
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
@@ -25,198 +26,94 @@ logging.basicConfig(
 )
 logger = logging.getLogger("bambu-analysis")
 
-# Environment Variables
-HOST = "mqtt.bambulab.com"
-PORT = 8883
-USERNAME = "bblp"
-PASSWORD = "bblp"
+HOST = "mqtt.bambulab.com"  # 默认值
+PORT = 8883  # 默认值
+USERNAME = "bblp"  # 默认值
+PASSWORD = "bblp"  # 默认值
 DEFAULT_SERIAL = "0309CA471800852"
 
-# 
-SIMULATION_MODE = True  # 默认启用模拟模式
-simulated_data = {
-    "bed_temperature": "60",
-    "nozzle_temperature": "200",
-    "status": "Printing",
-    "update_time": time.strftime("%Y-%m-%d %H:%M:%S")
-}
+# 尝试从环境变量获取
+if os.environ.get("host"):
+    HOST = os.environ.get("host")
+if os.environ.get("port"):
+    PORT = int(os.environ.get("port"))
+if os.environ.get("username"):
+    USERNAME = os.environ.get("username")
+if os.environ.get("password"):
+    PASSWORD = os.environ.get("password")
+
+logger.info(f"MQTT Configuration: HOST={HOST}, PORT={PORT}, USERNAME={USERNAME}")
 
-# 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
 client = None
-response_topic = None
+response_topic = None  # Will be set dynamically
 
 def create_client(host, port, username, password):
-    """创建并配置MQTT客户端"""
-    global client, SIMULATION_MODE
-    try:
-        logger.info(f"Creating MQTT client for {host}:{port}")
-        client = mqtt.Client()
-        client.username_pw_set(username, password)
-        client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
-        client.on_connect = on_connect
-        client.on_message = on_message
-        
-        logger.info("Connecting to MQTT broker...")
-        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
+    """完全使用同事的创建客户端函数"""
+    global client
+    client = mqtt.Client()
+    client.username_pw_set(username, password)
+    client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.connect(host, port)
+    client.loop_start()
 
 def on_connect(client, userdata, flags, rc):
-    """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消息回调"""
     global latest_data
     logger.info("Received message")
     try:
         data = json.loads(message.payload)
-        logger.debug(f"Message content: {data}")
-        
         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")
+        latest_data["update_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
     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, SIMULATION_MODE
+    global client, response_topic
+
+    if client is None:
+        create_client(HOST, PORT, USERNAME, PASSWORD)
+
+    request_topic = f"bambu_a1_mini/request/{serial}"
+    response_topic = f"bambu_a1_mini/response/{serial}"
     
-    logger.info(f"Requesting data for printer {serial}")
+    logger.info(f"Subscribing to {response_topic}")
+    client.subscribe(response_topic)
     
-    # 如果在模拟模式下，返回模拟数据
-    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"]
-        )
+    # 发送请求获取数据
+    logger.info(f"Publishing request to {request_topic}")
+    client.publish(request_topic, json.dumps("HI"))
     
-    # 尝试连接MQTT
-    if client is None:
-        logger.info("Creating new MQTT client")
-        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"]
-            )
-
-    # 尝试获取真实数据
-    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"]
-            )
+    global latest_data
+    latest_data["bed_temperature"] = "N/A"
+    timeout = 10
+    while latest_data["bed_temperature"] == "N/A" and timeout > 0:
+        time.sleep(1)
+        timeout -= 1
         
-        # 收到响应，返回真实数据
-        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"]
-        )
+    return (
+        latest_data["status"],
+        latest_data["bed_temperature"],
+        latest_data["nozzle_temperature"],
+        latest_data["update_time"]
+    )
 
 def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
     """发送打印参数到打印机"""
-    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,
@@ -227,7 +124,7 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
         
         request_topic = f"bambu_a1_mini/request/{serial}"
         
-        if client and client.is_connected():
+        if client:
             client.publish(request_topic, json.dumps({
                 'command': 'set_parameters',
                 'parameters': params
@@ -235,14 +132,8 @@ def send_print_parameters(nozzle_temp, bed_temp, print_speed, fan_speed):
             logger.info("Parameters sent successfully")
             return "Parameters sent successfully"
         else:
-            # 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"
+            logger.warning("MQTT not connected, parameters not sent")
+            return "MQTT not connected, parameters not sent"
     except Exception as e:
         logger.error(f"Error sending parameters: {e}")
         return f"Error sending parameters: {e}"
@@ -272,64 +163,32 @@ def capture_image():
     """捕获当前打印图像"""
     logger.info("Capturing print image")
     try:
-        # 在模拟模式下创建模拟图像
-        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
+        # 这里应该是实际的相机捕获代码
+        # 目前返回一个占位图像
+        logger.warning("Using placeholder image (camera not implemented)")
+        placeholder = Image.new('RGB', (300, 300), color=(200, 200, 200))
+        return placeholder
     except Exception as e:
         logger.error(f"Error capturing image: {e}")
-        # 返回一个简单的占位图像
-        return Image.new('RGB', (300, 300), color=(200, 200, 200))
+        return None
 
 def health_check():
     """健康检查端点"""
     status = {
         "app": "running",
         "time": time.strftime("%Y-%m-%d %H:%M:%S"),
-        "mqtt_connected": not SIMULATION_MODE,
-        "simulation_mode": SIMULATION_MODE,
-        "latest_update": latest_data["update_time"] if not SIMULATION_MODE else simulated_data["update_time"]
+        "mqtt_connected": client is not None,
+        "latest_update": latest_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)
@@ -351,15 +210,10 @@ 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=refresh_callback,
-        outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update, simulation_status]
+        fn=get_data,
+        outputs=[current_status, current_bed_temp, current_nozzle_temp, last_update]
     )
     
     capture_btn.click(
@@ -433,6 +287,13 @@ with gr.Blocks(title="Bambu A1 Mini Print Control") as demo:
 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,