Switch to async

This commit is contained in:
jimmy 2023-12-23 09:03:16 +00:00
parent 5821d81cf4
commit f8515bddde
1 changed files with 110 additions and 96 deletions

View File

@ -1,96 +1,110 @@
# Example using PIO to drive a set of WS2812 LEDs. # Example using PIO to drive a set of WS2812 LEDs.
import array, time import array, time
from machine import Pin from machine import Pin
import rp2 import rp2
from random import randint from time import sleep
from time import sleep import uasyncio as asyncio
# Configure the number of WS2812 LEDs. # Configure the number of WS2812 LEDs.
NUM_LEDS = 256 NUM_LEDS = 256
PIN_NUM = 0 PIN_NUM = 0
brightness = 0.2 brightness = 0.2
@rp2.asm_pio(sideset_init=rp2.PIO.OUT_LOW, out_shiftdir=rp2.PIO.SHIFT_LEFT, autopull=True, pull_thresh=24) @rp2.asm_pio(sideset_init=rp2.PIO.OUT_LOW, out_shiftdir=rp2.PIO.SHIFT_LEFT, autopull=True, pull_thresh=24)
def ws2812(): def ws2812():
T1 = 2 T1 = 2
T2 = 5 T2 = 5
T3 = 3 T3 = 3
wrap_target() wrap_target()
label("bitloop") label("bitloop")
out(x, 1) .side(0) [T3 - 1] out(x, 1) .side(0) [T3 - 1]
jmp(not_x, "do_zero") .side(1) [T1 - 1] jmp(not_x, "do_zero") .side(1) [T1 - 1]
jmp("bitloop") .side(1) [T2 - 1] jmp("bitloop") .side(1) [T2 - 1]
label("do_zero") label("do_zero")
nop() .side(0) [T2 - 1] nop() .side(0) [T2 - 1]
wrap() wrap()
class WS2812B: class WS2812B:
def __init__(self, num_leds, pin_num, brightness): def __init__(self, num_leds, pin_num, brightness, sm):
# Create the StateMachine with the ws2812 program, outputting on pin # Create the StateMachine with the ws2812 program, outputting on pin
self.sm = rp2.StateMachine(0, ws2812, freq=8_000_000, sideset_base=Pin(pin_num)) self.sm = rp2.StateMachine(sm, ws2812, freq=8_000_000, sideset_base=Pin(pin_num))
# Start the StateMachine, it will wait for data on its FIFO. # Start the StateMachine, it will wait for data on its FIFO.
self.sm.active(1) self.sm.active(1)
# Display a pattern on the LEDs via an array of LED RGB values. # Display a pattern on the LEDs via an array of LED RGB values.
self.ar = array.array("I", [0 for _ in range(num_leds)]) self.ar = array.array("I", [0 for _ in range(num_leds)])
self.num_leds = num_leds self.num_leds = num_leds
self.brightness = brightness self.brightness = brightness
########################################################################## ##########################################################################
def pixels_show(self): async def pixels_show(self):
dimmer_ar = array.array("I", [0 for _ in range(self.num_leds)]) self.sm.put(self.ar, 8)
for i,c in enumerate(self.ar): await asyncio.sleep(0.00001)
r = int(((c >> 8) & 0xFF) * self.brightness)
g = int(((c >> 16) & 0xFF) * self.brightness) def pixels_set(self, i, color):
b = int((c & 0xFF) * self.brightness) self.ar[i] = (int(color[1]*self.brightness)<<16) + (int(color[0]*self.brightness)<<8) + int(color[2]*self.brightness)
dimmer_ar[i] = (g<<16) + (r<<8) + b
self.sm.put(dimmer_ar, 8) def pixels_fill(self, color):
time.sleep_ms(10) for i in range(len(self.ar)):
self.pixels_set(i, color)
def pixels_set(self, i, color):
self.ar[i] = (color[1]<<16) + (color[0]<<8) + color[2] async def color_chase(self, color, wait):
for i in range(self.num_leds):
def pixels_fill(self, color): self.pixels_set(i, color)
for i in range(len(self.ar)): await asyncio.sleep(wait)
self.pixels_set(i, color) await self.pixels_show()
await asyncio.sleep(0.2)
def color_chase(self, color, wait):
for i in range(self.num_leds): def wheel(self, pos):
self.pixels_set(i, color) # Input a value 0 to 255 to get a color value.
time.sleep(wait) # The colours are a transition r - g - b - back to r.
self.pixels_show() if pos < 0 or pos > 255:
time.sleep(0.2) return (0, 0, 0)
if pos < 85:
def wheel(self, pos): return (255 - pos * 3, pos * 3, 0)
# Input a value 0 to 255 to get a color value. if pos < 170:
# The colours are a transition r - g - b - back to r. pos -= 85
if pos < 0 or pos > 255: return (0, 255 - pos * 3, pos * 3)
return (0, 0, 0) pos -= 170
if pos < 85: return (pos * 3, 0, 255 - pos * 3)
return (255 - pos * 3, pos * 3, 0)
if pos < 170:
pos -= 85 async def rainbow_cycle(self, wait):
return (0, 255 - pos * 3, pos * 3) for j in range(255):
pos -= 170 for i in range(self.num_leds):
return (pos * 3, 0, 255 - pos * 3) rc_index = (i * 256 // self.num_leds) + j
self.pixels_set(i, self.wheel(rc_index & 255))
await self.pixels_show()
def rainbow_cycle(self, wait): await asyncio.sleep(wait)
for j in range(255):
for i in range(self.num_leds): BLACK = (0, 0, 0)
rc_index = (i * 256 // self.num_leds) + j RED = (255, 0, 0)
self.pixels_set(i, self.wheel(rc_index & 255)) YELLOW = (255, 150, 0)
self.pixels_show() GREEN = (0, 255, 0)
time.sleep(wait) CYAN = (0, 255, 255)
BLUE = (0, 0, 255)
BLACK = (0, 0, 0) PURPLE = (180, 0, 255)
RED = (255, 0, 0) WHITE = (255, 255, 255)
YELLOW = (255, 150, 0) COLORS = (BLACK, RED, YELLOW, GREEN, CYAN, BLUE, PURPLE, WHITE)
GREEN = (0, 255, 0)
CYAN = (0, 255, 255)
BLUE = (0, 0, 255) async def run(pin, sm):
PURPLE = (180, 0, 255) ws = WS2812B(10, pin, 1, sm)
WHITE = (255, 255, 255) ws.pixels_fill(ws.RED)
COLORS = (BLACK, RED, YELLOW, GREEN, CYAN, BLUE, PURPLE, WHITE) while True:
await ws.rainbow_cycle(0)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
asyncio.create_task(run(0,0))
asyncio.create_task(run(17,1))
asyncio.create_task(run(18,2))
loop.run_forever()