Sending tuples from hub to pc.

[techcatgato]

Hi, I am currently working on a project involving pc hub communication with bleak, and i wonder if theres a way to send a tuple with floats from the hub to the pc using stdout.write() (no buffer). I have tried multiple solutions and i'm yet to find one.

[dlech]

my_tuple = (1.0, 2.0, 3.0)
print(my_tuple)
# or
stdout.write(repr(my_tuple))

[techcatgato]

I am working without a buffer, so far stdout.write() has been fine except for not being able to send tuples\list which would be beneficial to the program

[laurensvalk]

Sending a tuple or list is not really a thing. You'd have to encode it as either a string or bytes. And convert back to a tuple/list on the receiving end. If you want to send floating point values, packing as bytes is a good idea. You could do:

# Pack data before sending
data = ustruct.pack("<ff", *(3.14, 6.28))

# Unpack received data
struct.unpack('<ff', data)

For example, this will send b'\xc3\xf5H@\xc3\xf5\xc8@' over the air and gets unpacked into (3.140000104904175, 6.28000020980835)

[techcatgato]

Sending a tuple or list is not really a thing. You'd have to encode it as either a string or bytes. And convert back to a tuple/list on the receiving end. If you want to send floating point values, packing as bytes is a good idea. You could do:

# Pack data before sending
data = ustruct.pack("<ff", *(3.14, 6.28))

# Unpack received data
struct.unpack('<ff', data)

For example, this will send b'\xc3\xf5H@\xc3\xf5\xc8@' over the air and gets unpacked into (3.140000104904175, 6.28000020980835)

Thanks, but I still have 1 problem that prevents this from working. For an unkown reason, whenever the hub connects to the PC, before even turning on the program, the hub transmits bytearray(b'\x00\x00\x02\x00\x00\x00'). This seems to be regardless of the programs on hub/PC, and is messing with my ability to unpack the right data.

[DylDevs]

my_tuple = (1.0, 2.0, 3.0)
print(my_tuple)
# or
stdout.write(repr(my_tuple))

Whenever something is printed it's sent to the PC (assuming it's connected)?

[dlech]

Yes, that is how it works.

[techcatgato]

Hi, facing a different problem right now. every time the hub connects to the pc it automatically communicates bytearray(b'\x00@\x02\x00\x00\x00')
or
bytearray(b'\x00\x00\x02\x00\x00\x00') regardless of the program on the hub. This kindof prevents me from getting the correct message

[techcatgato]

So if the first byte is an indicator, when i read the data i should ignore it? like if i want to translate it back to a string i need to do data[1:]?

[laurensvalk]

Why not give it a try? 🙂 Try sending bABCD to stdout on the hub and see what comes in at your PC, and which byte it starts with.

[techcatgato]

Why not give it a try? 🙂 Try sending bABCD to stdout on the hub and see what comes in at your PC, and which byte it starts with.

Running this:

    def handle_rx(_, data: bytearray):
        while True:
            if data[0] == 0x01:
                #print(data)
                #print(data[1:])
                #print(bytearray_to_floats(data[1:]))
                print(data[1:].decode("utf-8"))
            elif data[0] == 0x00:
                print("0x0 detected")

results in an endless spam of 0x0 detected

[laurensvalk]

What is the while True doing there?

[techcatgato]

What is the while True doing there?

Oh my god removing that fixed it, Thank you! Now that i have this working, I'll see if I can make tuples possible, and if yes i'll update the discussion for anyone who might try.

[techcatgato]

Okay, i got tuple communication completely working, note this is not using the standart buffer, so it is just going to give you the latest message sent.
Spike/Robot inventor program:

import ustruct
from ustruct import pack
from usys import stdout
def floats_to_bytearray(float_list):
    # Create an empty bytearray
    byte_array = bytearray()

    for number in float_list:
        # Pack the float into bytes (using 'f' format for float)
        byte_array.extend(ustruct.pack('f', number))

    return byte_array
def bytearray_to_floats(byte_array):
    # Calculate the number of floats
    num_floats = len(byte_array) // 4
    float_list = []

    for i in range(num_floats):
        # Unpack 4 bytes at a time as a float
        float_value = ustruct.unpack('f', byte_array[i*4:(i+1)*4])[0]
        float_list.append(float_value)

    return float_list
tuple_to_convert = (1, 2, 3)
tuple_converted_to_bytearray = floats_to_bytearray(tuple_to_convert)
while True:
    stdout.write(tuple_converted_to_bytearray)

PC side

import struct
def bytearray_to_floats(byte_array):
    # Calculate the number of floats
    num_floats = len(byte_array) // 4
    float_list = []

    for i in range(num_floats):
        # Unpack 4 bytes at a time as a float
        float_value = round(struct.unpack('f', byte_array[i*4:(i+1)*4])[0], ndigits=1)
        float_list.append(float_value)

    return tuple(float_list)
def handle_rx(_, data: bytearray):
    if data[0] == 0x01:
        print(bytearray_to_floats(data[1:]))

[laurensvalk]

Nice!

[techcatgato]

Been expirementing and I once again encounter a problem where I just get 1 0x0 reading and then hub disconnects.

# SPDX-License-Identifier: MIT
# Copyright (c) 2020 Henrik Blidh
# Copyright (c) 2022-2023 The Pybricks Authors

"""
Example program for computer-to-hub communication.

Requires Pybricks firmware >= 3.3.0.
"""

import asyncio
from contextlib import suppress
from bleak import BleakScanner, BleakClient
import struct

PYBRICKS_COMMAND_EVENT_CHAR_UUID = "c5f50002-8280-46da-89f4-6d8051e4aeef"


# Replace this with the name of your hub if you changed
# it when installing the Pybricks firmware.
HUB_NAME = "TC"

def bytearray_to_floats(byte_array):
    # Calculate the number of floats
    num_floats = len(byte_array) // 4

    return tuple((round(struct.unpack('f', byte_array[i*4:(i+1)*4])[0], ndigits=1)) for i in range(num_floats))


def decodepath(byte_array):
    floats = []
    start = 0

    while start < len(byte_array):
        # Check if there's enough data left for a float
        if start + 4 > len(byte_array):
            break

        float_value = round(struct.unpack('f', byte_array[start:start + 4])[0], ndigits=1)
        floats.append(float_value)
        start += 4

    # Grouping the floats into tuples, assuming each pair of floats represents a nested tuple
    return tuple(tuple(floats[i:i + 2]) for i in range(0, len(floats), 2))




async def main():
    main_task = asyncio.current_task()

    def handle_disconnect(_):
        print("Hub was disconnected.")

        # If the hub disconnects before this program is done,
        # cancel this program so it doesn't get stuck waiting
        # forever.
        if not main_task.done():
            main_task.cancel()

    ready_event = asyncio.Event()

    def handle_rx(_, data: bytearray):
        if data[0] == 0x01:
            #decoded_data = bytearray_to_floats(data[1:])
            #print(bytearray_to_floats(data[1:]))
            decodedpath = data(data[1:])
            print(decodedpath)
        else:
            print(data)

    # Do a Bluetooth scan to find the hub.
    device = await BleakScanner.find_device_by_name(HUB_NAME)
    print("Device connected: ", device)

    if device is None:
        print(f"could not find hub with name: {HUB_NAME}")
        return

    # Connect to the hub.
    async with BleakClient(device, handle_disconnect) as client:

        # Subscribe to notifications from the hub.
        await client.start_notify(PYBRICKS_COMMAND_EVENT_CHAR_UUID, handle_rx)

        async def send(data):
            # Wait for hub to say that it is ready to receive data.
            await ready_event.wait()
            # Prepare for the next ready event.
            ready_event.clear()
            # Send the data to the hub.
            await client.write_gatt_char(
                PYBRICKS_COMMAND_EVENT_CHAR_UUID,
                b"\x06" + b"ok",  # prepend "write stdin" command (0x06)
                response=True
            )




# Run the main async program.
if __name__ == "__main__":
    with suppress(asyncio.CancelledError):
        asyncio.run(main())

To my knowledge, only thing i changed is the data sent, So I dont know why now its only 0x0.

[dlech]

There isn't anything in your program to keep it from ending immediately after connecting.

Generally, there would be an infinite loop that handles the sending and receiving that keeps the program running until the hub disconnects.

[techcatgato]

There isn't anything in your program to keep it from ending immediately after connecting.

Generally, there would be an infinite loop that handles the sending and receiving that keeps the program running until the hub disconnects.

How can I add something like this?