Skip to content

Commit 1c7fbe4

Browse files
committed
move Cooperative Multitasking code to repo
1 parent 12b023f commit 1c7fbe4

8 files changed

Lines changed: 323 additions & 0 deletions

File tree

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
import asyncio
2+
3+
import board
4+
import keypad
5+
import neopixel
6+
from rainbowio import colorwheel
7+
8+
pixel_pin = board.A0
9+
num_pixels = 24
10+
11+
pixels = neopixel.NeoPixel(pixel_pin, num_pixels, brightness=0.03, auto_write=False)
12+
13+
14+
class Controls:
15+
def __init__(self):
16+
self.reverse = False
17+
self.wait = 0.0
18+
19+
20+
async def rainbow_cycle(controls):
21+
while True:
22+
# Increment by 2 instead of 1 to speed the cycle up a bit.
23+
for j in range(255, -1, -2) if controls.reverse else range(0, 256, 2):
24+
for i in range(num_pixels):
25+
rc_index = (i * 256 // num_pixels) + j
26+
pixels[i] = colorwheel(rc_index & 255)
27+
pixels.show()
28+
await asyncio.sleep(controls.wait)
29+
30+
31+
async def monitor_buttons(reverse_pin, slower_pin, faster_pin, controls):
32+
"""Monitor buttons that reverse direction and change animation speed.
33+
Assume buttons are active low.
34+
"""
35+
with keypad.Keys(
36+
(reverse_pin, slower_pin, faster_pin), value_when_pressed=False, pull=True
37+
) as keys:
38+
while True:
39+
key_event = keys.events.get()
40+
if key_event and key_event.pressed:
41+
key_number = key_event.key_number
42+
if key_number == 0:
43+
controls.reverse = not controls.reverse
44+
elif key_number == 1:
45+
# Lengthen the interval.
46+
controls.wait = controls.wait + 0.001
47+
elif key_number == 2:
48+
# Shorten the interval.
49+
controls.wait = max(0.0, controls.wait - 0.001)
50+
# Let another task run.
51+
await asyncio.sleep(0)
52+
53+
54+
async def main():
55+
controls = Controls()
56+
57+
buttons_task = asyncio.create_task(
58+
monitor_buttons(board.A1, board.A2, board.A3, controls)
59+
)
60+
animation_task = asyncio.create_task(rainbow_cycle(controls))
61+
62+
# This will run forever, because no tasks ever finish.
63+
await asyncio.gather(buttons_task, animation_task)
64+
65+
66+
asyncio.run(main())
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
import board
2+
import digitalio
3+
import time
4+
5+
6+
def blink(pin, interval, count):
7+
with digitalio.DigitalInOut(pin) as led:
8+
led.switch_to_output(value=False)
9+
for _ in range(count):
10+
led.value = True
11+
time.sleep(interval)
12+
led.value = False
13+
time.sleep(interval)
14+
15+
16+
def main():
17+
blink(board.D1, 0.25, 10)
18+
print("done")
19+
20+
21+
main()
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
# DOESN'T WORK
2+
import board
3+
import digitalio
4+
import time
5+
6+
7+
def blink(pin, interval, count):
8+
with digitalio.DigitalInOut(pin) as led:
9+
led.switch_to_output(value=False)
10+
for _ in range(count):
11+
led.value = True
12+
time.sleep(interval)
13+
led.value = False
14+
time.sleep(interval)
15+
16+
17+
def main():
18+
blink(board.D1, 0.25, 10)
19+
# DOESN'T WORK
20+
# Second LED blinks only after the first one is finished.
21+
blink(board.D2, 0.1, 20)
22+
23+
24+
main()
Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
import time
2+
3+
import board
4+
import digitalio
5+
6+
7+
class Blinker:
8+
def __init__(self, led, interval, count):
9+
self.led = led
10+
self.interval = interval
11+
# Count both on and off.
12+
self.count2 = count * 2
13+
self.last_transition = 0
14+
15+
def blink(self):
16+
"""Return False when blinking is finished."""
17+
if self.count2 <= 0:
18+
return False
19+
now = time.monotonic()
20+
if now > self.last_transition + self.interval:
21+
self.led.value = not self.led.value
22+
self.last_transition = now
23+
self.count2 -= 1
24+
return True
25+
26+
27+
def main():
28+
with digitalio.DigitalInOut(board.D1) as led1, digitalio.DigitalInOut(
29+
board.D2
30+
) as led2:
31+
led1.switch_to_output(value=False)
32+
led2.switch_to_output(value=False)
33+
34+
blinker1 = Blinker(led1, 0.25, 10)
35+
blinker2 = Blinker(led2, 0.1, 20)
36+
running1 = True
37+
running2 = True
38+
while running1 or running2:
39+
running1 = blinker1.blink()
40+
running2 = blinker2.blink()
41+
print("done")
42+
43+
44+
main()
Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
import asyncio
2+
import board
3+
import digitalio
4+
import keypad
5+
6+
7+
class Interval:
8+
"""Simple class to hold an interval value. Use .value to to read or write."""
9+
10+
def __init__(self, initial_interval):
11+
self.value = initial_interval
12+
13+
14+
async def monitor_interval_buttons(pin_slower, pin_faster, interval):
15+
"""Monitor two buttons: one lengthens the interval, the other shortens it.
16+
Change interval.value as appropriate.
17+
"""
18+
# Assume buttons are active low.
19+
with keypad.Keys(
20+
(pin_slower, pin_faster), value_when_pressed=False, pull=True
21+
) as keys:
22+
while True:
23+
key_event = keys.events.get()
24+
if key_event and key_event.pressed:
25+
if key_event.key_number == 0:
26+
# Lengthen the interval.
27+
interval.value += 0.1
28+
else:
29+
# Shorten the interval.
30+
interval.value = max(0.1, interval.value - 0.1)
31+
print("interval is now", interval.value)
32+
# Let another task run.
33+
await asyncio.sleep(0)
34+
35+
36+
async def blink(pin, interval):
37+
"""Blink the given pin forever.
38+
The blinking rate is controlled by the supplied Interval object.
39+
"""
40+
with digitalio.DigitalInOut(pin) as led:
41+
led.switch_to_output()
42+
while True:
43+
led.value = not led.value
44+
await asyncio.sleep(interval.value)
45+
46+
47+
async def main():
48+
# Start blinking 0.5 sec on, 0.5 sec off.
49+
interval = Interval(0.5)
50+
51+
led_task = asyncio.create_task(blink(board.D1, interval))
52+
interval_task = asyncio.create_task(
53+
monitor_interval_buttons(board.D3, board.D4, interval)
54+
)
55+
# This will run forever, because neither task ever exits.
56+
await asyncio.gather(led_task, interval_task)
57+
58+
59+
asyncio.run(main())
Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
import asyncio
2+
import board
3+
import digitalio
4+
5+
6+
async def blink(pin, interval, count): # Don't forget the async!
7+
with digitalio.DigitalInOut(pin) as led:
8+
led.switch_to_output(value=False)
9+
for _ in range(count):
10+
led.value = True
11+
await asyncio.sleep(interval) # Don't forget the await!
12+
led.value = False
13+
await asyncio.sleep(interval) # Don't forget the await!
14+
15+
16+
async def main(): # Don't forget the async!
17+
led_task = asyncio.create_task(blink(board.D1, 0.25, 10))
18+
await asyncio.gather(led_task) # Don't forget the await!
19+
print("done")
20+
21+
22+
asyncio.run(main())
Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
import asyncio
2+
import board
3+
import digitalio
4+
import keypad
5+
6+
7+
class Interval:
8+
"""Simple class to hold an interval value. Use .value to to read or write."""
9+
10+
def __init__(self, initial_interval):
11+
self.value = initial_interval
12+
13+
14+
async def monitor_interval_buttons(pin_slower, pin_faster, interval):
15+
"""Monitor two buttons: one lengthens the interval, the other shortens it.
16+
Change interval.value as appropriate.
17+
"""
18+
# Assume buttons are active low.
19+
with keypad.Keys(
20+
(pin_slower, pin_faster), value_when_pressed=False, pull=True
21+
) as keys:
22+
while True:
23+
key_event = keys.events.get()
24+
if key_event and key_event.pressed:
25+
if key_event.key_number == 0:
26+
# Lengthen the interval.
27+
interval.value += 0.1
28+
else:
29+
# Shorten the interval.
30+
interval.value = max(0.1, interval.value - 0.1)
31+
print("interval is now", interval.value)
32+
# Let another task run.
33+
await asyncio.sleep(0)
34+
35+
36+
async def blink(pin, interval):
37+
"""Blink the given pin forever.
38+
The blinking rate is controlled by the supplied Interval object.
39+
"""
40+
with digitalio.DigitalInOut(pin) as led:
41+
led.switch_to_output()
42+
while True:
43+
led.value = not led.value
44+
await asyncio.sleep(interval.value)
45+
46+
47+
async def main():
48+
interval1 = Interval(0.5)
49+
interval2 = Interval(1.0)
50+
51+
led1_task = asyncio.create_task(blink(board.D1, interval1))
52+
led2_task = asyncio.create_task(blink(board.D2, interval2))
53+
interval1_task = asyncio.create_task(
54+
monitor_interval_buttons(board.D3, board.D4, interval1)
55+
)
56+
interval2_task = asyncio.create_task(
57+
monitor_interval_buttons(board.D5, board.D6, interval2)
58+
)
59+
60+
await asyncio.gather(led1_task, led2_task, interval1_task, interval2_task)
61+
62+
63+
asyncio.run(main())
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
import asyncio
2+
import board
3+
import digitalio
4+
5+
6+
async def blink(pin, interval, count):
7+
with digitalio.DigitalInOut(pin) as led:
8+
led.switch_to_output(value=False)
9+
for _ in range(count):
10+
led.value = True
11+
await asyncio.sleep(interval) # Don't forget the "await"!
12+
led.value = False
13+
await asyncio.sleep(interval) # Don't forget the "await"!
14+
15+
16+
async def main():
17+
led1_task = asyncio.create_task(blink(board.D1, 0.25, 10))
18+
led2_task = asyncio.create_task(blink(board.D2, 0.1, 20))
19+
20+
await asyncio.gather(led1_task, led2_task) # Don't forget "await"!
21+
print("done")
22+
23+
24+
asyncio.run(main())

0 commit comments

Comments
 (0)