我正在使用伟大的"evdev"库来听USB条形码阅读器输入,我需要检测设备是否突然拔出/无响应,否则python脚本读取循环在单个线程上达到100%的cpu使用率,并慢慢开始吃所有可用的内存,导致整个系统崩溃后一点。
这个想法是检测设备何时拔出并杀死当前脚本,导致管理员试图重新启动它,直到设备重新插入/响应。
我用来读取输入的代码如下:devices = map(InputDevice, list_devices())
keys = {
2: 1,
3: 2,
4: 3,
5: 4,
6: 5,
7: 6,
8: 7,
9: 8,
10: 9,
11: 0,
}
dev = None
for d in devices:
if d.name == 'Symbol Technologies, Inc, 2008 Symbol Bar Code Scanner':
print('%-20s %-32s %s' % (d.fn, d.name, d.phys))
dev = InputDevice(d.fn)
break
if dev is not None:
code = []
for event in dev.read_loop():
if event.type == ecodes.EV_KEY:
if event.value == 00:
if event.code != 96:
try:
code.append(keys[event.code])
except:
code.append('-')
else:
card = "".join(map(str, code))
print card
code = []
card = ""
那么我该如何以正确的方式去做呢?
我虽然这可能的工作方式将第二个脚本的运行从cron每1 - 5分钟检查说设备仍然可用,如果是现在,抓住一些文件和杀死进程的进程id,但这种方法的问题是,如果设备是不插电,然后插回检查"检验员"脚本之间的认为一切都是好的在主脚本慢慢崩溃——它不会重新激活后"拔掉"
python-evdev作者。知道自己的工作对别人有用是一种很棒的感觉。谢谢你!
你一定要看看linux的设备管理器- udev。每当添加或删除设备时,linux内核都会发出事件。要在Python程序中监听这些事件,您可以使用pyudev,它是一个出色的、基于ctypes的绑定到libdev(请参阅监视部分)。
下面是一个使用evdev
和pyudev
的例子:
import functools
import pyudev
from evdev import InputDevice
from select import select
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem='input')
monitor.start()
fds = {monitor.fileno(): monitor}
finalizers = []
while True:
r, w, x = select(fds, [], [])
if monitor.fileno() in r:
r.remove(monitor.fileno())
for udev in iter(functools.partial(monitor.poll, 0), None):
# we're only interested in devices that have a device node
# (e.g. /dev/input/eventX)
if not udev.device_node:
break
# find the device we're interested in and add it to fds
for name in (i['NAME'] for i in udev.ancestors if 'NAME' in i):
# I used a virtual input device for this test - you
# should adapt this to your needs
if u'py-evdev-uinput' in name:
if udev.action == u'add':
print('Device added: %s' % udev)
fds[dev.fd] = InputDevice(udev.device_node)
break
if udev.action == u'remove':
print('Device removed: %s' % udev)
def helper():
global fds
fds = {monitor.fileno(): monitor}
finalizers.append(helper)
break
for fd in r:
dev = fds[fd]
for event in dev.read():
print(event)
for i in range(len(finalizers)):
finalizers.pop()()