64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
from queue import Queue
|
|
from time import sleep
|
|
from typing import final
|
|
import pyudev
|
|
from gpiozero import Device
|
|
from gpiozero.pins.mock import MockFactory
|
|
from gpiozero import LED
|
|
from threading import Thread, Event
|
|
|
|
red = LED(17)
|
|
yellow = LED(27)
|
|
blue = LED(22)
|
|
|
|
|
|
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":
|
|
print("\nDisk connected")
|
|
print(f"Device: {device.device_node}")
|
|
print(f"Name: {device.sys_name}")
|
|
print(f"Model: {device.get('ID_MODEL', 'Unknown')}")
|
|
print(f"Serial: {device.get('ID_SERIAL_SHORT', 'Unknown')}")
|
|
print(f"Bus: {device.get('ID_BUS', 'Unknown')}")
|
|
|
|
elif device.action == "remove" and device.device_type == "disk":
|
|
print("\nDisk disconnected")
|
|
print(f"Name: {device.sys_name}")
|
|
|
|
if stopEvent.is_set():
|
|
return
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
stopEvent = Event()
|
|
diskEventQueue = Queue()
|
|
|
|
diskPollThread = Thread(
|
|
target=diskPollWorker,
|
|
name="disk_poll_thread",
|
|
args=(stopEvent, diskEventQueue),
|
|
daemon=True,
|
|
)
|
|
|
|
diskPollThread.start()
|
|
|
|
while True:
|
|
sleep(1)
|
|
|
|
finally:
|
|
stopEvent.set()
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|