I am looking for a solution for _thread in class object

[alien1983]

Hi,

This is an example of
threading

for a class...

Code: Select all

# A minimal threading example with an object
import threading
import time


class MyThread(threading.Thread):
    def run(self):                                        # Default called function with mythread.start()
        print("{} started!".format(self.getName()))        # "Thread-x started!"
        time.sleep(1)                                      # Pretend to work for a second
        print("{} finished!".format(self.getName()))      # "Thread-x finished!"

def main():
    for x in range(4):                                    # Four times...
        mythread = MyThread(name = "Thread-{}".format(x))  # ...Instantiate a thread and pass a unique ID to it
        mythread.start()                                  # ...Start the thread, run method will be invoked
        time.sleep(.9)                                    # ...Wait 0.9 seconds before starting another

if __name__ == '__main__':
    main()

Not work With Micropython, I am looking for a solution for _thread..

Code: Select all

import _thread
import time
...

class MyThread(_thread.Thread): ?

[alien1983]

I found this module in internet:



GitHuB micropython-lib/threading/



threading.py

Code: Select all

import _thread


class Thread:

    def __init__(self, group=None, target=None, name=None, args=(), kwargs=None):
        self.target = target
        self.args = args
        self.kwargs = {} if kwargs is None else kwargs

    def start(self):
        _thread.start_new_thread(self.run, ())

    def run(self):
        self.target(*self.args, **self.kwargs)

But not work with this example code...



Traceback (most recent call last):
File "main.py", line 237, in <module>
File "threading.py", line 12, in start
TypeError: function missing 1 required positional arguments

Code: Select all

[File "main.py", line 237, in <module>]

ERROR IN  >>>
mythread.start()                                  # ...Start the thread, run method will be invoked


[File "threading.py", line 12, in start]

>>>  
_thread.start_new_thread(self.run, ())

[jimmo]

This is an example of
threading

Yeah, MicroPython doesn't have a built-in implementation of "threading" (only _thread), but as you've discovered, micropython-lib provides a simple one.

But not work with this example code...

Can you share your main.py?

It should be something like

Code: Select all

def my_thread_func():
  ... do stuff

mythread = threading.Thread(target=my_thread_func)
mythread.start()

if you want to pass arguments to my_thread_func, you can do

Code: Select all

def my_thread_func(x, y):
  print(x, y)  # Will print "1, 2"
  ... do stuff

a, b = 1, 2
mythread = threading.Thread(target=my_thread_func, args=(a, b))
mythread.start()

[alien1983]

main.py

Code: Select all

# A minimal threading example with function calls
import threading
import time

def loop1_10():
    for i in range(1, 11):
        time.sleep(1)
        print(i)

threading.Thread(target=loop1_10).start()

# A minimal threading example with an object
import threading
import time


class MyThread(threading.Thread):
    def run(self):                                         # Default called function with mythread.start()
        print("{} started!".format(self.getName()))        # "Thread-x started!"
        time.sleep(1)                                      # Pretend to work for a second
        print("{} finished!".format(self.getName()))       # "Thread-x finished!"

def main():
    for x in range(4):                                     # Four times...
        mythread = MyThread(name = "Thread-{}".format(x))  # ...Instantiate a thread and pass a unique ID to it
        mythread.start()                                   # ...Start the thread, run method will be invoked
        time.sleep(.9)                                     # ...Wait 0.9 seconds before starting another

if __name__ == '__main__':
    main()

threading.py

Code: Select all

import _thread


class Thread:

    def __init__(self, group=None, target=None, name=None, args=(), kwargs=None):
        self.target = target
        self.args = args
        self.kwargs = {} if kwargs is None else kwargs

    def start(self):
        _thread.start_new_thread(self.run, ())

    def run(self):
        self.target(*self.args, **self.kwargs)

PROBLEM

Code: Select all

----------------
Filesystem size: 2341888 B
           Used: 5632 B
           Free: 2336256 B
----------------
Traceback (most recent call last):
  File "main.py", line 10, in <module>
  File "threading.py", line 12, in start
TypeError: function missing 1 required positional arguments
....

[jimmo]

This works for me... which board and firmware version are you using?

[alien1983]

Hi,
board: Heltec Wireless Stick.

Soft: MicroPython ESP32_LoBo_v3.2.24 - 2018-09-06 on ESP32 board with ESP32

[pythoncoder]

That firmware is very out of date. As far as I know the Loboris port was abandoned long ago.

[alien1983]

Hi,

I did a little revision to the threading class MP Lobo, and it finally works:)

Code: Select all

import _thread

class Thread:

    def __init__(self, group=None, target=None, name=None, args=(), kwargs=None):
        self.target = target
        self.args = args
        self.name = name
        self.kwargs = {} if kwargs is None else kwargs

    def start(self):
        _thread.start_new_thread(self.name, self.run,())

    def run(self):
        self.target(*self.args, **self.kwargs)

    def getName(self):
        return _thread.getSelfName()