How can I run a terminal command asynchronously using Python?

Question

I am working on a Linux environment, trying to automate these 2 Linux commands using a Python script:

sudo rfcomm connect 0 XX:XX:XX:XX:XX:XX
hcitool rssi XX:XX:XX:XX:XX:XX

Command 1 connects to my bluetooth device, and command 2 returns the received signal strength inidication (RSSI) as long as the device is connected. So the script must be able to connect and then extract RSSI without any human intervention. I am running these commands using the subprocess module.

The problem: When command 1 is run such that a connection is made, it says "Connected /dev/rfcomm0 to XX:XX:XX:XX:XX:XX on channel 1 Press CTRL-C for hangup", however it does not return. It only returns once the connection is lost, at which point it would be impossible to extract RSSI. Currently the python script does not continue to the next line of code after running command 1, so I am unable to run command 2 afterwards. Here is my code:

def get_rssi(): 
    while True:
        p1 = subprocess.run(['hcitool', 'rssi', shimmer_id], capture_output=True, text=True)
        
        if p1.stderr:
            print('stderr:', p1.stderr)
            reconnect()
        if p1.stdout:
            print('stdout:', p1.stdout)
            time.sleep(0.3)


def reconnect():
    reconnected = False
    print('[-] lost connection, reconnecting...')
    
    while reconnected is False:
        p2 = subprocess.run(['sudo', 'rfcomm', 'connect', '0', shimmer_id], capture_output=True, text=True)
        
# If the connection on the previous line is successful, nothing is executed from  here onward! The script only continues once the above line returns (i.e. when the connection is lost)

        if p2.stderr:
            if 'Host is down' in p2.stderr.decode():
                print('Host is down')
                time.sleep(0.3)
                reconnect()
        else:
            reconnected = True
            
    get_rssi()

I have tried threading and multiprocessing, but they do not help since command 1 never "joins" in python threading lingo. I am aware that I probably need to run command 1 asynchronously, but I am struggling to implement it properly in python even while referring to the asyncio docs. Any help is much appreciated!

You can certainly do this with threading. By using daemon mode, the thread is detached, and you don't have to join it. However, you can also use asyncio.create_subprocess_shell. — Tim Roberts
– Tim Roberts, Commented Apr 23, 2021 at 22:21
I wish to know when it returns, because I also need to know when the connection has failed (out of range). What method do you recommend? — Quentin James
– Quentin James, Commented Apr 23, 2021 at 22:30

ukBaz · Accepted Answer · 2021-04-24 08:06:00Z

1

It looks like you are trying to make an RFCOMM connection.

Both rfcomm and hcitool have been deprecated by the BlueZ project back in 2017 so I wouldn't recommend starting a project using these.

BlueZ has APIs that are designed to be accessed from languages like Python and these are documented at: https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/doc and the official examples are at: https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/test

However, RFCOMM (or Serial Port Profile) connections can be done with the socket library in the standard Python releases. There are examples at https://stackoverflow.com/a/63382835/7721752

There are also examples in the above answer of using the Bluedot library to do the same.

answered Apr 24, 2021 at 8:06

ukBaz

8,1842 gold badges11 silver badges42 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

How can I run a terminal command asynchronously using Python?

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related