I am trying in Python thanks to `subprocess` execute `adb shell` commands without re-initializing `adb shell` connection. How to figure out in Python when the command execution finished, as when `PS1` `FP4:/ $` appears again in:

```
~ $ adb shell
FP4:/ $ echo A && echo B
A
B
FP4:/ $
```

?

My current Python script is:

```python
import subprocess

adbShellProcess = subprocess.Popen('adb shell', shell = True, stdin = subprocess.PIPE, stdout = subprocess.PIPE)

def executeAdbCommand(command):
adbShellProcess.stdin.write(f'{command}\n'.encode())
adbShellProcess.stdin.flush()
output = adbShellProcess.stdout.readline().decode()
return output

print(f'{executeAdbCommand("echo A && echo B")=}')
```

Note that I don't want to assume the number of lines expected in each command output.
Aug 19, 11:12am

To detect when a command execution finishes in an adb shell session, you need to monitor the output for the reappearance of the shell prompt (FP4:/ $ in your case). This is because adb shell does not provide a built-in way to signal command completion programmatically.

Solution Approach

  1. Read Output Continuously: Read the output of the adb shell process in a loop until the shell prompt (FP4:/ $) reappears.
  2. Buffer Output: Accumulate the output lines until the prompt is detected.
  3. Return Buffered Output: Once the prompt is detected, return the accumulated output (excluding the prompt itself).

Updated Python Script

Here’s how you can modify your script to achieve this:

python
import subprocess
import threading
import queue

# Start the adb shell process
adb_shell_process = subprocess.Popen(
    'adb shell',
    shell=True,
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    bufsize=1,  # Line-buffered
)

# Queue to store output lines
output_queue = queue.Queue()

# Function to read output lines and put them in the queue
def read_output():
    for line in adb_shell_process.stdout:
        output_queue.put(line.decode().strip())

# Start the output reader thread
output_thread = threading.Thread(target=read_output, daemon=True)
output_thread.start()

def execute_adb_command(command):
    # Send the command
    adb_shell_process.stdin.write(f'{command}\n'.encode())
    adb_shell_process.stdin.flush()

    # Buffer to accumulate output
    output_buffer = []
    prompt = "FP4:/ $"

    # Read lines until the prompt reappears
    while True:
        line = output_queue.get()
        if line == prompt:
            break
        output_buffer.append(line)

    # Return the accumulated output
    return '\n'.join(output_buffer)

# Example usage
print(f"Output: {execute_adb_command('echo A && echo B')}")

Key Points

  1. Threading: A separate thread reads the output of the adb shell process and puts each line into a queue. This avoids blocking the main thread.
  2. Prompt Detection: The execute_adb_command function reads lines from the queue until it detects the shell prompt (FP4:/ $).
  3. Buffering: The output lines are accumulated in a buffer and returned once the prompt is detected.

Notes

  • Prompt Customization: Replace FP4:/ $ with the exact prompt of your adb shell if it differs.
  • Error Handling: Add error handling for cases where the prompt does not reappear (e.g., due to a crash or timeout).
  • Performance: This approach is line-buffered, so it works well for commands with line-based output. For binary or non-line-buffered output, a more complex solution may be needed.

Alternative: Use pexpect

If you can install third-party libraries, pexpect is a more robust solution for interacting with shell processes. It allows you to spawn a process and expect specific patterns (like the shell prompt) to detect command completion.

Would you like an example using pexpect?

Aug 19, 11:12am

This is a shared conversation and may include unverified content that doesn't reflect Mistral AI's views. Login to resume chatting.

Vibe can make mistakes. Check answers. Learn more