78 lines
1.8 KiB
Python
78 lines
1.8 KiB
Python
from queue import Empty, Queue
|
|
from time import sleep
|
|
import pyudev
|
|
from gpiozero import LED
|
|
from threading import Thread, Event
|
|
from dataclasses import dataclass
|
|
|
|
red = LED(17)
|
|
yellow = LED(27)
|
|
blue = LED(22)
|
|
|
|
|
|
@dataclass
|
|
class DiskEvent:
|
|
event: str
|
|
node: str
|
|
serial: str
|
|
|
|
|
|
def diskPollWorker(stopEvent: Event, eventQueue: Queue) -> None:
|
|
context = pyudev.Context()
|
|
|
|
monitor = pyudev.Monitor.from_netlink(context)
|
|
monitor.filter_by(subsystem="block")
|
|
|
|
print("Waiting for disks...")
|
|
|
|
for device in iter(monitor.poll, None):
|
|
if device.action == "add" and device.device_type == "disk":
|
|
eventQueue.put(
|
|
DiskEvent(
|
|
event="connected",
|
|
node=f"{device.device_node}",
|
|
serial=f"{device.get('ID_SERIAL_SHORT', 'Unknown')}",
|
|
)
|
|
)
|
|
|
|
elif device.action == "remove" and device.device_type == "disk":
|
|
eventQueue.put(
|
|
DiskEvent(
|
|
event="disconnected",
|
|
node=f"{device.device_node}",
|
|
serial=f"{device.get('ID_SERIAL_SHORT', 'Unknown')}",
|
|
)
|
|
)
|
|
|
|
if stopEvent.is_set():
|
|
return
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
stopEvent = Event()
|
|
diskEventQueue: Queue[DiskEvent] = Queue()
|
|
|
|
diskPollThread = Thread(
|
|
target=diskPollWorker,
|
|
name="disk_poll_thread",
|
|
args=(stopEvent, diskEventQueue),
|
|
daemon=True,
|
|
)
|
|
|
|
diskPollThread.start()
|
|
|
|
try:
|
|
event = diskEventQueue.get_nowait()
|
|
print(event)
|
|
except Empty:
|
|
sleep(1)
|
|
|
|
finally:
|
|
stopEvent.set()
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|