-
Notifications
You must be signed in to change notification settings - Fork 0
/
thread.py
52 lines (40 loc) · 1.33 KB
/
thread.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from enum import Enum
from enum import auto
from itertools import cycle
from typing import Callable
from typing import Generator
from typing import Iterator
from typing import Sequence
class Thread(Enum):
RED = 0
GRN = auto()
BLU = auto()
YEL = auto()
ThreadIteratorFactory = Callable[[], Iterator[Thread]]
def cycle_threads(threads: Sequence[Thread], offset: int=0) -> Callable[[], Iterator[Thread]]:
"""Returns a function that returns an itertools.cycle over a sequence of Threads"""
if len(threads) == 0:
raise ValueError("sequence given to cycle_threads is empty")
def func():
gen = cycle(threads)
for _ in range(offset):
next(gen)
return gen
return func
def green_then_all_red_threads() -> Generator[Thread, None, None]:
yield Thread.GRN
while True:
yield Thread.RED
RGB = Thread.RED, Thread.GRN, Thread.BLU
BGR = tuple(reversed(RGB))
RGBY = Thread.RED, Thread.GRN, Thread.BLU, Thread.YEL
YBGR = tuple(reversed(RGBY))
THREAD_GENS: dict[str, ThreadIteratorFactory] = {
'g-r..': green_then_all_red_threads,
'r-g-b': cycle_threads(RGB),
'g-b-r': cycle_threads(RGB, offset=1),
'b-r-g': cycle_threads(RGB, offset=2),
'b-g-r': cycle_threads(BGR),
'g-r-b': cycle_threads(BGR, offset=1),
'r-b-g': cycle_threads(BGR, offset=2),
}