Moved state to mutable dataclass

This commit is contained in:
2026-07-29 17:30:43 +01:00
parent 690a013cd0
commit 6a73b7622d

51
main.py
View File

@@ -23,12 +23,19 @@ class DiskEvent:
DiskState = Literal["disconnected", "pairing", "connecting", "connected"] DiskState = Literal["disconnected", "pairing", "connecting", "connected"]
def onButtonPressed(diskState: DiskState) -> None: @dataclass
diskState = "pairing" class AppState:
stopEvent = Event()
diskEventQueue: Queue[DiskEvent] = Queue()
diskState: DiskState = "disconnected"
def onButtonPressed(state: AppState) -> None:
state.diskState = "pairing"
print("Changing disk state") print("Changing disk state")
def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None: def diskPollWorker(state) -> None:
context = pyudev.Context() context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context) monitor = pyudev.Monitor.from_netlink(context)
@@ -38,7 +45,7 @@ def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
for device in iter(monitor.poll, None): for device in iter(monitor.poll, None):
if device.action == "add" and device.device_type == "disk": if device.action == "add" and device.device_type == "disk":
eventQueue.put( state.diskEventQueue.put(
DiskEvent( DiskEvent(
event="connected", event="connected",
node=f"{device.device_node}", node=f"{device.device_node}",
@@ -47,7 +54,7 @@ def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
) )
elif device.action == "remove" and device.device_type == "disk": elif device.action == "remove" and device.device_type == "disk":
eventQueue.put( state.diskEventQueue.put(
DiskEvent( DiskEvent(
event="disconnected", event="disconnected",
node=f"{device.device_node}", node=f"{device.device_node}",
@@ -55,32 +62,32 @@ def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
) )
) )
if stopEvent.is_set(): if state.stopEvent.is_set():
return return
def ledWorker(stopEvent: Event, diskState: DiskState) -> None: def ledWorker(state: AppState) -> None:
yellowCtr = 0 yellowCtr = 0
previousState: DiskState | None = None previousState: DiskState | None = None
while not stopEvent.is_set(): while not state.stopEvent.is_set():
if diskState != previousState: if state.diskState != previousState:
yellowCtr = 0 yellowCtr = 0
previousState = diskState previousState = state.diskState
if diskState == "disconnected": if state.diskState == "disconnected":
yellow.off() yellow.off()
elif diskState == "connected": elif state.diskState == "connected":
yellow.on() yellow.on()
elif diskState == "pairing": elif state.diskState == "pairing":
if yellowCtr % 2 == 0: if yellowCtr % 2 == 0:
yellow.on() yellow.on()
else: else:
yellow.off() yellow.off()
elif diskState == "connecting": elif state.diskState == "connecting":
if yellowCtr % 4 < 2: if yellowCtr % 4 < 2:
yellow.on() yellow.on()
else: else:
@@ -88,20 +95,18 @@ def ledWorker(stopEvent: Event, diskState: DiskState) -> None:
yellowCtr += 1 yellowCtr += 1
stopEvent.wait(0.25) state.stopEvent.wait(0.25)
def main() -> None: def main() -> None:
try: try:
stopEvent = Event() state = AppState()
diskEventQueue: Queue[DiskEvent] = Queue() button.when_activated = partial(onButtonPressed, state)
diskState: DiskState = "disconnected"
button.when_activated = partial(onButtonPressed, diskState)
diskPollThread = Thread( diskPollThread = Thread(
target=diskPollWorker, target=diskPollWorker,
name="disk_poll_thread", name="disk_poll_thread",
args=(stopEvent, diskEventQueue), args=(state),
daemon=True, daemon=True,
) )
@@ -110,7 +115,7 @@ def main() -> None:
ledThread = Thread( ledThread = Thread(
target=ledWorker, target=ledWorker,
name="led_thread", name="led_thread",
args=(stopEvent, diskState), args=(state),
daemon=True, daemon=True,
) )
@@ -118,13 +123,13 @@ def main() -> None:
while True: while True:
try: try:
event = diskEventQueue.get_nowait() event = state.diskEventQueue.get_nowait()
print(event) print(event)
except Empty: except Empty:
sleep(1) sleep(1)
finally: finally:
stopEvent.set() state.stopEvent.set()
return return