Pan-Tilt Playground: Part 3 - Bluetooth gamepad controller

# change into project root directory
$ cd Pan-Tilt/

# create new directory
$ mkdir config

# create new file controllers.ini
$ touch config/controllers.ini

# create new directory inside libs
$ mkdir libs/controller

# create new file configuration.py
$ touch libs/configuration.py

# create new file bluetooth.py
$ touch libs/controller/bluetooth.py

# run tree command (optional)
$ tree .
.
|-- config
|   `-- controllers.ini
|-- libs
|   |-- configuration.py
|   |-- controller
|   |   `-- bluetooth.py
|   `-- servo.py
`-- main.py

3 directories, 5 files

[bluetooth]
enabled = true
mac = 
name = 8BitDo Zero 2 gamepad
btn_a = 304
btn_b = 305
btn_x = 307
btn_y = 308
btn_tl = 310
btn_tr = 311

from sys import exit
from pathlib import Path
from configparser import ConfigParser
from typing import Union


def parse_value(value: str) -> Union[bool, int, float, str]:
    """
    Parse the given string and convert it into its appropriate boolean, integer, or
    string value based on its content.

    :param value: Input string to be parsed.
    :type value: str
    :return: Parsed value as a boolean, integer, or string based on content.
    :rtype: Union[bool, int, float, str]
    """
    if value.lower() in {"true", "false"}:
        return value.lower() == "true"
    elif value.isdigit():
        return int(value)

    try:
        return float(value)
    except ValueError:
        return value


def load_configuration(path: str) -> dict:
    """
    Loads a configuration ini-file and processes sections to extract enabled data.

    :param path: Path to the configuration file relative to the current script directory
    :type path: str
    :return: A dictionary with key/value of the enabled section.
    :rtype: dict
    """
    enabled_section = None
    enabled_data = {}

    configuration_file = Path(__file__).parent.parent / path

    if not configuration_file.exists():
        print(f'[ERROR] Configuration file not found at {configuration_file}. Exiting...')
        exit(10)

    config = ConfigParser()
    config.read(configuration_file)

    for section in config.sections():
        if config.getboolean(section, 'enabled', fallback=False):
            if enabled_section:
                print(f'[ERROR] Multiple sections are enabled. Exiting...')
                exit(2)
            enabled_section = section
            enabled_data = {key: parse_value(value) for key, value in config.items(section)}

    if not enabled_section:
        print(f'[ERROR] No section is enabled in {configuration_file}. Exiting...')
        exit(3)

    enabled_data.pop('enabled', None)

    return {enabled_section: enabled_data}

from subprocess import run, CalledProcessError
from evdev import InputDevice, list_devices, ecodes
from threading import Thread, Lock
from typing import Optional


class BluetoothControllerHandler:
    """
    This class provides functionality to manage the lifecycle and input events of a game
    controller.
    """
    def __init__(self, name: str, mac: str, buttons: dict):
        """
        Represents a game controller interface for managing controller states and
        inputs.

        :param name: The name of the controller.
        :type name: str
        :param mac: The MAC address of the controller.
        :type mac: str
        :param buttons: A dictionary containing button configuration.
        :type buttons: dict
        """
        self._btn_status = {'left': False, 'right': False, 'down': False, 'up': False}

        self._name = name
        self._mac = mac
        self._buttons = buttons

        if not self._is_controller_connected():
            self._connect_controller()

        self._device_path = self._get_device_path()
        if not self._device_path:
            raise RuntimeError(f'[ERROR] No device path found for {self._name}')

        self._device = InputDevice(self._device_path)
        self._lock = Lock()
        self._thread = Thread(target=self._event_listener, daemon=True)
        self._thread.start()

    def _get_device_path(self) -> Optional[str]:
        """
        Determines the device path for an input device with a specific name.

        :return: The path of the input device if a matching device is found, or None.
        :rtype: Optional[str]
        """
        devices = [InputDevice(path) for path in list_devices()]

        for device in devices:
            if self._name in device.name:
                return device.path

        return None

    def _is_controller_paired(self) -> bool:
        """
        Checks if the Bluetooth controller is paired with the specified device.

        :raises CalledProcessError: Raised when a subprocess command fails.
        :return: Returns if the controller is paired.
        :rtype: bool
        """
        try:
            result = run(["bluetoothctl", "info", self._mac], check=True, capture_output=True, text=True)
            return "Paired: yes" in result.stdout
        except CalledProcessError:
            return False

    def _is_controller_connected(self) -> bool:
        """
        Determines if the Bluetooth controller is connected.

        :raises CalledProcessError: Raised when a subprocess command fails.
        :return: Returns if the controller is connected.
        :rtype: bool
        """
        try:
            result = run(["bluetoothctl", "info", self._mac], check=True, capture_output=True, text=True)
            return "Connected: yes" in result.stdout
        except CalledProcessError:
            return False

    def _connect_controller(self) -> bool:
        """
        This method attempts to establish a Bluetooth connection with a controller using its
        MAC address. If the controller is not paired, it initiates the pairing and trust process
        before attempting the connection.

        :raises CalledProcessError: Raised when a subprocess command fails.
        :return: Returns if the connection is successfully established.
        :rtype: bool
        """
        try:
            if not self._is_controller_paired():
                print(f'[INFO] Start Bluetooth Pairing.')
                run(["bluetoothctl", "pair", self._mac], check=True)
                run(["bluetoothctl", "trust", self._mac], check=True)

            print(f'[INFO] Start Bluetooth Connection.')
            run(["bluetoothctl", "connect", self._mac], check=True)
            return True
        except CalledProcessError as err:
            print(f'[ERROR] {err}')
            return False

    def _event_listener(self) -> None:
        """
        Listens to input device events and updates button state accordingly.

        :return: None
        """
        try:
            for event in self._device.read_loop():
                if event.type == ecodes.EV_KEY:
                    if event.code == self._buttons['btn_a']:
                        if event.value == 1:
                            self._btn_status['left'] = True
                        else:
                            self._btn_status['left'] = False

                    if event.code == self._buttons['btn_b']:
                        if event.value == 1:
                            self._btn_status['right'] = True
                        else:
                            self._btn_status['right'] = False

                    if event.code == self._buttons['btn_x']:
                        if event.value == 1:
                            self._btn_status['up'] = True
                        else:
                            self._btn_status['up'] = False

                    if event.code == self._buttons['btn_y']:
                        if event.value == 1:
                            self._btn_status['down'] = True
                        else:
                            self._btn_status['down'] = False

        except Exception as err:
            print(f'[ERROR] Controller event listener error: {err}')

    def get_status(self) -> dict:
        """
        Retrieves the current button status.

        :return: The current button status.
        :rtype: dict
        """
        with self._lock:
            return self._btn_status.copy()

from sys import exit
from time import sleep
from typing import Optional
from pinpong.board import Board
from libs.configuration import load_configuration
from libs.controller.bluetooth import BluetoothControllerHandler
from libs.servo import DFServo300


PAN_PIN: int = 2
PAN_END_ANGLE: int = 180
PAN_START_ANGLE: int = PAN_END_ANGLE // 2
PAN_SERVO_STEP: int = 4
TILT_PIN: int = 3
TILT_END_ANGLE: int = 90
TILT_START_ANGLE: int = TILT_END_ANGLE // 2
TILT_SERVO_STEP: int = 2
LONG_DELAY: float = .5
SHORT_DELAY: float = .015

CONFIGURATION_PATH: str = 'config/controllers.ini'


def move_servo(servo: DFServo300, nth: int, step: int, clockwise: bool = True, delay: Optional[float] = None) -> None:
    """
    Moves the servo motor by a specified step in a given direction, ensuring the
    angle remains within the permissible range.

    :param servo: An instance of the targeted servo motor.
    :type servo: DFServo300
    :param nth: Maximum permissible angle for the servo's operation.
    :type : int
    :param step: Incremental angle to move the servo.
    :type step: int
    :param clockwise: Boolean indicating the direction of rotation (default: True).
    :type clockwise: bool
    :param delay: Optional time in seconds to wait after moving the servo (default: None).
    :type delay: Optional[float]
    :return: None
    """
    current_angle = servo.get_angle()
    new_angle = current_angle + step if clockwise else current_angle - step

    if 0 <= new_angle <= nth and new_angle != current_angle:
        servo.angle(new_angle)

        if delay:
            sleep(delay)


if __name__ == '__main__':
    Board("UNIHIKER").begin()

    config_data = load_configuration(path=CONFIGURATION_PATH)
    button_mappings = {
        key: int(value) for key, value in config_data["bluetooth"].items() if key.startswith("btn_")
    }

    controller = BluetoothControllerHandler(name=config_data['bluetooth']['name'],
                                            mac=config_data['bluetooth']['mac'],
                                            buttons=button_mappings)

    pan = DFServo300(PAN_PIN)
    tilt = DFServo300(TILT_PIN)

    try:
        print('[INFO] Move to initial position...')
        pan.angle(PAN_START_ANGLE)
        tilt.angle(TILT_START_ANGLE)
        sleep(LONG_DELAY)

        while True:
            controller_status = controller.get_status()

            if controller_status['left'] and not controller_status['right']:
                move_servo(servo=pan, nth=PAN_END_ANGLE, step=PAN_SERVO_STEP)

            if controller_status['right'] and not controller_status['left']:
                move_servo(servo=pan, nth=PAN_END_ANGLE, clockwise=False, step=PAN_SERVO_STEP)

            if controller_status['down'] and not controller_status['up']:
                move_servo(servo=tilt, nth=TILT_END_ANGLE, clockwise=False, step=TILT_SERVO_STEP)

            if controller_status['up'] and not controller_status['down']:
                move_servo(servo=tilt, nth=TILT_END_ANGLE, step=TILT_SERVO_STEP)

            sleep(SHORT_DELAY)

    except KeyboardInterrupt:
        print('[INFO] KeyboardInterrupt triggered...')
    finally:
        print('[INFO] Move to reset position...')
        pan.reset()
        tilt.reset()
        sleep(LONG_DELAY)

        exit(0)

# start SSH connection to Unihiker
$ ssh [email protected]

# start hcitool scan
$ hcitool lescan

# run evdev.evtest (on Unihiker)
$ python3 -m evdev.evtest