2

I'm extremely new to Raspberry Pi and Bluetooth in general and have been having some trouble and confusion connecting between my computer and Pi programmatically.

I'm trying to open a Bluetooth connection from my Laptop/Phone/etc. to my Raspberry Pi. This probably isn't relevant but I'm using PyBluez on the Pi (python), and InTheHand.Net.Bluetooth on my Laptop (.NET).

On my Raspberry Pi, I run roughly the following Python code:

import bluetooth

socket = bluetooth.BluetoothSocket()
socket.bind(('', 1))
socket.listen(1)
client_socket, addr =  socket.accept()
# etc... send/receive messages and stuff

However, I was getting stuck in that when trying to connect to the Pi, after scanning and obtaining its address, I was getting errors like 'A socket operation was attempted to an unreachable network'. However, once I paired the two devices, I was suddenly able to connect and send messages perfectly.

My question is, why am I required to pair the devices in the first place? Like the internet, if I have the address and port, shouldn't I be able to connect and send/receive messages? I realize that the answer probably involves some knowledge about the Bluetooth protocol that isn't relevant to this site, so I'll add: would it be possible to connect without having to pair? Maybe something with bluetoothctl or something?

Any help or incite would be very much appreciated.

1
  • @ukBaz I think you've misread the question, I got the script to work after pairing. I am able to pair without any issues. I wanted to know if it was possible to open a connection without having to first pair.
    – nreh
    Jan 2 at 9:37

1 Answer 1

4

I am assuming it is a Serial Port Profile (SPP) connection that you are trying to establish between the two machines. The Linux Bluetooth stack uses BlueZ which has an API for setting up classic Profiles such as SPP. This is documented at:

https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/doc/profile-api.txt

This has the option to set RequireAuthentication to false so that connection to the server is allowed without pairing.

In the BlueZ source tree there is a Python test script that shows how to use this API. The test script will need to be modified to use the required values for your situation.


UPDATE: Include example

I've written the following test to prove that I could connect two Raspberry Pi's together using SPP without the need for pairing or authentication.

The server reverses the received values and then sends it back. It is io_cb method that you would need to modify for your own purposes.

I have used PyGObject (gi.repository) because the python-dbus library used in the BlueZ example/test has been deprecated. This does make the example a little bit more verbose but as I said it should only be the io_cb that you need to adjust.

SERVER:

import os

from gi.repository import Gio, GLib

# Introspection data for DBus
profile_xml = """
<node>
    <interface name="org.bluez.Profile1">
        <method name="Release"/>
        <method name="NewConnection">
            <arg type="o" name="device" direction="in"/>
            <arg type="h" name="fd" direction="in"/>
            <arg type="a{sv}" name="fd_properties" direction="in"/>
        </method>
        <method name="RequestDisconnection">
            <arg type="o" name="device" direction="in"/>
        </method>
    </interface>
</node>
"""


class DbusService:
    """Class used to publish a DBus service on to the DBus System Bus"""
    def __init__(self, introspection_xml, publish_path):
        self.node_info = Gio.DBusNodeInfo.new_for_xml(introspection_xml).interfaces[0]
        # start experiment
        method_outargs = {}
        method_inargs = {}
        property_sig = {}
        for method in self.node_info.methods:
            method_outargs[method.name] = '(' + ''.join([arg.signature for arg in method.out_args]) + ')'
            method_inargs[method.name] = tuple(arg.signature for arg in method.in_args)
        self.method_inargs = method_inargs
        self.method_outargs = method_outargs

        self.con = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
        self.con.register_object(
            publish_path,
            self.node_info,
            self.handle_method_call,
            self.prop_getter,
            self.prop_setter)

    def handle_method_call(self,
                           connection: Gio.DBusConnection,
                           sender: str,
                           object_path: str,
                           interface_name: str,
                           method_name: str,
                           params: GLib.Variant,
                           invocation: Gio.DBusMethodInvocation
                           ):
        """
        This is the top-level function that handles method calls to
        the server.
        """
        args = list(params.unpack())
        # Handle the case where it is a Unix filedescriptor
        for i, sig in enumerate(self.method_inargs[method_name]):
            if sig == 'h':
                msg = invocation.get_message()
                fd_list = msg.get_unix_fd_list()
                args[i] = fd_list.get(args[i])
        func = self.__getattribute__(method_name)
        result = func(*args)
        if result is None:
            result = ()
        else:
            result = (result,)
        outargs = ''.join([_.signature
                           for _ in invocation.get_method_info().out_args])
        send_result = GLib.Variant(f'({outargs})', result)
        invocation.return_value(send_result)

    def prop_getter(self,
                    connection: Gio.DBusConnection,
                    sender: str,
                    object: str,
                    iface: str,
                    name: str):
        """Return requested values on DBus from Python object"""
        py_value = self.__getattribute__(name)
        signature = self.node_info.lookup_property(name).signature
        if py_value:
            return GLib.Variant(signature, py_value)
        return None

    def prop_setter(self,
                    connection: Gio.DBusConnection,
                    sender: str,
                    object: str,
                    iface: str,
                    name: str,
                    value: GLib.Variant):
        """Set values on Python object from DBus"""
        self.__setattr__(name, value.unpack())
        return True


class Profile(DbusService):

    def __init__(self, introspection_xml, publish_path):
        super().__init__(introspection_xml, publish_path)
        self.fd = -1

    def Release(self):
        print('Release')

    def NewConnection(self, path, fd, properties):
        self.fd = fd
        print(f'NewConnection({path}, {self.fd}, {properties})')
        for key in properties.keys():
            if key == 'Version' or key == 'Features':
                print('  %s = 0x%04x' % (key, properties[key]))
            else:
                print('  %s = %s' % (key, properties[key]))
        io_id = GLib.io_add_watch(self.fd,
                                  GLib.PRIORITY_DEFAULT,
                                  GLib.IO_IN | GLib.IO_PRI,
                                  self.io_cb)

    def io_cb(self, fd, conditions):
        data = os.read(fd, 1024)
        print('Callback Data: {0}'.format(data.decode('ascii')))
        os.write(fd, bytes(list(reversed(data.rstrip()))) + b'\n')
        return True

    def RequestDisconnection(self, path):
        print('RequestDisconnection(%s)' % (path))
        if self.fd > 0:
            os.close(self.fd)
            self.fd = -1


def main():
    obj_mngr = Gio.DBusObjectManagerClient.new_for_bus_sync(
        bus_type=Gio.BusType.SYSTEM,
        flags=Gio.DBusObjectManagerClientFlags.NONE,
        name='org.bluez',
        object_path='/',
        get_proxy_type_func=None,
        get_proxy_type_user_data=None,
        cancellable=None,
    )

    manager = obj_mngr.get_object('/org/bluez').get_interface('org.bluez.ProfileManager1')
    adapter = obj_mngr.get_object('/org/bluez/hci0').get_interface('org.freedesktop.DBus.Properties')
    mainloop = GLib.MainLoop()

    discoverable = adapter.Get('(ss)', 'org.bluez.Adapter1', 'Discoverable')

    if not discoverable:
        print('Making discoverable...')
        adapter.Set('(ssv)', 'org.bluez.Adapter1',
                    'Discoverable', GLib.Variant.new_boolean(True))

    profile_path = '/org/bluez/test/profile'
    server_uuid = '00001101-0000-1000-8000-00805f9b34fb'
    opts = {
        'Version': GLib.Variant.new_uint16(0x0102),
        'AutoConnect': GLib.Variant.new_boolean(True),
        'Role': GLib.Variant.new_string('server'),
        'Name': GLib.Variant.new_string('SerialPort'),
        'Service': GLib.Variant.new_string('00001101-0000-1000-8000-00805f9b34fb'),
        'RequireAuthentication': GLib.Variant.new_boolean(False),
        'RequireAuthorization': GLib.Variant.new_boolean(False),
        'Channel': GLib.Variant.new_uint16(1),
    }

    print('Starting Serial Port Profile...')

    profile = Profile(profile_xml, profile_path)

    manager.RegisterProfile('(osa{sv})', profile_path, server_uuid, opts)

    try:
        mainloop.run()
    except KeyboardInterrupt:
        mainloop.quit()


if __name__ == '__main__':
    main()

Below is the client I used on the second machine to write a word and check that the result came back reversed.

CLIENT:

import socket

server_address = "xx:xx:xx:xx:xx:xx"
server_port = 1
with socket.socket(socket.AF_BLUETOOTH,
                   socket.SOCK_STREAM,
                   socket.BTPROTO_RFCOMM) as c:

    c.connect((server_address, server_port))
    c.send(b"desserts")
    print(c.recv(1024).decode())

I also ran the above client on a Windows PC running Python 3.9 and it connected to the RPi running the server without being paired.

I'm sure the server code could be structured better but it was good enough for my test.

And remember, if you are prepared to live with having to pair the two machines once before you use them together for the first time then the server can be done with standard Python:

import socket

server_address = 'xx:xx:xx:xx:xx:xx'
server_port = 1
backlog = 1
size = 1024
while True:
    with socket.socket(socket.AF_BLUETOOTH,
                       socket.SOCK_STREAM,
                       socket.BTPROTO_RFCOMM) as s:
        s.bind((server_address, server_port))
        s.listen(backlog)
        print('Waiting for connection...')
        client, address = s.accept()
        print(f'Connection from {address}')
        while True:
            try:
                data = client.recv(size)
                if data:
                    print(data)
                    client.send(data)
            except ConnectionResetError:
                print('Client disconnected')
                break
4
  • Thank you for the help. The BlueZ API is extremely daunting to me and I have tried and failed to understand it before. Would it be possible to implement this with PyBlueZ?
    – nreh
    Jan 2 at 10:02
  • 1
    As it says on the PyBlueZ page, "This project is not under active development". This would make me very nervous about using that library so I have not investigated if it can be done with it. I have got the server to accept connections without pairing using the DBus API. I'll update my answer with the test I've built.
    – ukBaz
    Jan 2 at 15:28
  • Thank you! Just to clarify, what is the difference between GIO and BlueZ? Does GIO communicate with BlueZ, or are they both doing the same thing?
    – nreh
    Jan 2 at 22:43
  • Okay I spent the day reading up on dbus and how it is used to communicate with BlueZ, I think I understand now. Thanks so much for your help, I'll take a look at creating a profile.
    – nreh
    Jan 3 at 8:02

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.