mirror of
https://git.kernel.org/pub/scm/libs/libgpiod/libgpiod.git
synced 2026-10-25 12:39:18 +03:00
485b81cd4b
* Sort imports * Use f-strings where possible * Remove usage of `tuple` for config key values Signed-off-by: Vincent Fazio <vfazio@gmail.com> Link: https://patch.msgid.link/20260401001459.19159-7-vfazio@gmail.com Signed-off-by: Bartosz Golaszewski <bartosz.golaszewski@oss.qualcomm.com>
80 lines
2.5 KiB
Python
Executable File
80 lines
2.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
# SPDX-FileCopyrightText: 2023 Kent Gibson <warthog618@gmail.com>
|
|
|
|
"""Minimal example of asynchronously watching for edges on a single line."""
|
|
|
|
import select
|
|
from datetime import timedelta
|
|
|
|
import gpiod
|
|
from gpiod.edge_event import EdgeEvent
|
|
from gpiod.line import Bias, Edge
|
|
|
|
|
|
def edge_type_str(event: EdgeEvent) -> str:
|
|
if event.event_type is event.Type.RISING_EDGE:
|
|
return "Rising"
|
|
if event.event_type is event.Type.FALLING_EDGE:
|
|
return "Falling"
|
|
return "Unknown"
|
|
|
|
|
|
def async_watch_line_value(chip_path: str, line_offset: int, done_fd: int) -> None:
|
|
# Assume a button connecting the pin to ground,
|
|
# so pull it up and provide some debounce.
|
|
with gpiod.request_lines(
|
|
chip_path,
|
|
consumer="async-watch-line-value",
|
|
config={
|
|
line_offset: gpiod.LineSettings(
|
|
edge_detection=Edge.BOTH,
|
|
bias=Bias.PULL_UP,
|
|
debounce_period=timedelta(milliseconds=10),
|
|
)
|
|
},
|
|
) as request:
|
|
poll = select.poll()
|
|
poll.register(request.fd, select.POLLIN)
|
|
# Other fds could be registered with the poll and be handled
|
|
# separately using the return value (fd, event) from poll():
|
|
poll.register(done_fd, select.POLLIN)
|
|
while True:
|
|
for fd, _event in poll.poll():
|
|
if fd == done_fd:
|
|
# perform any cleanup before exiting...
|
|
return
|
|
# handle any edge events
|
|
for event in request.read_edge_events():
|
|
print(
|
|
f"offset: {event.line_offset}"
|
|
f" type: {edge_type_str(event):<7}"
|
|
f" event #{event.line_seqno}"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import os
|
|
import threading
|
|
|
|
# run the async executor (select.poll) in a thread to demonstrate a graceful exit.
|
|
done_fd = os.eventfd(0)
|
|
|
|
def bg_thread() -> None:
|
|
try:
|
|
async_watch_line_value("/dev/gpiochip0", 5, done_fd)
|
|
except OSError as ex:
|
|
print(ex, "\nCustomise the example configuration to suit your situation")
|
|
print("background thread exiting...")
|
|
|
|
t = threading.Thread(target=bg_thread)
|
|
t.start()
|
|
|
|
# Wait for two minutes, unless bg_thread exits earlier, then graceful exit.
|
|
t.join(timeout=120)
|
|
if t.is_alive():
|
|
os.eventfd_write(done_fd, 1)
|
|
t.join()
|
|
os.close(done_fd)
|
|
print("main thread exiting...")
|