Upload files to "/"

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+*.py[cod]
+.venv/
+venv/

README.md

@@ -0,0 +1,185 @@
+# ESP32 + IMU 采集与夹爪控制
+
+本项目使用 MicroPython 在 ESP32 上读取 IMU 的 TTL 串口数据，并通过板载 CP2102 USB 串口将数据发送到电脑。
+
+## 硬件连接概览
+
+```text
+IMU -> TTL UART -> ESP32 -> CP2102 -> USB -> 电脑
+```
+
+## 一、准备软件
+
+1. 安装 [Thonny](https://thonny.org/)。
+2. 使用支持数据传输的 Micro-USB 线连接 ESP32。
+3. 打开 Windows 设备管理器，确认出现类似设备：
+
+   ```text
+   Silicon Labs CP210x USB to UART Bridge (COM5)
+   ```
+
+4. 记住对应的 `COM` 端口，例如 `COM5`。
+
+## 二、安装 MicroPython 固件
+
+打开 Thonny，然后执行以下步骤：
+
+1. 选择 `工具 -> 选项 -> 解释器`。
+2. 解释器选择：
+
+   ```text
+   MicroPython (ESP32)
+   ```
+
+3. 端口选择 ESP32 对应的 `COM` 端口。
+4. 点击“安装或更新 MicroPython”。
+5. 目标端口选择对应的 `COM` 端口。
+6. 芯片选择 `ESP32`。
+7. 建议勾选“擦除闪存”。
+8. 点击安装。
+
+如果提示无法连接：
+
+1. 按住 ESP32 上的 `BOOT` 按钮。
+2. 点击安装。
+3. 出现连接或写入进度后松开 `BOOT`。
+
+安装完成后，Thonny 下方 Shell 中通常会显示：
+
+```text
+MicroPython ...
+>>>
+```
+
+## 三、上传 main.py
+
+ESP32 端程序位于：
+
+[main.py](./main.py)
+
+在 Thonny 中：
+
+1. 选择 `文件 -> 打开`。
+2. 打开仓库中的 `main.py`。
+3. 选择 `文件 -> 另存为`。
+4. 选择保存到“MicroPython设备”。
+5. 将文件名填写为 `main.py`。
+
+保存后，Thonny 左侧“MicroPython设备”中应能看到：
+
+```text
+/main.py
+```
+
+ESP32 启动时会依次自动执行：
+
+```text
+boot.py
+main.py
+```
+
+因此，程序必须准确命名为 `main.py`。
+
+## 四、连接 IMU
+
+### TTL 串口
+
+| IMU | ESP32 |
+|---|---|
+| VCC | 按模块要求连接 3.3V 或 5V |
+| GND | GND |
+| TX | GPIO16 |
+| RX | GPIO17 |
+
+```text
+IMU VCC -> ESP32 3.3V/5V
+IMU GND -> ESP32 GND
+IMU TX  -> ESP32 GPIO16
+IMU RX  -> ESP32 GPIO17
+```
+
+> ESP32 GPIO 不能承受 5V。连接前必须确认 IMU 的 TTL 信号电平兼容 3.3V。
+
+### 四路数字输入
+
+| 输入 | ESP32 |
+|---|---|
+| DIN1 | GPIO25 |
+| DIN2 | GPIO26 |
+| DIN3 | GPIO27 |
+| DIN4 | GPIO32 |
+| 信号地 | GND |
+
+数字输入电压不能超过 `3.3V`。
+
+## 五、重新上电运行
+
+上传完成后：
+
+1. 关闭 Thonny，或者点击 Thonny 的“停止/断开”。
+2. 按一下 ESP32 的 `EN/RESET` 按钮。
+3. 也可以拔掉 USB 后重新连接。
+
+以后每次 ESP32 上电时，都会自动运行 `main.py`：
+
+```text
+IMU -> TTL -> ESP32 -> CP2102 -> USB -> 电脑
+```
+
+## 六、电脑接收数据
+
+电脑端程序位于：
+
+[pc_reader.py](./pc_reader.py)
+
+安装 Python 串口依赖：
+
+```powershell
+pip install pyserial
+```
+
+运行程序：
+
+```powershell
+python pc_reader.py COM5
+```
+
+将 `COM5` 替换为 ESP32 的实际串口号。
+
+正常输出类似：
+
+```text
+#       1  200.0Hz DIN=[1, 1, 0, 1] DOUT=[0, 0] ACC=[...] GYRO=[...] ANGLE=[...]
+```
+
+电脑接收程序使用 `230400` 波特率。
+
+> 运行 `pc_reader.py` 前必须断开 Thonny，否则串口可能被 Thonny 占用。
+
+## 七、夹爪控制命令
+
+运行 `pc_reader.py` 后，可以输入以下命令并按回车：
+
+| 命令 | 操作 |
+|---|---|
+| `o` | 打开夹爪 |
+| `c` | 闭合夹爪 |
+| `s` | 停止，两路控制输出均释放 |
+
+ESP32 使用以下管脚控制外部驱动电路：
+
+| 功能 | ESP32 |
+|---|---|
+| 控制输出1 | GPIO18 |
+| 控制输出2 | GPIO19 |
+
+ESP32 GPIO 只能输出 `3.3V`，不能直接连接 12V 工业控制信号。必须使用光耦隔离模块、晶体管或其他合适的电平转换和驱动电路。
+
+## 项目文件
+
+```text
+.
+|-- README.md
+|-- main.py
+`-- pc_reader.py
+```

main.py

@@ -0,0 +1,243 @@
+from machine import Pin, UART
+import struct
+import time
+
+# Wiring:
+# IMU TX -> ESP32 GPIO16
+# IMU RX -> ESP32 GPIO17
+# IMU GND -> ESP32 GND
+IMU_RX_PIN = 16
+IMU_TX_PIN = 17
+
+# CP2102 USB serial. Binary output is used because JSON is too large at 200 Hz.
+USB_BAUD = 230400
+
+DIN_PINS = (25, 26, 27, 32)
+digital_inputs = [Pin(pin, Pin.IN, Pin.PULL_UP) for pin in DIN_PINS]
+
+# Two controllable digital outputs. Both default to LOW at boot.
+DOUT_PINS = (18, 19)
+digital_outputs = [Pin(pin, Pin.OUT, value=0) for pin in DOUT_PINS]
+
+FRAME_LEN = 11
+HEADER = 0x55
+
+# WIT normal-protocol frame types
+TYPE_ACC = 0x51
+TYPE_GYRO = 0x52
+TYPE_ANGLE = 0x53
+
+# Latest raw values: acc XYZ, gyro XYZ, angle XYZ, temperature
+raw_values = [0] * 10
+rx_buffer = bytearray()
+command_buffer = bytearray()
+
+
+def write_register(uart, address, value):
+    uart.write(bytes((
+        0xFF,
+        0xAA,
+        address & 0xFF,
+        value & 0xFF,
+        (value >> 8) & 0xFF,
+    )))
+
+
+def has_valid_frame(uart, timeout_ms=400):
+    test_buffer = bytearray()
+    deadline = time.ticks_add(time.ticks_ms(), timeout_ms)
+
+    while time.ticks_diff(deadline, time.ticks_ms()) > 0:
+        count = uart.any()
+        if count:
+            chunk = uart.read(count)
+            if chunk:
+                test_buffer.extend(chunk)
+
+        while len(test_buffer) >= FRAME_LEN:
+            if test_buffer[0] != HEADER:
+                del test_buffer[0]
+                continue
+
+            frame = test_buffer[:FRAME_LEN]
+            if (sum(frame[:10]) & 0xFF) == frame[10]:
+                return True
+
+            del test_buffer[0]
+
+        time.sleep_ms(1)
+
+    return False
+
+
+def open_imu_uart(baudrate):
+    return UART(
+        2,
+        baudrate=baudrate,
+        bits=8,
+        parity=None,
+        stop=1,
+        rx=IMU_RX_PIN,
+        tx=IMU_TX_PIN,
+        timeout=0,
+        rxbuf=2048,
+    )
+
+
+def start_imu_at_200hz():
+    # If data already arrives at 115200, enforce 200 Hz for this session.
+    # No SAVE is needed here, avoiding repeated writes to sensor flash.
+    uart = open_imu_uart(115200)
+    if has_valid_frame(uart):
+        write_register(uart, 0x69, 0xB588)
+        time.sleep_ms(200)
+        write_register(uart, 0x02, 0x000E)  # ACC + GYRO + ANGLE only
+        time.sleep_ms(50)
+        write_register(uart, 0x03, 0x000B)  # RRATE = 200 Hz
+        time.sleep_ms(100)
+        return uart
+
+    # Factory/default examples use 9600 baud.
+    uart.deinit()
+    uart = open_imu_uart(9600)
+
+    if not has_valid_frame(uart):
+        raise RuntimeError("IMU not detected at 9600 or 115200 baud")
+
+    # Official protocol sequence:
+    # unlock -> change baud -> switch local UART -> unlock -> set 200 Hz -> save
+    write_register(uart, 0x69, 0xB588)
+    time.sleep_ms(200)
+    write_register(uart, 0x04, 0x0006)  # BAUD = 115200
+    time.sleep_ms(100)
+
+    uart.deinit()
+    uart = open_imu_uart(115200)
+    time.sleep_ms(200)
+
+    write_register(uart, 0x69, 0xB588)
+    time.sleep_ms(200)
+    write_register(uart, 0x02, 0x000E)  # ACC + GYRO + ANGLE only
+    time.sleep_ms(50)
+    write_register(uart, 0x03, 0x000B)  # RRATE = 200 Hz
+    time.sleep_ms(100)
+    write_register(uart, 0x00, 0x0000)  # SAVE
+    time.sleep_ms(500)
+
+    return uart
+
+
+def update_raw_values(frame):
+    frame_type = frame[1]
+    x, y, z, extra = struct.unpack_from("<hhhh", frame, 2)
+
+    if frame_type == TYPE_ACC:
+        raw_values[0:3] = (x, y, z)
+        raw_values[9] = extra
+    elif frame_type == TYPE_GYRO:
+        raw_values[3:6] = (x, y, z)
+    elif frame_type == TYPE_ANGLE:
+        raw_values[6:9] = (x, y, z)
+        return True
+
+    return False
+
+
+def poll_imu(uart):
+    count = uart.any()
+    if count:
+        chunk = uart.read(count)
+        if chunk:
+            rx_buffer.extend(chunk)
+
+    angle_updated = False
+
+    while len(rx_buffer) >= FRAME_LEN:
+        if rx_buffer[0] != HEADER:
+            del rx_buffer[0]
+            continue
+
+        frame = rx_buffer[:FRAME_LEN]
+        if (sum(frame[:10]) & 0xFF) != frame[10]:
+            del rx_buffer[0]
+            continue
+
+        del rx_buffer[:FRAME_LEN]
+        if update_raw_values(frame):
+            angle_updated = True
+
+    return angle_updated
+
+
+def make_packet():
+    din_mask = 0
+    for index, pin in enumerate(digital_inputs):
+        din_mask |= pin.value() << index
+
+    dout_mask = 0
+    for index, pin in enumerate(digital_outputs):
+        dout_mask |= pin.value() << index
+
+    # Packet layout:
+    # A5 5A | version | DIN | DOUT | ticks_ms | 10 signed int16 | checksum
+    packet = struct.pack(
+        "<2sBBBI10h",
+        b"\xA5\x5A",
+        2,
+        din_mask,
+        dout_mask,
+        time.ticks_ms() & 0xFFFFFFFF,
+        *raw_values
+    )
+    return packet + bytes((sum(packet) & 0xFF,))
+
+
+def set_gripper(mode):
+    # GPIO HIGH means the external transistor/optocoupler is conducting.
+    # Always release both inputs briefly before changing direction.
+    digital_outputs[0].value(0)
+    digital_outputs[1].value(0)
+    time.sleep_ms(20)
+
+    if mode == 1:       # OPEN: line 2 = 12 V, line 3 = 0 V
+        digital_outputs[1].value(1)
+    elif mode == 2:     # CLOSE: line 2 = 0 V, line 3 = 12 V
+        digital_outputs[0].value(1)
+
+
+def poll_usb_commands(uart):
+    count = uart.any()
+    if count:
+        chunk = uart.read(count)
+        if chunk:
+            command_buffer.extend(chunk)
+
+    # Command: C3 3C | mode | reserved | checksum
+    # mode: 0=STOP, 1=OPEN, 2=CLOSE
+    while len(command_buffer) >= 5:
+        if command_buffer[0] != 0xC3 or command_buffer[1] != 0x3C:
+            del command_buffer[0]
+            continue
+
+        command = command_buffer[:5]
+        if (sum(command[:4]) & 0xFF) != command[4]:
+            del command_buffer[0]
+            continue
+
+        del command_buffer[:5]
+        mode = command[2]
+        if mode <= 2:
+            set_gripper(mode)
+
+
+imu_uart = start_imu_at_200hz()
+usb_uart = UART(0, baudrate=USB_BAUD, bits=8, parity=None, stop=1, tx=1, rx=3)
+
+while True:
+    poll_usb_commands(usb_uart)
+
+    # The angle frame is used as the 200 Hz snapshot trigger.
+    if poll_imu(imu_uart):
+        usb_uart.write(make_packet())
+
+    time.sleep_ms(0)

pc_reader.py

@@ -0,0 +1,118 @@
+import argparse
+import struct
+import threading
+
+import serial
+
+
+SYNC = b"\xA5\x5A"
+PACKET_FORMAT = "<2sBBBI10hB"
+PACKET_SIZE = struct.calcsize(PACKET_FORMAT)
+
+
+def decode_packet(packet):
+    fields = struct.unpack(PACKET_FORMAT, packet)
+    _, version, din_mask, dout_mask, time_ms, *rest = fields
+    raw = rest[:-1]
+
+    acc_scale = 16.0 / 32768.0
+    gyro_scale = 2000.0 / 32768.0
+    angle_scale = 180.0 / 32768.0
+
+    return {
+        "version": version,
+        "t_ms": time_ms,
+        "din": [(din_mask >> bit) & 1 for bit in range(4)],
+        "dout": [(dout_mask >> bit) & 1 for bit in range(2)],
+        "acc_g": [round(value * acc_scale, 5) for value in raw[0:3]],
+        "gyro_dps": [round(value * gyro_scale, 4) for value in raw[3:6]],
+        "angle_deg": [round(value * angle_scale, 4) for value in raw[6:9]],
+        "temp_c": round(raw[9] / 100.0, 2),
+    }
+
+
+def send_gripper_command(ser, mode):
+    command = bytearray((0xC3, 0x3C, mode, 0))
+    command.append(sum(command) & 0xFF)
+    ser.write(command)
+
+
+def keyboard_control(ser):
+    print("Commands: o=OPEN, c=CLOSE, s=STOP")
+    while True:
+        text = input().strip().lower()
+        if text == "o":
+            send_gripper_command(ser, 1)
+        elif text == "c":
+            send_gripper_command(ser, 2)
+        elif text == "s":
+            send_gripper_command(ser, 0)
+        else:
+            print("Unknown command. Use o, c, or s.")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("port", help="ESP32 serial port, for example COM5")
+    parser.add_argument("--baud", type=int, default=230400)
+    args = parser.parse_args()
+
+    buffer = bytearray()
+    packet_count = 0
+    last_time_ms = None
+
+    with serial.Serial(args.port, args.baud, timeout=1) as ser:
+        print("Reading {} at {} baud".format(args.port, args.baud))
+        threading.Thread(
+            target=keyboard_control,
+            args=(ser,),
+            daemon=True,
+        ).start()
+
+        while True:
+            chunk = ser.read(ser.in_waiting or 1)
+            if chunk:
+                buffer.extend(chunk)
+
+            while len(buffer) >= PACKET_SIZE:
+                sync_index = buffer.find(SYNC)
+                if sync_index < 0:
+                    del buffer[:-1]
+                    break
+                if sync_index:
+                    del buffer[:sync_index]
+                if len(buffer) < PACKET_SIZE:
+                    break
+
+                packet = bytes(buffer[:PACKET_SIZE])
+                if (sum(packet[:-1]) & 0xFF) != packet[-1]:
+                    del buffer[0]
+                    continue
+
+                del buffer[:PACKET_SIZE]
+                data = decode_packet(packet)
+                packet_count += 1
+
+                if last_time_ms is None:
+                    frequency = 0.0
+                else:
+                    delta = (data["t_ms"] - last_time_ms) & 0xFFFFFFFF
+                    frequency = 1000.0 / delta if delta else 0.0
+                last_time_ms = data["t_ms"]
+
+                print(
+                    "#{:8d} {:6.1f}Hz DIN={} DOUT={} "
+                    "ACC={} GYRO={} ANGLE={}".format(
+                        packet_count,
+                        frequency,
+                        data["din"],
+                        data["dout"],
+                        data["acc_g"],
+                        data["gyro_dps"],
+                        data["angle_deg"],
+                    )
+                )
+
+
+if __name__ == "__main__":
+    main()