Threading with Python

Question

complete python newbie... I'm working with the Arduino pyfirmata package and Im trying to do something quite simple.

Depending on a user input to python, I want an LED to flash or not.

My problem is that the python program only asks for the user input once but I would like it to always ask for the input so the user can change function at any time.

I have tried using the threading package but no success... Perhaps there is a simpler way, but I am totally new to coding so I do not know of any other. Open to suggestions!!

Here is my code,

import pyfirmata
import threading
import time

board = pyfirmata.Arduino('/dev/cu.usbmodem14101')

def flash():
    for i in range(1000):
        board.digital[13].write(1)
        time.sleep(1)
        board.digital[13].write(0)
        time.sleep(1)

def stop():
    board.digital[13].write(0)


while True:
    runMode = input("Run or Stop? ")

    if runMode == "Run":
        x = threading.Thread(target=flash(), args=(1,))
        x.start()
        # x.join()

    elif runMode == "Stop":
        x = threading.Thread(target=stop(), args=(1,))
        x.start()
        #x.join()

Charlie · Answer 1 · 2020-02-22T21:04:16.603

If your looking to just kill the thread you could use mulitiprocessing which a multiprocessing.Process can p.terminate()

p = Process(target=flash, args=(,))
while True:
    runMode = input("Run or Stop? ")
    if runMode == "Run":
        p.start()
    elif runMode == "Stop":
        p.terminate()

However this is not recommended to just kill threads as it can cause errors if the process is handling critical resources or dependant on other threads see here for a better explanation Is there any way to kill a Thread?

A better option as described here is to use flags to handle your flashing, they allow a simple communication between threads

from threading import Event

e = event()

def check_for_stop(e):
    while not e.isSet():
         flash()
    print("Flashing Ended")

while True:
    runMode = input("Run or Stop? ")

    if runMode == "Run":
        x = threading.Thread(target=check_for_stop, args=(e,))
        x.start()
        # x.join()

    elif runMode == "Stop":
        e.set() #set flag true
        e.clear() #reset flag

here is the documentation for more info on event objects https://docs.python.org/2.0/lib/event-objects.html

I havent tested this code is just an example so apologies if it doesnt work straight away

Edit: Just looking at your function again you would want to check for the flag during the flashing thats my mistake aplogies so your flash function would look like

def flash():
    while e.isSet():
        board.digital[13].write(1)
        time.sleep(1)
        board.digital[13].write(0)
        time.sleep(1)

and you would pass this into the thread as you have before

x = threading.Thread(target=flash(), args=(1,))
x.start()

Thanks, it doesnt work straight away but I'll play around with it! — pythonosaurusrex, Feb 22 '20 at 22:09

score 0 · Answer 2 · answered Feb 22 '20 at 21:20

0

You got an error in the code.

You should create the thread via:

x = threading.Thread(target=flash)

Note: You gave the entered 'flash()' therefore executing the method in the main thread. And also your method doesn't have any arguments therefore you can remove the args values

answered Feb 22 '20 at 21:20

thanks, i just copied and pasted it from somewhere and tried to change what I thought i needed to! – pythonosaurusrex Feb 22 '20 at 22:08

martineau · Accepted Answer · 2020-02-22T21:34:32.627

You can do it in an object-oriented way by creating your own Thread subclass something like the Flasher class below.

One of the advantages to this approach is that it would be relatively easy to extend the Flasher class and make it control LEDs connected to different outputs, or to allow the delay between flashes to be specified at creation time. Doing the former would allow multiple instances to be running at the same time.

import pyfirmata
import threading
import time

OFF, ON = False, True

class Flasher(threading.Thread):
    DELAY = 1

    def __init__(self):
        super().__init__()
        self.daemon = True
        self.board = pyfirmata.Arduino('/dev/cu.usbmodem14101')
        self.flashing = False
        self.LED_state = OFF

    def turn_LED_on(self):
        self.board.digital[13].write(1)
        self.LED_state = ON

    def turn_LED_off(self):
        self.board.digital[13].write(0)
        self.LED_state = OFF

    def run(self):
        while True:
            if self.flashing:
                if self.LED_state == ON:
                    self.turn_LED_off()
                else:
                    self.turn_LED_on()
            time.sleep(self.DELAY)

    def start_flashing(self):
        if self.LED_state == OFF:
            self.turn_LED_on()
        self.flashing = True

    def stop_flashing(self):
        if self.LED_state == ON:
            self.turn_LED_off()
        self.flashing = False


flasher = Flasher()
flasher.start()

while True:
    runMode = input("Run or Stop? ").strip().lower()

    if runMode == "run":
        flasher.start_flashing()
    elif runMode == "stop":
        flasher.stop_flashing()
    else:
        print('Unknown response ignored')

Thanks! this works great, I just need to try and understand exactly whats happening now! — pythonosaurusrex, Feb 22 '20 at 22:10
That's good to hear because I couldn't test it properly myself (since I don't have the hardware). I realized the code might be a little bit too advanced for a newbie when I wrote it, so am sorry about that. If there's something in particular you can't figure-out, feel free to ask. — martineau, Feb 22 '20 at 22:20
Thanks for the offer... At the moment I am just working through, line by line, and researching all the parts I don't understand.. baby steps!! — pythonosaurusrex, Feb 23 '20 at 10:42

Threading with Python

3 Answers3