How to discover, connect and receive data from Bluetooth in Android using Pyjnius

[tamalmaity-dev]

Question --

Here I want to discover, scan, and non-pair devices , and transfer data in Android....
I am working with Pyjnius ....here I get the paired Bluetooth devices ... but not able to make the non-paired devices and to transfer/ receive data ....Here can someone give me the for BlueBluetooth Classic and BLE ....please 🙏

Code

from jnius import autoclass

# Get BluetoothAdapter instance
BluetoothAdapter = autoclass('android.bluetooth.BluetoothAdapter')
bluetooth_adapter = BluetoothAdapter.getDefaultAdapter()

if bluetooth_adapter is None:
    print("Bluetooth not supported on this device")
else:
    if not bluetooth_adapter.isEnabled():
        print("Bluetooth is disabled. Please enable it.")
    else:
        print("Bluetooth is enabled.")

        # Get paired devices
        paired_devices = bluetooth_adapter.getBondedDevices()
        for device in paired_devices.toArray():
            print(f"Device Name: {device.getName()}, MAC Address: {device.getAddress()}")

[tamalmaity-dev]

I got the paired devices but not discovering the devices like Available devices in runtime scanning...
and not received the data .....please check my code ....and i make this for Bluetooth Classics but I also want for BLE devices also...

Here my full code:

import flet as ft
import flet_permission_handler as fph
import time
import asyncio
from jnius import autoclass
from typing import Callable, Optional

# Android Bluetooth classes
BluetoothAdapter = autoclass('android.bluetooth.BluetoothAdapter')
BluetoothDevice = autoclass('android.bluetooth.BluetoothDevice')
BluetoothSocket = autoclass('android.bluetooth.BluetoothSocket')
UUID = autoclass('java.util.UUID')

# Define the Java interface implementation
class BluetoothReceiver:
    def __init__(self, callback):
        self.callback = callback
        self.registered = False
        
    def register(self, activity, filter):
        # This is a simplified implementation without using PythonJavaClass
        # We'll use a different approach for Android interop
        pass
        
    def onReceive(self, device):
        if device:
            self.callback(device)

class BluetoothManager:
    def __init__(self, page):
        self.page = page
        self.bluetooth_adapter = BluetoothAdapter.getDefaultAdapter()
        self.devices = {}
        self.connected_socket = None
        self.connected_device = None
        self.is_scanning = False
        self.is_connected = False
        self.receiver_task = None
        self.uuid = UUID.fromString("00001101-0000-1000-8000-00805F9B34FB")  # Standard SerialPortService ID
        self.on_data_received: Optional[Callable[[str], None]] = None  # Properly typed callback
        self.receiver = None

    def check_bluetooth_enabled(self):
        if not self.bluetooth_adapter:
            return False
        return self.bluetooth_adapter.isEnabled()

    def enable_bluetooth(self):
        if not self.bluetooth_adapter:
            return False
        return self.bluetooth_adapter.enable()

    async def scan_devices(self):
        try:
            if not self.check_bluetooth_enabled():
                self.enable_bluetooth()
                await asyncio.sleep(1)  # Give time for Bluetooth to enable
                
            self.is_scanning = True
            self.devices.clear()
            
            # Get paired devices
            try:
                paired_devices = self.bluetooth_adapter.getBondedDevices().toArray()
                for device in paired_devices:
                    self.devices[device.getAddress()] = {
                        "name": device.getName() or "Unknown Device",
                        "address": device.getAddress(),
                        "paired": True,
                        "device": device
                    }
            except Exception as e:
                print(f"Error getting paired devices: {e}")
            
            # Instead of using a custom receiver, we'll just use the paired devices
            # This is more stable and less likely to crash
            
            # Wait a moment to ensure we've collected all paired devices
            await asyncio.sleep(2)
            
            self.is_scanning = False
            return self.devices
            
        except Exception as e:
            print(f"Scan error: {e}")
            self.is_scanning = False
            return self.devices

    async def connect_to_device(self, device_address):
        if self.is_connected:
            self.disconnect()
            
        device = self.devices[device_address]["device"]
        self.bluetooth_adapter.cancelDiscovery()  # Always cancel discovery before connecting
        
        try:
            socket = device.createRfcommSocketToServiceRecord(self.uuid)
            socket.connect()
            self.connected_socket = socket
            self.connected_device = device
            self.is_connected = True
            
            # Start a task to receive data
            self.receiver_task = asyncio.create_task(self.receive_data())
            
            return True
        except Exception as e:
            print(f"Connection error: {e}")
            return False

    def disconnect(self):
        if self.connected_socket:
            try:
                self.connected_socket.close()
            except:
                pass
        self.connected_socket = None
        self.connected_device = None
        self.is_connected = False
        self.receiver_task = None

    def send_data(self, data):
        if not self.is_connected or self.connected_socket is None:
            print("Cannot send data: No active connection")
            return False
        
        try:
            output_stream = self.connected_socket.getOutputStream()
            output_stream.write(data.encode())
            output_stream.flush()
            return True
        except Exception as e:
            print(f"Send error: {e}")
            self.disconnect()
            return False

    async def receive_data(self):
        if not self.is_connected or self.connected_socket is None:
            print("Cannot receive data: No active connection")
            return
            
        try:
            input_stream = self.connected_socket.getInputStream()
            buffer = bytearray(1024)
            
            while self.is_connected and self.connected_socket is not None:
                try:
                    bytes_read = input_stream.read(buffer)
                    if bytes_read > 0:
                        received_data = buffer[:bytes_read].decode('utf-8')
                        # Update UI with received data
                        self.update_received_data(received_data)
                    await asyncio.sleep(0.1)  # Small delay to prevent CPU hogging
                except Exception as e:
                    print(f"Receive error: {e}")
                    self.disconnect()
                    break
        except Exception as e:
            print(f"Error setting up receiver: {e}")
            self.disconnect()

    def update_received_data(self, data):
        # This method will be called from the UI thread
        if hasattr(self, 'on_data_received') and callable(self.on_data_received):
            self.on_data_received(data)


class BluetoothApp:
    # Update the button definitions to use async handlers
    def __init__(self, page: ft.Page):
        self.page = page
        self.page.title = "Bluetooth Classic Scanner"
        self.page.theme_mode = ft.ThemeMode.LIGHT
        self.page.padding = 20
        
        self.bt_manager = BluetoothManager(page)
        self.bt_manager.on_data_received = self.on_data_received
        
        # Initialize permission handler
        self.permission_handler = fph.PermissionHandler()
        self.page.overlay.append(self.permission_handler)
        
        # Define Bluetooth permissions
        self.bluetooth_permissions = [
            fph.PermissionType.BLUETOOTH,
            fph.PermissionType.BLUETOOTH_SCAN,
            fph.PermissionType.BLUETOOTH_ADVERTISE,
            fph.PermissionType.BLUETOOTH_CONNECT,
            fph.PermissionType.LOCATION,
        ]
        
        # UI Components
        self.status_text = ft.Text("Click 'Check All Bluetooth Permissions' to begin", size=16)
        self.scan_button = ft.ElevatedButton(
            "Scan Devices",
            bgcolor=ft.colors.BLUE,
            color=ft.colors.WHITE,
            on_click=self.scan_devices
        )
        self.scan_button.disabled = True
        
        # Permission buttons with proper class methods
        self.check_permissions_button = ft.OutlinedButton(
            "Check All Bluetooth Permissions",
            on_click=self.check_multiple_permissions,
        )
        
        self.request_permissions_button = ft.OutlinedButton(
            "Request All Bluetooth Permissions",
            on_click=self.request_multiple_permissions,
        )
        
        self.settings_button = ft.OutlinedButton(
            "Open App Settings",
            on_click=self.open_app_settings,
        )
        
        # Snackbar for notifications
        self.snackbar = ft.SnackBar(
            content=ft.Text(""),
            bgcolor=ft.colors.GREEN,
            action="DISMISS"
        )
        self.page.overlay.append(self.snackbar)
        
        self.devices_list = ft.ListView(
            expand=1,
            spacing=10,
            padding=20,
        )
        
        self.connection_status = ft.Text("Not connected", color=ft.colors.RED)
        
        self.received_data = ft.TextField(
            label="Received Data",
            multiline=True,
            read_only=True,
            min_lines=3,
            max_lines=5,
            expand=True
        )
        
        self.send_data_field = ft.TextField(
            label="Data to Send",
            expand=True
        )
        
        self.send_button = ft.ElevatedButton(
            "Send",
            on_click=self.send_data,
            disabled=True
        )
        
        # Layout
        self.page.add(
            ft.Container(
                expand=True,
                content=ft.ListView(
                    expand=True,
                    auto_scroll=True,
                    spacing=10,
                    padding=20,
                    controls=[
                        ft.Column([
                            ft.Text("Bluetooth Classic Scanner", size=28, weight=ft.FontWeight.BOLD),
                            
                            # Permission controls
                            ft.Column([
                                self.check_permissions_button,
                                self.request_permissions_button,
                                self.settings_button
                            ], alignment=ft.MainAxisAlignment.CENTER, spacing=6),
                            
                            self.status_text,
                            ft.Row([self.scan_button], alignment=ft.MainAxisAlignment.CENTER),
                            ft.Container(
                                content=self.devices_list,
                                border=ft.border.all(1, ft.colors.GREY_400),
                                border_radius=10,
                                padding=10,
                                expand=True
                            ),
                            self.connection_status,
                            ft.Container(
                                content=self.received_data,
                                margin=ft.margin.only(top=10)
                            ),
                            ft.Row(
                                [self.send_data_field, self.send_button],
                                alignment=ft.MainAxisAlignment.SPACE_BETWEEN
                            )
                        ], spacing=20, horizontal_alignment=ft.CrossAxisAlignment.CENTER)
                    ]
                ),
               
            )
        )







    # Add these permission methods to the class
    def check_multiple_permissions(self, e):
        results = []
        for permission in self.bluetooth_permissions:
            status = self.permission_handler.check_permission(permission)
            results.append(f"{permission.name}: {status}")
        
        # Update status text with permission check results
        self.status_text.value = "Checked permissions:\n" + "\n".join(results)
        self.page.update()
        
        # Enable scan button if all permissions are granted
        all_granted = all(self.permission_handler.check_permission(p) for p in self.bluetooth_permissions)
        self.scan_button.disabled = not all_granted
        self.page.update()

    def request_multiple_permissions(self, e):
        results = []
        for permission in self.bluetooth_permissions:
            status = self.permission_handler.request_permission(permission)
            results.append(f"{permission.name}: {status}")
            
        # Update status text with permission request results
        self.status_text.value = "Requested permissions:\n" + "\n".join(results)
        self.page.update()
        
        # Enable scan button if all permissions are granted
        all_granted = all(self.permission_handler.check_permission(p) for p in self.bluetooth_permissions)
        self.scan_button.disabled = not all_granted
        self.page.update()



    def open_app_settings(self, e):
        self.permission_handler.open_app_settings()
        self.show_snackbar("Opening app settings", ft.colors.BLUE)
    
    def show_snackbar(self, message, color=ft.colors.GREEN):
        # Replace the content with a new Text control instead of modifying the value
        self.snackbar.content = ft.Text(message)
        self.snackbar.bgcolor = color
        self.snackbar.open = True
        self.page.update()

    async def scan_devices(self, e=None):
        self.status_text.value = "Scanning for devices..."
        self.scan_button.disabled = True
        self.devices_list.controls.clear()
        self.page.update()
        
        # Run the scan in the background
        self.page.run_task(self.scan_devices_background)
    
    async def scan_devices_background(self):
        devices = await self.bt_manager.scan_devices()
        await self.update_device_list(devices)

    async def update_device_list(self, devices):
        self.devices_list.controls.clear()
        
        if not devices:
            self.devices_list.controls.append(ft.Text("No devices found"))
            self.show_snackbar("No devices found", ft.colors.ORANGE)
        else:
            for address, device_info in devices.items():
                name = device_info["name"]
                paired = "Paired" if device_info["paired"] else "Not Paired"
                
                # Create a closure to properly handle the async call
                async def connect_wrapper(e, addr=address):
                    await self.connect_device(addr)
                
                self.devices_list.controls.append(
                    ft.Container(
                        content=ft.Column([
                            ft.Text(name, weight=ft.FontWeight.BOLD),
                            ft.Text(f"{address} ({paired})", size=12)
                        ]),
                        padding=10,
                        border_radius=5,
                        bgcolor=ft.colors.BLUE_50,
                        on_click=connect_wrapper
                    )
                )
            self.show_snackbar(f"Found {len(devices)} devices", ft.colors.GREEN)
        
        self.status_text.value = "Scan complete"
        self.scan_button.disabled = False
        self.page.update()

    # Make connect_device async so it can be awaited
    async def connect_device(self, device_address):
        self.status_text.value = f"Connecting to {self.bt_manager.devices[device_address]['name']}..."
        self.page.update()
        
        success = await self.bt_manager.connect_to_device(device_address)
        self.update_connection_status(success, device_address)

    def update_connection_status(self, success, device_address):
        if success:
            device_name = self.bt_manager.devices[device_address]["name"]
            self.status_text.value = f"Connected to {device_name}"
            self.connection_status.value = f"Connected to {device_name}"
            self.connection_status.color = ft.colors.GREEN
            self.send_button.disabled = False
            self.show_snackbar(f"Connected to {device_name}", ft.colors.GREEN)
        else:
            self.status_text.value = "Connection failed"
            self.connection_status.value = "Connection failed"
            self.connection_status.color = ft.colors.RED
            self.send_button.disabled = True
            self.show_snackbar("Connection failed", ft.colors.RED)
        
        self.page.update()

    def send_data(self, e):
        data = self.send_data_field.value
        if not data:
            return
            
        success = self.bt_manager.send_data(data)
        if success:
            self.send_data_field.value = ""
            self.show_snackbar("Data sent successfully", ft.colors.GREEN)
            self.page.update()
        else:
            self.status_text.value = "Failed to send data"
            self.show_snackbar("Failed to send data", ft.colors.RED)
            self.page.update()

    def on_data_received(self, data):
        current = self.received_data.value or ""
        self.received_data.value = current + data
        self.page.update()


def main(page: ft.Page):
    page.theme_mode = ft.ThemeMode.SYSTEM
    page.padding = 0
    page.appbar = ft.AppBar(
        title=ft.Text("Bluetooth Classic Scanner"),
        center_title=True,
        bgcolor=ft.colors.BLUE
    )
    app = BluetoothApp(page)


ft.app(target=main, view=ft.AppView.FLET_APP)