def run(self):
self.exc = None
try:
sysfs.edge(self._pin, self._trigger)
initial_edge = True
with sysfs.value_descriptor(self._pin) as fd:
e = select.epoll()
e.register(fd, EPOLLIN | EPOLLET | EPOLLPRI)
try:
while not self._finished:
events = e.poll(0.1, maxevents=1)
if initial_edge:
initial_edge = False
elif len(events) > 0:
with self._lock:
self._event_detected = True
self.notify_callbacks()
finally:
e.unregister(fd)
e.close()
except BaseException as e:
self.exc = e
finally:
sysfs.edge(self._pin, NONE)
评论列表
文章目录