How to receive response from direct_command_reply from EV3 using Python?

Question

I've established and tested a Bluetooth connection (by playing a tone) between EV3 and Mac running Yosemite with Python 2.7.11.

I used file write as in bt.write(tone) in Python to play the tone with the "tone" variable having the direct command. Now, I'm trying to read a .rtf text file that is on the brick using Python.

I'm using a direct command for this. I have created a byte stream for the direct command using opFile (with open_read, read_value and close) all packaged into one direct command (DIRECT_COMMAND_REPLY).

Now, since I'm using Python file write to write this direct command to the Bluetooth serial port, how do I get the value returned (the text value in the .rtf file) by the direct command?

Python file write does not have a return value. Please help on how I can get the reply from the direct command with Python.

Answer 1

Reading files is better done by system commands. Here is a class EV3, that sends direct commands and system commands. The program sends a direct command (does nothing) and uses the system commands BEGIN_UPLOAD and CONTINUE_UPLOAD to read file /bin/usb-devices from the EV3 device:

#!/usr/bin/env python3

import socket
import struct
import time

class EV3():
    def __init__(self, host: str):
        self._sock = socket.socket(
            socket.AF_BLUETOOTH,
            socket.SOCK_STREAM,
            socket.BTPROTO_RFCOMM
        )
        self._sock.connect((host, 1))

    def __del__(self):
        self._sock.close()

    def send_direct_cmd(self, ops: bytes, local_mem: int=0, global_mem: int=0) -> bytes:
        dir_cmd = b''.join([
            struct.pack('<h', len(ops) + 5),
            struct.pack('<h', 42),
            DIRECT_COMMAND_REPLY,
            struct.pack('<h', local_mem*1024 + global_mem),
            ops
        ])
        self._sock.send(dir_cmd)
        self.printHex('Sent', dir_cmd)
        reply = self._sock.recv(5 + global_mem)
        self.printHex('Recv', reply)
        return reply

    def send_system_cmd(self, cmd: bytes) -> bytes:
        sys_cmd = b''.join([
            struct.pack('<h', len(cmd) + 3),
            struct.pack('<h', 42),
            SYSTEM_COMMAND_REPLY,
            cmd
        ])
        self._sock.send(sys_cmd)
        self.printHex('Sent', sys_cmd)
        time.sleep(0.1)
        return self._sock.recv(1024)

    def printHex (self, desc: str, hex: bytes) -> None:
        """Print a hex-String in human readable form"""
        print(desc + ' 0x|' + ':'.join('{:02X}'.format(byte) for byte in hex) + '|')

DIRECT_COMMAND_REPLY = b'\x00'
SYSTEM_COMMAND_REPLY = b'\x01'
opNop                = b'\x01'
BEGIN_UPLOAD = b'\x94'
CONTINUE_UPLOAD = b'\x95'

# direct command
my_ev3 = EV3('00:16:53:42:2B:99')
ops = opNop
my_ev3.send_direct_cmd(ops)

# system commands
path = '/bin/usb-devices'
cmd = b''.join([
    BEGIN_UPLOAD,
    struct.pack('<H', 1012),      # SIZE
    str.encode(path) + b'\x00'    # NAME
])
reply = my_ev3.send_system_cmd(cmd)
(size, handle) = struct.unpack('<IB', reply[7:12])
part_size = min(1012, size)
fmt = str(part_size) + 's'
data = struct.unpack(fmt, reply[12:])[0]
rest = size - part_size
while rest > 0:
    part_size = min(1016, rest)
    cmd = b''.join([
        CONTINUE_UPLOAD,
        struct.pack('<BH', handle, part_size)
    ])
    reply = my_ev3.send_system_cmd(cmd)
    fmt = 'B' + str(part_size) + 's'
    (handle, part) = struct.unpack(fmt, reply[7:])
    data += part
    rest -= part_size
print(data.decode("utf8"))

You find additional details in Lesson 11. Or look in github. I hope, this helps!