Added led worker thread and disk state
This commit is contained in:
29
main.py
29
main.py
@@ -1,5 +1,6 @@
|
|||||||
from queue import Empty, Queue
|
from queue import Empty, Queue
|
||||||
from time import sleep
|
from time import sleep
|
||||||
|
from typing import Literal
|
||||||
import pyudev
|
import pyudev
|
||||||
from gpiozero import LED
|
from gpiozero import LED
|
||||||
from threading import Thread, Event
|
from threading import Thread, Event
|
||||||
@@ -17,6 +18,9 @@ class DiskEvent:
|
|||||||
serial: str
|
serial: str
|
||||||
|
|
||||||
|
|
||||||
|
DiskState = Literal["disconnected", "pairing", "connected"]
|
||||||
|
|
||||||
|
|
||||||
def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
|
def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
|
||||||
context = pyudev.Context()
|
context = pyudev.Context()
|
||||||
|
|
||||||
@@ -48,10 +52,26 @@ def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
|
|||||||
return
|
return
|
||||||
|
|
||||||
|
|
||||||
|
def ledWorker(stopEvent: Event, diskState: DiskState) -> None:
|
||||||
|
yellowCtr: int = 0
|
||||||
|
while not stopEvent.is_set():
|
||||||
|
if diskState == "disconnected":
|
||||||
|
yellow.off()
|
||||||
|
elif diskState == "connected":
|
||||||
|
yellow.on()
|
||||||
|
elif diskState == "pairing":
|
||||||
|
if yellowCtr % 2 == 0:
|
||||||
|
yellow.on()
|
||||||
|
else:
|
||||||
|
yellow.off()
|
||||||
|
sleep(1)
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
try:
|
try:
|
||||||
stopEvent = Event()
|
stopEvent = Event()
|
||||||
diskEventQueue: Queue[DiskEvent] = Queue()
|
diskEventQueue: Queue[DiskEvent] = Queue()
|
||||||
|
diskState: DiskState = "pairing"
|
||||||
|
|
||||||
diskPollThread = Thread(
|
diskPollThread = Thread(
|
||||||
target=diskPollWorker,
|
target=diskPollWorker,
|
||||||
@@ -62,6 +82,15 @@ def main() -> None:
|
|||||||
|
|
||||||
diskPollThread.start()
|
diskPollThread.start()
|
||||||
|
|
||||||
|
ledThread = Thread(
|
||||||
|
target=ledWorker,
|
||||||
|
name="led_thread",
|
||||||
|
args=(stopEvent, diskState),
|
||||||
|
daemon=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
ledThread.start()
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
event = diskEventQueue.get_nowait()
|
event = diskEventQueue.get_nowait()
|
||||||
|
|||||||
Reference in New Issue
Block a user