【試用報告】RP2040上的MicroPython環(huán)境中多線程編程

# file: multicore_test01.py
import machine
import _thread
import utime


led = machine.Pin(25, machine.Pin.OUT)
led.off()


key = 0
start_time = 0
def run_on_core1():
  global start_time
  while start_time == 0:
    pass


  while True:
    utime.sleep_ms(300)
    print((utime.ticks_us()-start_time)//100000, "led on")
    led.on()
    utime.sleep_ms(700)
    
def run_on_core0():
  global start_time
  start_time = utime.ticks_us()
  while True:
    utime.sleep_ms(700)
    print((utime.ticks_us()-start_time)//100000, "led off")
    led.off()
    utime.sleep_ms(300)


_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()

# file: multicore_test02.py


import machine
import _thread
import utime


def thread_1():
  while True:
    print("thread_1")
    utime.sleep_ms(1000)


def thread_2():
  while True:
    print("thread_2")
    utime.sleep_ms(1000)


_thread.start_new_thread(thread_1, ( ))
_thread.start_new_thread(thread_2, ( ))


while True:
  print("main")
  utime.sleep_ms(1000)

# file: multicore_test03.py
import machine
import _thread
import utime


led = machine.Pin(25, machine.Pin.OUT)
led.off()


status = 0
def run_on_core1():
  global status
  while True:
    if status:
      led.on()
    else:
      led.off()
    utime.sleep_ms(100)
    
def run_on_core0():
  global status
  while True:
    status = 1 if not status else 0
    utime.sleep_ms(1000)


_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()

# file: multicore_test04.py
import machine
import _thread
import utime


led = machine.Pin(25, machine.Pin.OUT)
led.off()


status = 0
data = []
def run_on_core1():
  global status, data
  while True:
    if status:
      led.on()
    else:
      led.off()
    str_data = ''.join(data)
    print("str_data: len=%d content=%s" % (len(str_data), str_data))
    utime.sleep_ms(1000)
    
def run_on_core0():
  global status, data
  while True:
    status = 1 if not status else 0
    data = []
    for i in range(100):
      data.append(str(status))
      utime.sleep_ms(10)
    utime.sleep_ms(1000)


_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()

# file: multicore_test05.py
import machine
import _thread
import utime


led = machine.Pin(25, machine.Pin.OUT)
led.off()


status = 0
data = []
def run_on_core1():
  global status, data
  while True:
    if status:
      led.on()
    else:
      led.off()
    lock.acquire()
    str_data = ''.join(data)
    print("str_data: len=%d content=%s" % (len(str_data), str_data))
    lock.release()
    utime.sleep_ms(1000)
    
def run_on_core0():
  global status, data
  while True:
    status = 1 if not status else 0
    lock.acquire()
    data = []
    for i in range(100):
      data.append(str(status))
      utime.sleep_ms(10)
    lock.release()
    utime.sleep_ms(1000)


lock = _thread.allocate_lock()
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()

# file: multicore_test06.py
import machine
import _thread
import utime


def run_on_core1():
  print("[core1] run thread")
  utime.sleep_ms(100)


def run_on_core0():
  while True:
    print("[core0] start thread:")
    _thread.start_new_thread(run_on_core1, ( ))
    utime.sleep_ms(1000)
  
run_on_core0()

# file: multicore_test07.py
import machine
import _thread
import utime


def run_on_core1():
  print("[core1] run thread")
  utime.sleep_ms(1000)


def run_on_core0():
  while True:
    print("[core0] start thread:")
    _thread.start_new_thread(run_on_core1, ( ))
    utime.sleep_ms(100)
  
run_on_core0()

# file: multicore_test08.py
import machine
import _thread
import utime


def run_on_core1():
  lock.acquire()
  print("[core1] run thread")
  utime.sleep_ms(1000)
  lock.release()


def run_on_core0():
  while True:
    print("[core0] start thread:")
    lock.acquire()
    _thread.start_new_thread(run_on_core1, ( ))
    lock.release()
    utime.sleep_ms(100)


lock = _thread.allocate_lock()
run_on_core0()

# file: multicore_test09.py
import machine
import _thread
import utime
from ws2812 import WS2812


led = machine.Pin(25, machine.Pin.OUT)
led.off()


BLACK = (0, 0, 0)
RED = (255, 0, 0)
YELLOW = (255, 150, 0)
GREEN = (0, 255, 0)
CYAN = (0, 255, 255)
BLUE = (0, 0, 255)
PURPLE = (180, 0, 255)
WHITE = (255, 255, 255)
COLORS = (BLACK, RED, YELLOW, GREEN, CYAN, BLUE, PURPLE, WHITE)


ws = WS2812(3, 1) #WS2812(pin_num,led_count)
ws.pixels_fill(BLACK)
ws.pixels_show()


def run_on_core1():
  while True:
    for color in COLORS:
      ws.pixels_fill(color)
      ws.pixels_show()
      utime.sleep_ms(200)
    
def run_on_core0():
  duty = 0
  step = 1
  count = 0
  while True:
    led.on()
    utime.sleep_ms(duty)
    led.off()
    utime.sleep_ms(10-duty)
    
    count = count + 1
    if count>10:
      count = 0
      duty = duty + step
      if duty >= 10:
        step = -1
      if duty <= 0 :
        step = 1


_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()

# file: multicore_test10.py
from machine import SPI,Pin
from ssd1306 import SSD1306_SPI
import framebuf
import _thread
import utime


spi = SPI(1, 100000, mosi=Pin(11), sck=Pin(10))
display = SSD1306_SPI(128, 64, spi, Pin(9),Pin(8), Pin(1))


def run_on_core1():
  global fbuf
  while True:
    if not fbuf == None:
      display.fill(0)
      
      lock.acquire()
      display.blit(fbuf,19,0)
      fbuf = None      
      lock.release()
      
      display.show()


    utime.sleep_ms(100)


def run_on_core0():
  global fbuf
  while True:
    for i in range(1,139):
      dirt = 'BAD_APPLE/' + str(i) + '.pbm'
      print(i, dirt)
      with open(dirt,'rb') as f :
        f.readline()
        f.readline()
        data = bytearray(f.read())
        
        lock.acquire()
        fbuf = framebuf.FrameBuffer(data,88,64,framebuf.MONO_HLSB)
        lock.release()
        
        utime.sleep_ms(100)


fbuf = None
lock = _thread.allocate_lock()
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()