Reading data from multiple devices on raspberry pi at the same time

[wahadatjan]

I am trying to read data from multiple devices where each device has its own service and characteristics but when i tried to read data it reads from one of them not all at the same time . when i run it on mac it works fine but not on linux installed on raspberry pi

import json
import asyncio
import struct

import requests
from bleak import BleakScanner, BleakClient
from django.shortcuts import render
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.middleware.csrf import get_token
from asgiref.sync import async_to_sync
from .services.atmotube_service import AtmotubeService
from .services.heart_rate_service import HeartRateService
import logging

from .mqtt_client import mqtt_client
import datetime

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='app.log',
    filemode='a'
)

logger = logging.getLogger(__name__)

# Initialize locks
heart_rate_lock = asyncio.Lock()
atmotube_lock = asyncio.Lock()

def scan_ble_devices(request):
    return render(request, 'ble_devices.html')

async def connect_matched_devices(matched_devices, csrf_token):
    connect_results = []
    for device in matched_devices:
        device_uuid = device.address
        device_type = device.name if device.name else 'Unknown'
        success = False
        error_message = ''
        try:
            async with BleakClient(device) as client:
                logger.info(f"Connected to device: {device_type} with UUID: {device_uuid}")
                connect_results.append({
                    'device_uuid': device_uuid,
                    'type': device_type,
                    'success': True,
                    'message': 'Device connected successfully',
                    'csrf_token': csrf_token
                })
                success = True
        except Exception as e:
            error_message = str(e)
            logger.error(f"Failed to connect to device: {device_type} with UUID: {device_uuid}. Error: {e}")

        if not success:
            connect_results.append({
                'device_uuid': device_uuid,
                'type': device_type,
                'success': False,
                'error': error_message
            })

    return connect_results

async def scan_devices():
    try:
        devices = await BleakScanner.discover()
        formatted_devices = [{'uuid': device.address, 'name': device.name if device.name else 'Unknown'} for device in devices]
        logger.info(f"Devices found: {formatted_devices}")
        return devices  # Return the actual BLEDevice objects
    except Exception as e:
        logger.error(f"Error occurred during scanning: {e}")
        return {'error': str(e)}

@csrf_exempt
def fetch_participant_ble_devices(request):
    if request.method == 'GET':
        return async_to_sync(_fetch_participant_ble_devices)(request)
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)

async def _fetch_participant_ble_devices(request):
    api_url = 'http://localhost:8000/ble=3'
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        participant_devices = response.json()
        print("participant_devices", len(participant_devices))
        if len(participant_devices) > 0:
            devices = await scan_devices()
            print("SCAN DEVICES", devices)
            if 'error' in devices:
                return JsonResponse({'participant_devices': participant_devices, 'scanned_devices': [],
                                     'error': 'Failed to scan BLE devices', 'details': devices['error']},
                                    status=500)
            matched_devices = []
            scanned_device_addresses = [{'uuid': device.address, 'name': device.name} for device in devices]

            for participant in participant_devices:
                device_address = participant.get('device_address')
                if device_address:
                    matching_devices = [device for device in devices if device.address == device_address]
                    if matching_devices:
                         matched_devices.extend(matching_devices)

            print("MATCHED", matched_devices)
            csrf_token = get_token(request)
            if len(matched_devices) > 0:
                connect_results = await connect_matched_devices(matched_devices, csrf_token)
                return JsonResponse({'participant_devices': participant_devices, 'scanned_devices': scanned_device_addresses,
                                     'connect_results': connect_results})
            else:
                return JsonResponse({'participant_devices': participant_devices, 'scanned_devices': [],
                                     'error': 'No participant devices found'}, status=404)
        else:
            return JsonResponse({'participant_devices': participant_devices, 'scanned_devices': [],
                                 'error': 'No participant devices found'}, status=404)
    except requests.RequestException as e:
        logger.error(f"Error fetching participant devices: {e}")
        return JsonResponse({'error': str(e)}, status=500)

@csrf_exempt
def connect_device(request):
    if request.method == 'POST':
        return async_to_sync(_connect_device)(request)
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)

async def _connect_device(request):
    data = json.loads(request.body)
    device_uuid = data.get('device_uuid')
    device_name = data.get('device_name')
    if device_uuid:
        try:
            async with BleakClient(device_uuid) as client:
                csrf_token = get_token(request)
                return JsonResponse(
                    {'success': True, 'message': 'Device connected successfully', 'csrf_token': csrf_token,
                     'device_name': device_name})
        except Exception as e:
            return JsonResponse({'success': False, 'device_name': device_name, 'error': str(e)})
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)



@csrf_exempt
def fetch_heart_rate(request):
    if request.method == 'POST':
        return async_to_sync(_fetch_heart_rate)(request)
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)

@csrf_exempt
async def _fetch_heart_rate(request):
    data = json.loads(request.body)
    device_uuid = data.get('device_uuid')
    device_name = data.get('device_name')
    if device_uuid:
        try:
            async with heart_rate_lock:  # Acquire lock before accessing Heart Rate service
                result = await get_heart_rate(device_uuid, device_name)
            return JsonResponse(result)
        except Exception as e:
            return JsonResponse({'success': False, 'error': str(e)})
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)


async def get_heart_rate(device_uuid, device_name):
    HEART_RATE_SERVICE_UUID = "0000180d-0000-1000-8000-00805f9b34fb"
    HEART_RATE_CHARACTERISTIC_UUID = "00002a37-0000-1000-8000-00805f9b34fb"
    try:
        async with BleakClient(device_uuid, timeout=10.0) as client:  # Set a timeout of 10 seconds
            services = await client.get_services()
            heart_rate_characteristic = None

            for service in services:
                if service.uuid.lower() == HEART_RATE_SERVICE_UUID:
                    for characteristic in service.characteristics:
                        if characteristic.uuid.lower() == HEART_RATE_CHARACTERISTIC_UUID:
                            heart_rate_characteristic = characteristic

            if heart_rate_characteristic:
                # Define a function to handle heart rate notifications
                async def notification_handler(sender, data):
                    # Process the received heart rate data
                    heart_rate_value = data[1]  # Extract the second byte
                    print(f"Heart rate: {device_name} {heart_rate_value}")

                    timestamp = datetime.datetime.now().isoformat()

                    heart_rate_data = {'heart_rate_value': heart_rate_value, 'timestamp': timestamp,
                                       'environment': "demo-room"}
                    # Add the timestamp to the dictionary
                    heart_rate_data_json = json.dumps(heart_rate_data)
                    mqtt_client.publish("testTopic", heart_rate_data_json)

                # Subscribe to notifications for the heart rate characteristic
                await client.start_notify(heart_rate_characteristic.uuid, notification_handler)
                logger.info(f"Heart rate notifications enabled for device: {device_name}")

                # Keep the coroutine running to receive notifications continuously
                while True:
                    await asyncio.sleep(1)  # Keep the coroutine alive

    except Exception as e:
        logger.error(f"Failed to fetch heart rate for device: {device_name}. Error: {e}")
        return {'success': False, 'error': str(e)}

    return {'success': True, 'message': 'Heart rate notifications enabled.'}


@csrf_exempt
def fetch_atmotube_data(request):
    if request.method == 'POST':
        return async_to_sync(_fetch_atmotube_data)(request)
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)

async def _fetch_atmotube_data(request):
    data = json.loads(request.body)
    device_uuid = data.get('device_uuid')
    device_name = data.get('device_name')
    if device_uuid:
        try:
            async with atmotube_lock:  # Acquire lock before accessing Atmotube service
                result = await get_atmotube_data(device_uuid, device_name)
            return JsonResponse(result)
        except Exception as e:
            return JsonResponse({'success': False, 'error': str(e)})
    return JsonResponse({'success': False, 'error': 'Invalid request'}, status=400)


async def get_atmotube_data(device_uuid,device_name):
        ATMOTUBE_CHARACTERISTIC_UUIDS = {
            "ATMOTUBE": "DB450003-8E9A-4818-ADD7-6ED94A328AB4",
            "PM": "DB450005-8E9A-4818-ADD7-6ED94A328AB4",
            "VOC": "DB450002-8E9A-4818-ADD7-6ED94A328AB4"
        }
        try:
            async with BleakClient(device_uuid, timeout=30.0) as client:
                services = await client.get_services()
                characteristics = {
                    "ATMOTUBE": None,
                    "PM": None,
                    "VOC": None
                }

                for service in services:
                    for characteristic in service.characteristics:
                        for key, uuid in ATMOTUBE_CHARACTERISTIC_UUIDS.items():
                            if characteristic.uuid.upper() == uuid:
                                characteristics[key] = characteristic

                if any(characteristics.values()):
                    if characteristics["ATMOTUBE"]:
                        await client.start_notify(characteristics["ATMOTUBE"].uuid, atmotube_notification_handler)

                    if characteristics["PM"]:
                        await client.start_notify(characteristics["PM"].uuid, atmotube_pm_notification_handler)

                    if characteristics["VOC"]:
                        await client.start_notify(characteristics["VOC"].uuid, atmotube_voc_notification_handler)

                    while True:
                        await asyncio.sleep(1)  # Keep the coroutine alive

        except Exception as e:
            logger.error(f"Failed to fetch Atmotube data for device: {device_name}. Error: {e}")
            return {'success': False, 'error': str(e)}

        logger.info(f"Atmotube notifications enabled for device: {device_name}")
        return {'success': True, 'message': 'Atmotube notifications enabled.'}


async def atmotube_notification_handler(sender, data):
    humidity, temp1, pressure, temp2 = struct.unpack('<BBiH', data)
    humidity_percentage = humidity
    temp1_celsius = temp1
    temp2_celsius = temp2 / 100.0

    logger.info(f"Humidity: {humidity_percentage}%, Temperature1: {temp1_celsius}°C")

    atmotube_data = {
        'humidity': humidity_percentage,
        'temperature1': temp1_celsius
    }
    timestamp = datetime.datetime.now().isoformat()

    # Add the timestamp to the dictionary
    atmotube_data['timestamp'] = timestamp
    atmotube_data['environment'] = "demo-room"
    atmotube_data_json = json.dumps(atmotube_data)
    mqtt_client.publish("testTopic", atmotube_data_json)


async def atmotube_pm_notification_handler(sender, data):
    pm1, pm2_5, pm10, pm4 = struct.unpack('<3s3s3s3s', data)
    pm1 = int.from_bytes(pm1, byteorder='little') / 100
    pm2_5 = int.from_bytes(pm2_5, byteorder='little') / 100
    pm10 = int.from_bytes(pm10, byteorder='little') / 100
    pm4 = int.from_bytes(pm4, byteorder='little') / 100

    print(f"PM1: {pm1} ug/m³, PM2.5: {pm2_5} ug/m³, PM10: {pm10} ug/m³, PM4: {pm4} ug/m³")

    atmotube_pm_data = {
        'pm1': pm1,
        'pm2_5': pm2_5,
        'pm10': pm10,
        'pm4': pm4
    }
    timestamp = datetime.datetime.now().isoformat()

    # Add the timestamp to the dictionary
    atmotube_pm_data['timestamp'] = timestamp
    atmotube_pm_data['environment'] = "demo-room"
    atmotube_pm_data_json = json.dumps(atmotube_pm_data)
    mqtt_client.publish("testTopic", atmotube_pm_data_json)


async def atmotube_voc_notification_handler(sender, data):
    sgpc3_ppb_tvoc, _ = struct.unpack('<HH', data)
    sgpc3_ppb_tvoc = sgpc3_ppb_tvoc / 1000.0  # Convert to ppb

    print(f"VOC: {sgpc3_ppb_tvoc} ppb")

    aqs = get_aqs(sgpc3_ppb_tvoc)
    print(f"Air Quality Score (AQS): {aqs}")

    atmotube_voc_data = {
        'sgpc3_ppb_tvoc': sgpc3_ppb_tvoc,
        'aqs': aqs
    }
    timestamp = datetime.datetime.now().isoformat()

    # Add the timestamp to the dictionary
    atmotube_voc_data['timestamp'] = timestamp
    atmotube_voc_data['environment'] = "demo-room"
    atmotube_voc_data_json = json.dumps(atmotube_voc_data)
    mqtt_client.publish("testTopic", atmotube_voc_data_json)


def get_aqs(voc):
    if voc < 0.5:
        return int(100 - 60 * voc)
    elif voc < 2:
        return int((118 - 26 * voc) / 1.5)
    else:
        aqs = int((374 - 44 * voc) / 6.5)
        return max(aqs, 0)

[dlech]

What is the error and the full stack trace of the error?

[wahadatjan]

sometimes i get Bluez operation in progress
I am able to connect to multiple devices but when it comes to get data from device characteristics . it only gives from one of the devices not from all the same time. Can you suggest solution to this
@dlech