测试代码

Test/mqtt_test.py

@@ -0,0 +1,118 @@
+import paho.mqtt.client as mqtt
+import json
+import time
+import picamera2
+import cv2
+
+picam2 = picamera2.Picamera2()
+picam2.configure(picam2.create_still_configuration(
+    main={"size": (2048, 2048),
+          "format": "RGB888"}
+))
+picam2.start()
+
+# MQTT 服务器配置
+MQTT_BROKER = "192.168.77.132"  # 机械臂的IP地址
+MQTT_PORT = 1883  # 端口
+MQTT_KEEP_ALIVE_INTERVAL = 60  # 保持连接时间间隔
+CLIENT_ID = "arm_client_001"  # 客户端ID
+
+# 定义 topic
+# 发送
+TOPIC_COMMAND = "arm_card_dealer/command"
+TOPIC_CAMERA_RESPONSE = "arm_card_dealer/camera/response"
+
+# 接收
+TOPIC_STATUS = "arm_card_dealer/status"
+TOPIC_ERROR = "arm_card_dealer/error"
+TOPIC_CAMERA_COMMAND = "arm_card_dealer/camera/command"
+
+
+def capture(img_path: str):
+    frame = picam2.capture_array()  # 捕获图像
+    # 保存图片查看结果
+    cv2.imwrite(img_path, frame)
+
+    return True
+
+
+# 定义发送指令函数
+def send_command(cmd, request_id, cycles=-1):
+    payload = {
+        "cmd": cmd,
+        "cycles": cycles,
+        "request_id": request_id,
+        "timestamp": int(time.time() * 1000)  # 当前时间戳
+    }
+    client.publish(TOPIC_COMMAND, json.dumps(payload), qos=1)
+    print(f"Sent command: {payload}")
+
+
+# success	integer	拍照是否成功（1=成功，0=失败）
+def send_camera_response(success=1, error_message=''):
+    payload = {
+        "success": success,
+        "error_message": "",
+        "timestamp": int(time.time() * 1000)
+    }
+    client.publish(TOPIC_CAMERA_RESPONSE, json.dumps(payload), qos=1)
+    print(f"Sent camera response: {payload}")
+
+
+# 定义处理接收到的消息的回调函数
+def on_message(client, userdata, msg):
+    print('------------------')
+    print(f"[Get Topic: {msg.topic}: {msg.payload.decode()}]")
+
+    # 处理状态消息
+    if msg.topic == TOPIC_STATUS:
+        status_data = json.loads(msg.payload.decode())
+        print("statue:", status_data)
+
+    # 处理错误消息
+    elif msg.topic == TOPIC_ERROR:
+        error_data = json.loads(msg.payload.decode())
+        print("error:", error_data)
+
+    elif msg.topic == TOPIC_CAMERA_COMMAND:
+        camera_command_data = json.loads(msg.payload.decode())
+        print("arm request capture: ", camera_command_data)
+
+        # 拍照
+        if camera_command_data.get("cmd") == "capture":
+            img_name = camera_command_data.get("id", "0")
+            capture(f"{img_name}.jpg")
+            time.sleep(0.5)
+            print(f"--- save img {img_name} ---")
+
+        # 通知拍照完成
+        send_camera_response(success=1)
+
+
+# 初始化 MQTT 客户端
+client = mqtt.Client(CLIENT_ID)
+
+# 设置回调函数
+client.on_message = on_message
+
+# 连接到 MQTT broker
+client.connect(MQTT_BROKER, MQTT_PORT, MQTT_KEEP_ALIVE_INTERVAL)
+
+# 订阅相关主题
+client.subscribe(TOPIC_STATUS, qos=1)
+client.subscribe(TOPIC_ERROR, qos=1)
+client.subscribe(TOPIC_CAMERA_COMMAND, qos=1)
+
+print("--已订阅主题--")
+
+# 发送 START 命令，假设 request_id 为 "req-2024-001"
+send_command("start", "req-2026-001", cycles=1)
+
+# 开始监听 MQTT 消息
+client.loop_start()
+
+# 在这里保持长时间运行，等待响应
+time.sleep(70)  # 可以根据实际需求调整
+
+# 停止监听
+client.loop_stop()

Test/test01.py

@@ -6,7 +6,8 @@ import time
 
 picam2 = picamera2.Picamera2()
 picam2.configure(picam2.create_still_configuration(
-    main={"size": (2048, 2048)}
+    main={"size": (2048, 2048),
+          "format": "RGB888"}
     # main={"size": (1800, 1800)}
 ))
 picam2.start()  # 开始相机流
@@ -25,11 +26,9 @@ except Exception as e:
 time.sleep(0.1)  # 等待相机初始化
 
 frame = picam2.capture_array()  # 捕获图像
-
 # 打印图像信息
 print(frame)
 print(frame.shape)  # 这里使用 shape 来查看图像的尺寸
 
-frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
 # 保存图片查看结果
 cv2.imwrite("test.jpg", frame)

Test/test02.py

@@ -1,4 +1,3 @@
-import os
+import paho.mqtt.client as mqtt
 
-t = os.getenv("APP_TITLE", "Raspberry Pi Camera API")
-print(t)
+print(1)

app/services/camera_service.py

@@ -54,21 +54,26 @@ class CameraService:
     def capture_jpeg(self) -> bytes:
         """
         抓拍一张图片，并直接返回 JPEG 二进制内容。
-
-        接口层拿到这个字节流之后，就可以直接通过 HTTP 返回
         """
         with self._lock:
             camera = self._ensure_camera()
 
             try:
                 frame = camera.capture_array()
+                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
             except Exception as exc:
                 raise CameraCaptureError("相机抓拍失败，请检查相机连接和占用状态。") from exc
 
-            rgb_frame = self._normalize_frame(frame)
+            # 当前设备输出 shape=(H, W, 3)，PIL 直接接收前3通道颜色正确
+            if getattr(frame, "ndim", 0) != 3 or frame.shape[2] < 3:
+                raise CameraCaptureError(
+                    f"相机返回了无法识别的帧结构: shape={getattr(frame, 'shape', None)}"
+                )
 
             try:
-                image = Image.fromarray(rgb_frame)
+                image = Image.fromarray(frame[:, :, :3])
+
+                # image.save("temp.jpg")
                 buffer = BytesIO()
                 image.save(
                     buffer,
@@ -192,34 +197,6 @@ class CameraService:
             except Exception:
                 logger.warning("当前摄像头不支持自动白平衡，已跳过。", exc_info=True)
 
-    def _normalize_frame(self, frame: Any) -> Any:
-        """
-        统一处理相机返回的帧格式，最终转换成适合 JPEG 编码的 RGB 数据。
-
-        `Picamera2` 在不同配置下，返回的数组通道数可能不同：
-        - 4 通道：常见于 RGBA
-        - 3 通道：通常是 RGB
-        - 1 通道：灰度图
-        """
-        if frame is None:
-            raise CameraCaptureError("相机返回了空帧。")
-
-        if getattr(frame, "ndim", 0) == 2:
-            return cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
-
-        if getattr(frame, "ndim", 0) != 3:
-            raise CameraCaptureError("相机返回了无法识别的帧结构。")
-
-        channel_count = frame.shape[2]
-        if channel_count == 4:
-            return cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
-        if channel_count == 3:
-            return frame
-        if channel_count == 1:
-            return cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
-
-        raise CameraCaptureError(f"暂不支持 {channel_count} 通道的图像数据。")
-
 
 _camera_service: Optional[CameraService] = None
 _camera_service_lock = threading.Lock()

requirement.txt

@@ -0,0 +1 @@
+paho-mqtt==1.6.1