bluetoothctl_communicate

 avatar
unknown
python
a year ago
3.1 kB
11
Indexable
    @staticmethod
    def communicate(write_uuid, read_uuid, data):
        try:
            """
            Requires a writable and/or a readable GATT uuid and data to be written
            The method returns the data retrieved from the readable GATT uuid
            """

            BluetoothControl._check_service_and_adapter_running()
            BluetoothControl._check_connection_state()

            # Converting data to HEX values
            hex_string = ''
            for char in data:
                hex_string += hex(ord(char)) + ' '

            # # Starting a pseudo terminal to read
            read_process = ptyprocess.PtyProcessUnicode.spawn(['/bin/bash'])
            read_process.setwinsize(100,100)
            commands = ['bluetoothctl', 'menu gatt', f'select-attribute {read_uuid}', 'notify on']
            for command in commands:
                for char in command:
                    read_process.write(char)
                read_process.write('\n')
                read_process.flush()

            # Starting a pseudo terminal to write
            write_process = ptyprocess.PtyProcessUnicode.spawn(['/bin/bash'])
            commands = f'bluetoothctl <<EOF\nmenu gatt\nselect-attribute {write_uuid}\nwrite "{hex_string}"\nEOF\n'
            write_process.write(commands)
            write_process.flush()

            # Read until new line ( hex: 0a )
            lines = ''
            for char in range(5000):
                lines += read_process.read(1)
                if lines[char] == 'a':
                    if lines[char - 1] == '0':
                        if lines[char - 2] == ' ':
                            break
                if lines[char] == '':
                    break
            lines = lines.splitlines()

            # Get the important part of the response
            response = []
            append = False
            for line_num, line in enumerate(lines):
                if "Value" in line:
                    append = True
                if append:
                    response.append(line)

            # Filter the HEX values
            hex_values = []
            matches = []
            for line in response:
                pattern1 = r'  (.*?)  '
                matches.extend(re.findall(pattern1, line))
                pattern2 = r'  .*?0a'
                matches.extend(re.findall(pattern2, line))
            for match in matches:
                for number in str(match).split():
                    hex_values.append(number)

            # Convert HEX values to an ASCII string
            ascii_string = ''.join(chr(int(hex_val, 16)) for hex_val in hex_values)

            logging.debug(f"Writing GATT Characteristic:\n\t"
                          f"Write UUID: {write_uuid}\n\t"
                          f"Read UUID: {read_uuid}\n\t"
                          f"Data: {repr(data)}\n\t"
                          f"HEX: {hex_string}\n\t"
                          f"Response: {ascii_string}")

            return ascii_string
        except Exception as e:
            logging.error(f'Unable to communicate with Bluetooth Device\nException : {e}')
Editor is loading...
Leave a Comment