Python線程和Arduino通信

我想創建一個線程，需要不斷更新(cX, cY, angle)在python代碼中的3個變量（浮點數）並將其每30ms發送給arduino uno。Python線程和Arduino通信

我從來沒有使用過線程，作爲初學者，我很難理解它。

有人給我的想法是創建三個變量的元組，並在主要的python流中創建一個線程，將該元組發送到arduino（通過串行的PC到arduino）。

任何人都可以幫助我，或者至少能告訴我如何開始。

2016-10-31 Amro A.

線程是一種同時運行代碼的方式。這意味着你可以運行功能A，但是在它完成之前，你可以啓動功能B並且它們繼續執行，不一定同時執行（即並行編程）。想一想，在特定的時間點，如果你停下來看看你的代碼中發生了什麼，你會看到函數A執行了一些語句，但還沒有完成，但是一些函數B的語句也被執行了。

在Python線程模塊中，每個線程都是一個帶有一些可能參數的函數（或方法），並且當它啓動時函數的主體運行。當函數返回時，線程完成運行，並且線程的狀態更改爲not alive（請參閱threading.Thread.is_alive()）。

所以我們通過傳遞函數/方法的參考和可能的參數創建一個線程：

from threading import Thread 

simple_thread = Thread(target=func_with_no_args) 
thread_with_args = Thread(target=my_function, args=(arg1, arg2))

然後我們啓動線程，運行函數體。

simple_thread.start() 
thread_with_args.start()

但是代碼的當前流程仍在繼續（當前線程，程序中總是有主線程開始執行它）。所以我們不需要等待功能（func_with_no_args或my_function）完成。

如果我們需要檢查，如果我們在一個線程中執行的功能已經完成，我們可以檢查它是否還活着：

simple_thread.is_alive() # return bool

如果我們要等到在線程函數執行完畢，我們可以join的線程。 join線程完成後也是很好的選擇。

simple_thread.join()

這是IO任務主要是有幫助的，因爲當程序等待數據準備/ IO從它可以做其他的東西。

我想創建一個線程，需要3個變量（浮動）是...

我假設你想實現的東西，你正在使用的線程作爲工具。我的意思是創建線程不是這裏的目標，而是實現你想要的目標的一種方式。因爲應用程序的細節從問題中缺失，所以我不能提供確切的代碼/示例，所以我將使用抽象/通用答案，希望能夠將這個想法應用到您的應用程序中。

我認爲應用程序正在從某處（因此IO）接收數據，並且每30ms應該將新數據發送給ardunio（它又像IO一樣）。所以應用程序有2個IO臂，一個接收數據，另一個發送給ardunio。所以使用線程可能是合理的。

我們可以有2個函數，1個讀取數據和1個更新ardunio。我們在2個線程中運行它們，thread_read（讀取數據）和thread_ardunio（更新ardunio）。

如果是這樣，我們需要一種在線程之間共享信息的方法。線程可以很容易地使用內存（在進程內存中，可通過變量訪問），所以我們可以使用兩個函數都可訪問的變量。當變量由1個線程更新時，另一個線程也會看到更新的結果。

storage = None 

def read_data(): 
    global storage 
    # read data from DB, web API, user input, etc. 
    # maybe in a loop to keep receiving data. Or generate the data from here 
    storage = (cX, cY, angle) 

def send_to_ardunio(): 
    global storage 
    # send storage data to ardunio, maybe in a loop 
    # sleeping 30ms after each update 


thread_read = Thread(target=read_data) 
thread_ardunio = Thread(target=send_to_ardunio) 
thread_read.start() 
thread_ardunio.start()

這是一條路。恕我直言，不那麼漂亮，因爲有一個全球變量掛在那裏。我們可以做些什麼來消除變量？我們可以使用隊列（請參閱queue.Queue）。

隊列是線程之間溝通的好方法，我想。因此，具有數據的線程將它們放到隊列（a.k.a生產者）上，而另一個線程從隊列中選取物品（a.k.a消費者）。

事情是這樣的：

def read_data(queue_): 
    # read data from DB, web API, user input, etc. 
    # maybe in a loop to keep receiving data, or generate the data from here 
    data = (cX, cY, angle) 
    queue_.put(data) 

def send_to_ardunio(queue_): 
    # send data to ardunio, maybe in a loop 
    # sleeping 30ms after each update 
    data = queue_.get() 
    cX, cY, angle = data 

queue_ = Queue() # this will be used to transfer data 
thread_read = Thread(target=read_data, args=(queue_,)) 
thread_ardunio = Thread(target=send_to_ardunio, args=(queue_,)) 
thread_read.start() 
thread_ardunio.start()

看起來更好。

現在我們需要等待函數運行。所以我們可以在線程上調用join方法。同樣這次我冒昧地假設我們可以控制讀取數據需要多少時間。如果我們應該用新數據每30ms更新一次ardunio，那麼製作者可以調整頻率，消費者可以毫不猶豫地消費。

此外，我們需要一種方法來告訴線程停止生產/消費。我們可以使用Event（參見threading.Event），或者爲了簡單起見，只需通過協商一個隊列中的數據就可以代表消費者應該停止。上述

def read_data(queue_): 
    while True: 
     # calculate/get cX, cY, angle 
     queue_.put((cX, cY, angle)) 
     # sleep 30ms 
     # when we finished producing data, put something to the queue to tell the consumer there is no more data. I'll assume None is good option here 
    queue_.put(None) 


def send_to_ardunio(queue_): 
    while True: 
     data = queue_.get() 
     if data is None: 
      break 
     cX, cY, angle = data 
     # update ardunio, because the data is updated every 30ms on the producer, we don't need to do anything special. We can just wait when the data is ready, we'll update. 

queue_ = Queue() 
thread_read = Thread(target=read_data, args=(queue_,)) 
thread_ardunio = Thread(target=send_to_ardunio, args=(queue_,)) 

thread_read.start() 
thread_ardunio.start() 

thread_read.join() 
thread_ardunio.join()

的代碼假定生產者（thread_read）會知道停止生產的數據。

如果不是這種情況，我們可以使用Event來觸發這兩個功能來停止生產和消費。

最後，我在加入線程時遇到了一些小問題。如果主線程正在加入其他線程，則會被阻塞，並且不會很好地響應SIGINT。因此，如果您嘗試停止Python進程（通過按Ctrl + C或發送SIGINT）它將不會退出。

但是，我們可以嘗試使用超時連接線程。所以每一次主線程都可以查看接收到的信號並處理它們。

def read_data(queue_, should_stop): 
    while not should_stop.is_set(): 
     # calculate/get cX, cY, angle 
     queue_.put((cX, cY, angle)) 
     # sleep 30ms 

def send_to_ardunio(queue_, should_stop): 
    while not should_stop.is_set(): 
     data = queue_.get() 
     cX, cY, angle = data 
     # update ardunio 

def tell_when_to_stop(should_stop): 
    # detect when to stop, and set the Event. For example, we'll wait for 10 seconds and then ask all to stop 
    time.sleep(10) 
    should_stop.set() 


queue_ = Queue() 
should_stop = Event() 

thread_stop_decider = Thread(target=tell_when_to_stop, args=(should_stop,)) 
thread_read = Thread(target=read_data, args=(queue_, should_stop)) 
thread_ardunio = Thread(target=send_to_ardunio, args=(queue_, should_stop)) 

thread_read.start() 
thread_ardunio.start() 
thread_stop_decider.start() 

try: 
    while thread_read.is_alive(): 
     thread_read.join(1) 
except KeyboardInterrupt: 
     should_stop.set() 
thread_read.join() 
thread_ardunio.join() 
thread_stop_decider.join()

來源

2016-10-31 14:42:00 farzad

Python線程和Arduino通信

回答

相關問題