ESP32: Gesture-Controlled MP3 Player

from micropython import const
from lib.dfplayerpro import DFPlayerPro
from lib.DFRobot_GR10_30_I2C import (DFRobot_GR10_30_I2C,
                                     GESTURE_UP, GESTURE_DOWN,
                                     GESTURE_LEFT, GESTURE_RIGHT,
                                     GESTURE_CLOCKWISE_C, GESTURE_COUNTERCLOCKWISE_C)
from utime import sleep, sleep_ms


UART_TX_GPIO = const(17)
UART_RX_GPIO = const(16)
I2C_SDA_PIN = const(21)
I2C_SCL_PIN = const(22)

_START_VOLUME = const(20)
_PLAYER_MODE = const(2)
_DELAY_MS = const(100)


def handling_gesture(sensor: DFRobot_GR10_30_I2C, player: DFPlayerPro) -> None:
    """
    handling gesture from sensor and trigger player action

    :param sensor: sensor instance
    :type sensor: DFRobot_GR10_30_I2C
    :param player: player instance
    :type player: DFPlayerPro
    :return: None
    """
    delay_sec = 1
    gesture = int(sensor.get_gestures())

    if gesture == 4 or gesture == 8:
        player.play_pause()
    elif gesture == 32768:
        player.fast_forward(delay_sec)
    elif gesture == 16384:
        player.fast_rewind(delay_sec)
    elif gesture == 1:
        player.play_next()
    elif gesture == 2:
        player.play_last()
    else:
        print(f'[WARN] unknown gesture: {gesture}')


if __name__ == '__main__':
    dfplayer = DFPlayerPro(tx_pin=int(UART_TX_GPIO), rx_pin=int(UART_RX_GPIO))
    dfplayer.set_volume(_START_VOLUME)
    dfplayer.set_play_mode(_PLAYER_MODE)
    dfplayer.play_file_by_number(1)

    gesture_sensor = DFRobot_GR10_30_I2C(sda=int(I2C_SDA_PIN), scl=int(I2C_SCL_PIN))
    gesture_sensor.en_gestures(GESTURE_UP | GESTURE_DOWN |
                               GESTURE_LEFT | GESTURE_RIGHT |
                               GESTURE_CLOCKWISE_C | GESTURE_COUNTERCLOCKWISE_C)

    while True:
        if gesture_sensor.get_data_ready():
            handling_gesture(sensor=gesture_sensor, player=dfplayer)

        sleep_ms(_DELAY_MS)

from micropython import const
from machine import I2C, Pin
from utime import sleep


GR10_30_I2C_ADDR = const(0x73)

GR30_10_INPUT_REG_ADDR = const(0x02)
GR30_10_INPUT_REG_DATA_READY = const(0x06)
GR30_10_INPUT_REG_INTERRUPT_STATE = const(0x07)
GR30_10_INPUT_REG_EXIST_STATE = const(0x08)

GR30_10_HOLDING_REG_INTERRUPT_MODE = const(0x09)
GR30_10_HOLDING_REG_RESET = const(0x18)

GESTURE_UP = (1 << 0)
GESTURE_DOWN = (1 << 1)
GESTURE_LEFT = (1 << 2)
GESTURE_RIGHT = (1 << 3)
GESTURE_FORWARD = (1 << 4)
GESTURE_BACKWARD = (1 << 5)
GESTURE_CLOCKWISE = (1 << 6)
GESTURE_COUNTERCLOCKWISE = (1 << 7)
GESTURE_WAVE = (1 << 8)
GESTURE_HOVER = (1 << 9)
GESTURE_UNKNOWN = (1 << 10)
GESTURE_CLOCKWISE_C = (1 << 14)
GESTURE_COUNTERCLOCKWISE_C = (1 << 15)


class DFRobot_GR10_30_I2C:
    """
    MicroPython class for communication with the GR10_30 from DFRobot via I2C
    """

    def __init__(self, sda, scl, i2c_addr=GR10_30_I2C_ADDR, i2c_bus=0):
        """
        Initialize the DFRobot_GR10_30 communication
        :param sda: I2C SDA Pin
        :param scl: I2C SCL Pin
        :param i2c_addr: I2C address
        :param i2c_bus: I2C bus number
        """
        self._addr = i2c_addr
        self._temp_buffer = [0] * 2

        try:
            self._i2c = I2C(i2c_bus, sda=Pin(sda), scl=Pin(scl))
        except Exception as err:
            print(f'Could not initialize i2c! bus: {i2c_bus}, sda: {sda}, scl: {scl}, error: {err}')

    def _write_reg(self, reg, data) -> None:
        """
        Write data to the I2C register
        :param reg: register address
        :param data: data to write
        :return: None
        """
        if isinstance(data, int):
            data = [data]

        try:
            self._i2c.writeto_mem(self._addr, reg, bytearray(data))
        except Exception as err:
            print(f'Write issue: {err}')

    def _read_reg(self, reg, length) -> bytes:
        """
        Reads data from the I2C register
        :param reg: I2C register address
        :param length: number of bytes to read
        :return: bytes
        """
        try:
            result = self._i2c.readfrom_mem(self._addr, reg, length)
        except Exception as err:
            print(f'Read issue: {err}')
            result = [0, 0]

        return result

    def _detect_device_address(self) -> int:
        """
        Detect I2C device address
        :return: int
        """
        r_buf = self._read_reg(GR30_10_INPUT_REG_ADDR, 2)
        data = r_buf[0] << 8 | r_buf[1]

        return data

    def _reset_sensor(self) -> None:
        """
        Reset sensor
        :return: None
        """
        self._temp_buffer[0] = 0x55
        self._temp_buffer[1] = 0x00
        self._write_reg(GR30_10_HOLDING_REG_RESET, self._temp_buffer)

        sleep(0.1)

    def begin(self) -> bool:
        """
        Initialise the sensor
        :return: bool
        """
        if self._detect_device_address() != GR10_30_I2C_ADDR:
            return False

        self._reset_sensor()
        sleep(0.5)

        return True

    def en_gestures(self, gestures) -> None:
        """
        Set what gestures the sensor can recognize
        :param gestures: constants combined with bitwise OR
        :return: None
        """
        gestures = gestures & 0xc7ff

        self._temp_buffer[0] = (gestures >> 8) & 0xC7
        self._temp_buffer[1] = gestures & 0x00ff
        self._write_reg(GR30_10_HOLDING_REG_INTERRUPT_MODE, self._temp_buffer)

        sleep(0.1)

    def get_exist(self) -> bool:
        """
        Get the existence of an object in the sensor detection range
        :return: bool
        """
        r_buf = self._read_reg(GR30_10_INPUT_REG_EXIST_STATE, 2)
        data = r_buf[0] * 256 + r_buf[1]

        return bool(data)

    def get_data_ready(self) -> bool:
        """
        Get if a gesture is detected
        :return: bool
        """
        r_buf = self._read_reg(GR30_10_INPUT_REG_DATA_READY, 2)
        data = r_buf[0] * 256 + r_buf[1]

        if data == 0x01:
            return True
        else:
            return False

    def get_gestures(self) -> int:
        """
        Get the gesture number of an gesture
        :return: int
        """
        r_buf = self._read_reg(GR30_10_INPUT_REG_INTERRUPT_STATE, 2)
        data = (r_buf[0] & 0xff) * 256 + r_buf[1]

        return int(data)

from micropython import const
from machine import Pin, UART
from time import sleep_ms


class DFPlayerPro:
    """
    MicroPython class for communication with the DFRobot DFPlayer Pro over UART.
    """

    DELAY_SEND_COMMAND = const(100)
    DELAY_TEST_CONNECTION = const(500)

    def __init__(self, tx_pin: int, rx_pin: int, baudrate: int = 115200, uart_id: int = 1):
        """
        Initialize the UART object.

        :param tx_pin: The GPIO number used for TX (transmit) communication.
        :type tx_pin: int
        :param rx_pin: The GPIO number used for RX (receive) communication.
        :type rx_pin: int
        :param baudrate: The desired baud rate for the UART communication (default: 115200).
        :type baudrate: int
        :param uart_id: The ID of the UART peripheral to use (default: 1).
        :type uart_id: int
        """
        uart = int(uart_id)
        baud = int(baudrate)
        tx = int(tx_pin)
        rx = int(rx_pin)

        self.uart = UART(uart, baudrate=baud, tx=Pin(tx), rx=Pin(rx))
        self.uart.init(baudrate=baudrate, bits=8, parity=None, stop=1)

    def _send_command(self, command: str) -> str:
        """
        Sends a command to a device and returns the response.

        :param command: The command to send.
        :type command: str
        :return: The response received from the device.
        :rtype: str
        """
        full_command = f"AT+{command}\r\n"

        print(f"[INFO] request: {full_command}")
        self.uart.write(full_command)
        sleep_ms(self.DELAY_SEND_COMMAND)

        response = self.uart.read()
        print(f"[INFO] response: {response}")
        return response.decode('utf-8') if response else None

    def test_connection(self):
        """
        Test the connection by sending an AT command via UART.

        :return: The response received from the device as a decoded string, or None if no response was received.
        """
        self.uart.write("AT\r\n")
        sleep_ms(self.DELAY_TEST_CONNECTION)

        response = self.uart.read()
        print(f"[INFO] response: {response}")
        return response.decode('utf-8') if response else None

    def set_volume(self, volume: int) -> str:
        """
        Set the volume of the device.

        :param volume: The volume level to set.
        :return: The response from sending the volume command.
        :rtype: str
        :raises ValueError: If the volume is not between 0 and 30.
        """
        if volume < 0 or volume > 30:
            raise ValueError("[ERROR] Volume must be between 0 and 30")

        return self._send_command(f"VOL={volume}")

    def set_play_mode(self, mode: int) -> str:
        """
        Set the play mode of the media player.
            1: repeat one song
            2: repeat all
            3: play one song and pause
            4: Play randomly
            5: Repeat all in the folder

        :param mode: The play mode to set. Valid values are: 1, 2, 3, 4, and 5.
        :return: The response message from the media player.
        :rtype: str
        :raises: ValueError: If the provided play mode is invalid.
        """
        if int(mode) not in [1, 2, 3, 4, 5]:
            raise ValueError("[ERROR] Invalid play mode")

        return self._send_command(f"PLAYMODE={mode}")

    def set_baudrate(self, baudrate: int) -> str:
        """
        Set UART baudrate

        :param baudrate: The baud rate to be set. Valid values are: 9600, 19200, 38400, 57600, 115200.
        :return: A string indicating the success of the operation.
        :raises ValueError: If the provided baud rate is invalid.
        """
        if int(baudrate) not in [9600, 19200, 38400, 57600, 115200]:
            raise ValueError("[ERROR] Invalid baudrate")

        return self._send_command(f"BAUDRATE={baudrate}")

    def query_volume(self) -> str:
        """
        Query the volume level in the device.

        :return: The volume level as a string.
        :rtype: str
        """
        return self._send_command("VOL=?")

    def query_play_mode(self) -> str:
        """
        Query the play mode of the device.

        :return: The current play mode of the device.
        :rtype: str
        """
        return self._send_command("PLAYMODE=?")

    def query_playing_file_number(self) -> str:
        """
        Query the playing file number.

        :return: The playing file number.
        :rtype: str
        """
        return self._send_command("QUERY=1")

    def query_total_file_count(self) -> str:
        """
        Query the total file count.

        :return: The total file count as a string.
        :rtype: str
        """
        return self._send_command("QUERY=2")

    def query_played_time(self) -> str:
        """
        Query the played time.

        :return: The played time as a string.
        :rtype: str
        """
        return self._send_command("QUERY=3")

    def query_total_time(self) -> str:
        """
        Queries and returns the total time.

        :return: The total time as a string.
        :rtype: str
        """
        return self._send_command("QUERY=4")

    def query_playing_file_name(self) -> str:
        """
        Returns the name of the currently playing file.

        :return: The name of the currently playing file as a string.
        :rtype: str
        """
        return self._send_command("QUERY=5")

    def play_next(self) -> str:
        """
        Plays the next track in the playlist.

        :return: The response from the "PLAY=NEXT" command.
        :rtype: str
        """
        return self._send_command("PLAY=NEXT")

    def play_last(self) -> str:
        """
        Plays the previous track in the playlist.

        :return: The response from the "PLAY=LAST" command.
        :rtype: str
        """
        return self._send_command("PLAY=LAST")

    def play_pause(self) -> str:
        """
        Play or pause the playback.

        :return: The result of the "PLAY=PP" command.
        :rtype: str
        """
        return self._send_command("PLAY=PP")

    def fast_rewind(self, seconds: int) -> str:
        """
        Fast Rewind the playback by a specified number of seconds.

        :param seconds: The number of seconds to rewind the media.
        :type seconds: int
        :return: The result of the "TIME=-{seconds}" command.
        :rtype: str
        """
        sec = int(seconds)

        return self._send_command(f"TIME=-{sec}")

    def fast_forward(self, seconds: int) -> str:
        """
        Fast forwards the playback by the specified number of seconds.

        :param seconds: The number of seconds to forward the media.
        :type seconds: int
        :return: The result of the "TIME=+{seconds}" command.
        :rtype: str
        """
        sec = int(seconds)

        return self._send_command(f"TIME=+{sec}")

    def start_from_second(self, second: int) -> str:
        """
        Starts the playback by a specified number of seconds.

        :param second: The second to start from as an integer.
        :type second: int
        :return: The result of the "TIME={second}" command.
        :rtype: str
        """
        sec = int(second)

        return self._send_command(f"TIME={sec}")

    def play_file_by_number(self, number: int) -> str:
        """
        Play a file by its number.

        :param number: The number of the file to be played.
        :type number: int
        :return: The result of the "PLAYNUM={number}" command.
        :rtype: str
        """
        num = int(number)

        return self._send_command(f"PLAYNUM={num}")

    def play_file_by_path(self, path: str) -> str:
        """
        Plays a file by the given file path.

        :param path: The file path of the file to play.
        :type path: str
        :return: The result of the "PLAYFILE={path}" command.
        :rtype: str
        """
        file_path = str(path)

        return self._send_command(f"PLAYFILE={file_path}")