Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit c9b09f2b authored by JohnLai's avatar JohnLai
Browse files

Floss: Support Connection type 

- Introduce a connection type to keep Bluetooth connection information.
- Remove duplicate Bluetooth Transport enum type.
- Fix typos.

Bug: 312653014
Test: mma packages/modules/Bluetooth && pts-bot GAP/BOND/BON/BV-01
Tag: #floss
Change-Id: I381fde05ce24faef3d2ba0a75f2b4a10aa0cc14b
parent b190fe32

floss/pandora/floss/floss_enums.py

+2 −9
Original line number Diff line number Diff line
@@ -19,8 +19,9 @@ import enum 
class BtTransport(enum.IntEnum):

    """Bluetooth transport type."""

    AUTO = 0

    BR_EDR = 1

    BREDR = 1

    LE = 2

    DUAL = 3





class GattWriteRequestStatus(enum.IntEnum):
@@ -142,14 +143,6 @@ class BondState(enum.IntEnum): 
    BONDED = 2





class Transport(enum.IntEnum):

    """Bluetooth transport type."""

    AUTO = 0

    BREDR = 1

    LE = 2

    DUAL = 3





class PairingVariant(enum.IntEnum):

    """Bluetooth pairing variant type."""

    # SSP variants.

floss/pandora/floss/utils.py

+30 −5
Original line number Diff line number Diff line
@@ -13,16 +13,17 @@ 
# limitations under the License.

"""All floss utils functions."""



from typing import List, Optional



import functools

import logging

import threading

import time

from typing import List, Optional



from floss.pandora.floss import floss_enums

from gi.repository import GLib

from google.protobuf import any_pb2

from pandora import host_pb2

from pandora_experimental import os_pb2



# All GLIB method calls should wait this many seconds by default

GLIB_METHOD_CALL_TIMEOUT = 2
@@ -343,15 +344,39 @@ class PropertySet: 
        return setter(*args)





class Connection:

    """A Bluetooth connection."""



    def __init__(self, address: str, transport: floss_enums.BtTransport):

        self.address = address

        self.transport = transport





def connection_to(connection: Connection):

    """Converts Connection from Floss format to gRPC format."""

    internal_connection_ref = os_pb2.InternalConnectionRef(address=address_to(connection.address),

                                                           transport=connection.transport)

    cookie = any_pb2.Any(value=internal_connection_ref.SerializeToString())

    return host_pb2.Connection(cookie=cookie)





def connection_from(connection: host_pb2.Connection):

    """Converts Connection from gRPC format to Floss format."""

    internal_connection_ref = os_pb2.InternalConnectionRef()

    internal_connection_ref.ParseFromString(connection.cookie.value)

    return Connection(address=address_from(internal_connection_ref.address),

                      transport=internal_connection_ref.transport)





def address_from(request_address: bytes):

    """Converts address from grpc server format to floss format."""

    """Converts address from gRPC format to Floss format."""

    address = request_address.hex()

    address = f'{address[:2]}:{address[2:4]}:{address[4:6]}:{address[6:8]}:{address[8:10]}:{address[10:12]}'

    return address.upper()





def address_to(address: str):

    """Converts address from floss format to grpc server format."""

    """Converts address from Floss format to gRPC format."""

    request_address = bytes.fromhex(address.replace(':', ''))

    return request_address
@@ -367,7 +392,7 @@ def uuid32_to_uuid128(uuid32: str): 
def advertise_data_from(request_data: host_pb2.DataTypes):

    """Mapping DataTypes to a dict.



    The dict content follows the format of floss AdvertiseData.

    The dict content follows the format of Floss AdvertiseData.



    Args:

        request_data : advertising data.

floss/pandora/server/bluetooth.py

+2 −3
Original line number Diff line number Diff line
@@ -13,7 +13,6 @@ 
# limitations under the License.

"""All functions relative to the bluetooth procedure."""



import asyncio

import logging

import threading

import traceback
@@ -177,8 +176,8 @@ class Bluetooth(object): 
    def get_address(self):

        return self.adapter_client.get_address()



    def get_remote_type(self):

        return self.adapter_client.get_remote_property('Type')

    def get_remote_type(self, address):

        return self.adapter_client.get_remote_property(address, 'Type')



    def is_connected(self, address):

        return self.adapter_client.is_connected(address)

floss/pandora/server/host.py

+9 −9
Original line number Diff line number Diff line
@@ -24,7 +24,6 @@ from floss.pandora.floss import floss_enums 
from floss.pandora.floss import scanner_client

from floss.pandora.floss import utils

from floss.pandora.server import bluetooth as bluetooth_module

from google.protobuf import any_pb2

from google.protobuf import empty_pb2

import grpc

from pandora import host_grpc_aio
@@ -179,7 +178,7 @@ class HostService(host_grpc_aio.HostServicer): 
                    name = utils.create_observer_name(observer)

                    self.bluetooth.adapter_client.register_callback_observer(name, observer)



                    if not self.bluetooth.create_bond(address, floss_enums.BtTransport.BR_EDR):

                    if not self.bluetooth.create_bond(address, floss_enums.BtTransport.BREDR):

                        raise RuntimeError('Failed to call create_bond.')



                    success, reason = await create_bond
@@ -192,8 +191,8 @@ class HostService(host_grpc_aio.HostServicer): 
            finally:

                self.bluetooth.adapter_client.unregister_callback_observer(name, observer)



        cookie = any_pb2.Any(value=utils.address_to(address))

        return host_pb2.ConnectResponse(connection=host_pb2.Connection(cookie=cookie))

        return host_pb2.ConnectResponse(

            connection=utils.connection_to(utils.Connection(address, floss_enums.BtTransport.BREDR)))



    async def WaitConnection(self, request: host_pb2.WaitConnectionRequest,

                             context: grpc.ServicerContext) -> host_pb2.WaitConnectionResponse:
@@ -229,8 +228,8 @@ class HostService(host_grpc_aio.HostServicer): 
                self.bluetooth.adapter_client.unregister_callback_observer(name, observer)

            self.waited_connections.add(address)



        cookie = any_pb2.Any(value=utils.address_to(address))

        return host_pb2.WaitConnectionResponse(connection=host_pb2.Connection(cookie=cookie))

        return host_pb2.WaitConnectionResponse(

            connection=utils.connection_to(utils.Connection(address, floss_enums.BtTransport.BREDR)))



    async def ConnectLE(self, request: host_pb2.ConnectLERequest,

                        context: grpc.ServicerContext) -> host_pb2.ConnectLEResponse:
@@ -239,7 +238,8 @@ class HostService(host_grpc_aio.HostServicer): 
        raise NotImplementedError('Method not implemented!')



    async def Disconnect(self, request: host_pb2.DisconnectRequest, context: grpc.ServicerContext) -> empty_pb2.Empty:

        address = utils.address_from(request.connection.cookie.value)

        address = utils.connection_from(request.connection).address



        if self.bluetooth.is_connected(address):

            self.bluetooth.disconnect_device(address)

        return empty_pb2.Empty()
@@ -382,8 +382,8 @@ class HostService(host_grpc_aio.HostServicer): 
                address = await connections.get()

                logging.info('Advertise: Connected to %s', address)



                cookie = any_pb2.Any(value=utils.address_to(address))

                yield host_pb2.AdvertiseResponse(connection=host_pb2.Connection(cookie=cookie))

                yield host_pb2.AdvertiseResponse(

                    connection=utils.connection_to(utils.Connection(address, floss_enums.BtTransport.LE)))



                # Wait a small delay before restarting the advertisement.

                await asyncio.sleep(1)

floss/pandora/server/security.py

+4 −4
Original line number Diff line number Diff line
@@ -141,9 +141,9 @@ class SecurityService(security_grpc_aio.SecurityServicer): 
                elif variant == floss_enums.PairingVariant.PIN_ENTRY:

                    transport = self.bluetooth.get_remote_type(address)



                    if transport == floss_enums.Transport.BREDR:

                    if transport == floss_enums.BtTransport.BREDR:

                        event.pin_code_request.CopyFrom(empty_pb2.Empty())

                    elif transport == floss_enums.Transport.LE:

                    elif transport == floss_enums.BtTransport.LE:

                        event.passkey_entry_request.CopyFrom(empty_pb2.Empty())

                    else:

                        logging.error('Cannot determine pairing variant from unknown transport.')
@@ -154,9 +154,9 @@ class SecurityService(security_grpc_aio.SecurityServicer): 
                    transport = self.bluetooth.get_remote_type(address)

                    [pincode] = variables



                    if transport == floss_enums.Transport.BREDR:

                    if transport == floss_enums.BtTransport.BREDR:

                        event.pin_code_notification = pincode.encode()

                    elif transport == floss_enums.Transport.LE:

                    elif transport == floss_enums.BtTransport.LE:

                        event.passkey_entry_notification = int(pincode)

                    else:

                        logging.error('Cannot determine pairing variant from unknown transport.')