2024-06-05

Python subprocess.Popen 用法與範例

本篇 ShengYu 介紹 Python subprocess.Popen 用法與範例，在進行 Python 程式開發時，常常需要與外部程式進行互動，例如執行系統指令、啟動外部應用程式或執行其他腳本。在這些情境下，Python 的 subprocess 模組是非常有用的工具，其中的 subprocess.Popen 提供了強大的功能和靈活性，讓我們能夠更精細地控制子行程。本文將詳盡介紹 subprocess.Popen 的用途，並提供簡單與進階範例，幫助你掌握這一功能。

什麼是 subprocess.Popen？

subprocess.Popen 是 Python 的 subprocess 模組中的一個類，用於啟動和管理子行程。相比於 subprocess.run 和 subprocess.call 等函式，Popen 提供了更大的靈活性，允許我們：

執行外部指令或腳本。
在執行指令時進行進階的 I/O 操作，如管道通訊。
取得子行程的回傳碼。
在非同步或同步模式下執行子行程。

為什麼使用 subprocess.Popen？

使用 subprocess.Popen 有以下幾個主要優點：

高靈活性：允許我們對子行程的標準輸入、輸出和錯誤流進行精細控制。
非阻塞操作：可以非同步執行子行程，避免阻塞主程式的執行。
管道通訊：支持將多個子行程的輸出和輸入進行管道連結，實現複雜的資料處理流程。

subprocess.Popen 基本範例

以下是一個基本範例，示範如何使用 subprocess.Popen 執行一個簡單的指令並取得其輸出。

import subprocess

# 建立一個子行程，執行 `ls` 指令（列出當前目錄中的檔案和資料夾）
process = subprocess.Popen(['ls'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

# 讀取標準輸出和標準錯誤
stdout, stderr = process.communicate()

# 將標準輸出和標準錯誤的結果解碼為字串並打印
print("標準輸出:")
print(stdout.decode())

if stderr:
    print("標準錯誤:")
    print(stderr.decode())

# 取得子行程的回傳碼
return_code = process.returncode
print(f"回傳碼: {return_code}")

在這個範例中，我們使用 subprocess.Popen 執行了 ls 指令，並透過 communicate() 方法讀取了標準輸出和標準錯誤。最終，我們還取得了子行程的回傳碼。

subprocess.Popen 進階範例

下面是一個進階範例，示範如何使用 subprocess.Popen 進行管道通訊。這個範例中，我們首先使用 seq 指令產生一個數字清單，然後使用 grep 指令過濾掉奇數。

import subprocess

# 執行第一個子行程，產生一個數字清單
process1 = subprocess.Popen(['seq', '1', '10'], stdout=subprocess.PIPE)

# 執行第二個子行程，過濾掉奇數
process2 = subprocess.Popen(['grep', '-v', '1$'], stdin=process1.stdout, stdout=subprocess.PIPE)

# 關閉 process1 的 stdout 以避免死鎖
process1.stdout.close()

# 讀取第二個子行程的標準輸出
output, _ = process2.communicate()

# 將結果解碼為字串並打印
print("過濾後的數字:")
print(output.decode())

在這個範例中，兩個子行程透過管道進行通訊。process1 產生了一個數字清單，並將輸出透過管道傳遞給 process2，process2 過濾掉奇數並輸出結果。

subprocess.Popen 常用參數

在使用 subprocess.Popen 時，有一些關鍵參數需要了解：

args: 要執行的指令和參數，通常以清單形式提供。
stdin, stdout, stderr: 用於設定子行程的標準輸入、輸出和錯誤流，可以是 subprocess.PIPE、檔案物件或 None。
shell: 如果為 True，指令會透過 shell 執行，允許使用 shell 特性（如管道和重導向）。
cwd: 設定子行程的當前工作目錄。
env: 設定子行程的環境變數。

subprocess.Popen 是 blocking 阻塞函式嗎？

subprocess.Popen 本身不是阻塞（blocking）的。當你呼叫 subprocess.Popen 時，它會啟動一個子行程並立即回傳一個 Popen 物件，而不會等待子行程完成。這意味著主行程可以繼續執行其他操作，而子行程在後台執行。

但是，與 Popen 物件的某些方法互動時，可能會產生阻塞行為，例如：

1. Popen.communicate()

這個方法會阻塞主行程，直到子行程完成並且其所有輸出和錯誤資料都被讀取完畢。範如如下，

import subprocess

process = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = process.communicate()  # 這裡會阻塞，直到子行程完成
print(stdout.decode())

2. Popen.wait()

這個方法會阻塞主行程，直到子行程終止。範如如下，

import subprocess

process = subprocess.Popen(['sleep', '5'])
process.wait()  # 這裡會阻塞，直到子行程完成
print("子行程已結束")

3. 讀取標準輸入/輸出/錯誤

如果你使用 Popen.stdout.read() 或 Popen.stderr.read() 等方法來讀取子行程的輸出，這些操作也可能會阻塞，特別是當子行程產生大量輸出且你沒有適當地處理這些輸出的時候。範如如下，

import subprocess

process = subprocess.Popen(['cat'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
process.stdin.write(b'Hello\n')
output = process.stdout.read()  # 這裡會阻塞，直到有資料可以讀取
print(output.decode())

整理一下，subprocess.Popen 本身不阻塞，啟動子行程後立即回傳。

但是某些方法如 communicate()、wait() 和讀取輸出的方法會阻塞，直到特定條件滿足（例如子行程結束或輸出被完全讀取）。

subprocess.Popen 執行 adb logcat

以下是示範用 subprocess.Popen 執行 adb logcat 並將其輸出 print 出來

程式碼如下所示：

import subprocess

def run_adb_logcat():
    command = ['adb', 'logcat']
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    
    try:
        while True:
            output = process.stdout.readline()
            if output == b'' and process.poll() is not None:
                break
            if output:
                print(output.decode('utf-8').strip())
    finally:
        process.kill()
        process.wait()

if __name__ == "__main__":
    run_adb_logcat()

這個程式碼會使用 Python 的 subprocess 模組來呼叫 adb 指令並捕獲其輸出。當你執行這個程式時，它會開始顯示 adb logcat 的輸出，直到你手動停止它（例如按下 Ctrl+C）或程式本身結束。

subprocess.Popen 執行 adb logcat 並將輸出寫到 log 裡

如果你想將 adb logcat 的輸出寫入到日誌檔案中，以下是程式碼範例：

import subprocess

def run_adb_logcat(output_file):
    command = ['adb', 'logcat']
    with open(output_file, 'w') as log_file:
        process = subprocess.Popen(command, stdout=log_file, stderr=subprocess.PIPE)
        
        try:
            process.wait()
        except KeyboardInterrupt:
            process.kill()
            process.wait()

if __name__ == "__main__":
    log_file = 'adb_logcat_output.txt'
    run_adb_logcat(log_file)
    print(f"adb logcat 輸出已寫入到 {log_file}")

將 subprocess.Popen 的 stdout 參數設定為指向一個打開的日誌檔案 (log_file)，這樣 adb logcat 的輸出會直接寫入這個檔案中。
程式會等待 adb logcat 指令執行完畢，並將輸出寫入到指定的日誌檔案中。
在主程式的結尾，印出日誌檔案的位置，以提示使用者輸出已完成。

這樣做可以讓你方便地將 adb logcat 的輸出保存到指定的日誌檔案中，以便後續查閱或分析。

如果你希望主程式在等待 adb logcat 的同時還能夠執行其他操作或處理其他任務，你可以將 adb logcat 的執行放在單獨的執行緒中，或者使用非阻塞的方式來捕獲 adb logcat 的輸出。

import subprocess
import threading

def run_adb_logcat(output_file):
    command = ['adb', 'logcat']
    with open(output_file, 'w') as log_file:
        process = subprocess.Popen(command, stdout=log_file, stderr=subprocess.PIPE)
        
        # 定義一個函式來非阻塞地捕獲輸出
        def capture_output():
            try:
                for line in iter(process.stdout.readline, b''):
                    log_file.write(line.decode('utf-8'))
            except Exception as e:
                print(f"錯誤發生：{e}")
        
        # 啟動一個執行緒來捕獲輸出
        output_thread = threading.Thread(target=capture_output)
        output_thread.start()
        
        try:
            process.wait()  # 等待 adb logcat 指令執行完畢
        except KeyboardInterrupt:
            process.kill()
            process.wait()
        finally:
            output_thread.join()  # 等待捕獲輸出的執行緒完成

if __name__ == "__main__":
    log_file = 'adb_logcat_output.txt'
    run_adb_logcat(log_file)
    print(f"adb logcat 輸出已寫入到 {log_file}")

總結

subprocess.Popen 是一個功能強大且靈活的工具，能夠滿足從簡單指令執行到複雜行程間通訊的各種需求。無論你是需要簡單地執行一個外部指令，還是需要在多個子行程間進行資料傳遞，subprocess.Popen 都能提供合適的解決方案。

透過上述範例和說明，相信你已經對 subprocess.Popen 有了一定的了解和掌握。希望這些資訊能夠幫助你在未來的開發工作中更好地使用 Python 進行行程管理。

以上就是 Python subprocess.Popen 用法與範例的介紹，
如果你覺得我的文章寫得不錯、對你有幫助的話記得 Facebook 按讚支持一下！