Unable to write to the device using bleak library through write_gatt

[SrishtiKaranth]

• bleak version: bleak==0.22.2
• Python version: Python 3.12.3
• Operating System: MacOS Solana

Description

I want to set the time on my BLE device to the current time in UTC using the write_gatt_char. However, I am facing this issue wherein the BLE notifies that it's successful but when I fetch the time back again from the device, it is still an old wrong time, sometime in 1970.

What I Did

Currently I have this code wherein I communicate with the device and try to set the time as well as fetch the time. But it's not working as intended.

import asyncio
from datetime import datetime, timezone
from bleak import BleakClient, BleakError
device_address = ""
characteristic_uuid = ""
CMD_TIME_SET = 0x06
CMD_TIME_GET = 0x07

response_data = None

def number_to_byte_array(number):
    return number.to_bytes(4, byteorder='little')

async def cmd_time_set(client, current_time_seconds):
    global response_data
    response_data = None

    
    cmd_bytes = bytearray([CMD_TIME_SET]) + number_to_byte_array(current_time_seconds)
    cmd_bytes = bytearray([CMD_TIME_SET]) + number_to_byte_array(current_time_seconds)
    print("Command to send (cmd_time_set):", bytes(cmd_bytes))
    print(f"Setting time: {current_time_seconds} ({datetime.fromtimestamp(current_time_seconds, tz=timezone.utc)})")

    try:
        await client.write_gatt_char(characteristic_uuid, bytes(cmd_bytes))
        print("\033[94mTime set command sent successfully.\033[0m")

        
        for _ in range(10):
            if response_data:
                break
            await asyncio.sleep(0.1)

        if response_data:
            print("Response data:", response_data)
            status = response_data[1]  # Assuming status is at index 1 based on your Kotlin logic
            return status
        else:
            print("No response received within timeout period")
            return None

    except Exception as e:
        print(f"Error during time set command: {repr(e)}")
        return None

async def cmd_time_get(client):
    global response_data
    response_data = None

    
    cmd_bytes = bytearray([CMD_TIME_GET])
    print("Command to send (cmd_time_get):", bytes(cmd_bytes))

    try:
        await client.write_gatt_char(characteristic_uuid, bytes(cmd_bytes))
        print("\033[94mTime get command sent successfully.\033[0m")

        
        for _ in range(10):
            if response_data:
                break
            await asyncio.sleep(0.1)

        if response_data:
            print("Response data:", response_data)
            status = response_data[1]  
            time_bytes = response_data[2:6]  
            print("Extracted time bytes:", time_bytes)
            current_time = int.from_bytes(time_bytes, byteorder='little')  
            return status, current_time
        else:
            print("No response received within timeout period")
            return None, None

    except Exception as e:
        print(f"Error during time get command: {repr(e)}")
        return None, None

async def main():
    async with BleakClient(device_address) as client:
        try:
            await client.start_notify(characteristic_uuid, notification_handler)

            
            current_time_seconds = int(datetime.now(timezone.utc).timestamp())
            print("Current time in seconds (UTC):", current_time_seconds)

            
            status = await cmd_time_set(client, current_time_seconds)
            if status is not None:
                print(f"Time set successful. Status: {status}")
            else:
                print("Time set failed.")

            
            status, current_time = await cmd_time_get(client)
            if status is not None and current_time is not None:
               
                print(f"Time get successful. Status: {status}")

             
                current_time_utc = datetime.fromtimestamp(current_time, tz=timezone.utc)
                print(f"Current time (UTC): {current_time_utc.strftime('%Y-%m-%d %H:%M:%S')}")

            else:
                print("Time get failed.")

        except BleakError as e:
            print(f"Failed to connect to the device: {e}")

        finally:
            await client.stop_notify(characteristic_uuid)


def notification_handler(sender, data):
    global response_data
    print(f"Notification from {sender}: {data}")  # Print the raw data
    response_data = data

asyncio.run(main())

Logs

Current time in seconds (UTC): 1720190642
Command to send (cmd_time_set): b'\x06\xb2\x06\x88f'
Setting time: 1720190642 (2024-07-05 14:44:02+00:00)
Time set command sent successfully.
Notification from  characteristic_uuid (Handle: 15): Unknown: bytearray(b'\x06\x00')
Response data: bytearray(b'\x06\x00')
Time set successful. Status: 0
Command to send (cmd_time_get): b'\x07'
Time get command sent successfully.
Notification from characteristic_uuid (Handle: 15): Unknown: bytearray(b'\x07\x00\xa8\x8c\x0e\x00')
Response data: bytearray(b'\x07\x00\xa8\x8c\x0e\x00')
Extracted time bytes: bytearray(b'\xa8\x8c\x0e\x00')
Time get successful. Status: 0
Current time (UTC): 1970-01-12 00:51:52

[dlech]

Other than explicitly specifying response=True in write_gatt_char(), I'm not sure what else to suggest.

Without any information on the actual device, it is impossible for us to know how it is supposed to work.