PlatformIO UART filter for Rd-03E (AI-Thinker)

filter_rd03e.py

"""
Custom PlatformIO filter for parsing Rd-03E radar sensor data from UART.

This filter is designed to process UART frames from the Rd-03E gesture recognition radar sensor.
It extracts distance and gesture information from the sensor's output and formats it for easy reading.

Usage:
1. Place this file in the `monitor` directory of your PlatformIO project as `filter_rd03e.py`
2. Run the PlatformIO device monitor with the filter enabled:
   `platformio device monitor -b 256000 -p /dev/ttyUSB0 -f rd03e`

Frame Format:
The Rd-03E outputs data in the following frame format:
- Start marker: `0xAA`
- Distance information: 2 bytes (little-endian, in centimeters)
- Gesture information: 1 byte
- End marker: `0x55`

Example Frames:
- `AA 2D 00 00 55`: Distance = 0x002D = 45 cm, Gesture = 0x00 (no gesture)
- `AA 36 00 01 55`: Distance = 0x0036 = 54 cm, Gesture = 0x01 (gesture detected)

Output:
The filter outputs the parsed distance and gesture information in the following format:
`distance = <distance> cm, gesture = <gesture_hex>`

Additional Notes:
- The distance is calculated in centimeters using little-endian format.
- The gesture result has an output delay of approximately 0.25 seconds.
- Invalid or incomplete frames are ignored.

For more details, refer to the Rd-03E user manual:
https://docs.ai-thinker.com/_media/rd-03e_gesture_recognition_user_manual.pdf
"""

import serial

from platformio.device.monitor.filters.base import DeviceMonitorFilterBase


# NOTE: it's rewrited Hexlify filter
class RD03E(DeviceMonitorFilterBase):
    NAME = "rd03e"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._buffer = bytearray()

    def set_running_terminal(self, terminal):
        # Force to Latin-1, issue #4732
        if terminal.input_encoding == "UTF-8":
            terminal.set_rx_encoding("Latin-1")
        super().set_running_terminal(terminal)

    def rx(self, text):
        result = ""
        buffer = self._buffer

        for c in serial.iterbytes(text):
            asciicode = ord(c)
            buffer.extend((asciicode).to_bytes())

            # Search for the start marker of the frame (0xAA)
            aa_pos = buffer.find(b"\xaa")
            if aa_pos == -1:
                # No start marker found, clear the buffer
                buffer.clear()
                continue

            # Check if there are enough bytes for a complete frame
            if aa_pos + 4 >= len(buffer):
                continue  # Not enough data

            # Check the end marker of the frame (0x55)
            if buffer[aa_pos + 4] != 0x55:
                # Remove the incorrect start marker and continue searching
                del buffer[: aa_pos + 1]
                continue

            # Extract frame data
            d_low = buffer[aa_pos + 1]
            d_high = buffer[aa_pos + 2]
            gesture = buffer[aa_pos + 3]

            # Calculate distance (little-endian)
            distance = (d_high << 8) | d_low

            # Format the output
            result = f"distance = {distance} cm, gesture = {gesture:02X}\n"

            # Remove the processed frame from the buffer
            buffer.clear()

        return result