Skip to content

Commit

Permalink
Clean up blocking_wait_for_edge (#104)
Browse files Browse the repository at this point in the history
GPIO object need to be cleaned up from the list when the function return
  • Loading branch information
anhmiuhv authored Jan 9, 2024
1 parent aab2ffd commit 8474cb3
Show file tree
Hide file tree
Showing 2 changed files with 10 additions and 5 deletions.
10 changes: 7 additions & 3 deletions lib/python/Jetson/GPIO/gpio_event.py
Original file line number Diff line number Diff line change
Expand Up @@ -460,6 +460,8 @@ def blocking_wait_for_edge(chip_fd, chip_name, channel, request, bouncetime, tim
except (OSError, IOError) as e:
raise cdev.GPIOError(e.errno, "Opening input line event handle: " + e.strerror)
gpio_obj = _Gpios(request.fd, bouncetime)
# As object is added to gpio event list here, we need to return it when
# the function return
_add_gpio_event(chip_name, channel, gpio_obj)

ret = select.select([request.fd], [], [], timeout)
Expand All @@ -468,21 +470,23 @@ def blocking_wait_for_edge(chip_fd, chip_name, channel, request, bouncetime, tim
try:
data = os.read(request.fd, ctypes.sizeof(cdev.gpioevent_data))
except OSError as e:
remove_edge_detect(chip_name, channel)
raise cdev.GPIOError(e.errno, "Reading GPIO event: " + e.strerror)

event_data = cdev.gpioevent_data.from_buffer_copy(data)

if (event_data.id != cdev.GPIOEVENT_REQUEST_RISING_EDGE and
event_data.id != cdev.GPIOEVENT_REQUEST_FALLING_EDGE):
warnings.warn("Unknown event caught", RuntimeWarning)

remove_edge_detect(chip_name, channel)
return -2

remove_edge_detect(chip_name, channel)
return int(ret != [])
elif len(ret[0]) == 0:
# Timeout
remove_edge_detect(chip_name, channel)
return 0

remove_edge_detect(chip_name, channel)
return -2


Expand Down
5 changes: 3 additions & 2 deletions samples/test_all_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -506,8 +506,9 @@ def test_wait_for_edge_timeout():
GPIO.setmode(GPIO.BOARD)
GPIO.setup(pin_data['out_a'], GPIO.OUT, initial=GPIO.LOW)
GPIO.setup(pin_data['in_a'], GPIO.IN)
val = GPIO.wait_for_edge(pin_data['in_a'], GPIO.BOTH, timeout=5)
assert val is None
for i in range(3):
val = GPIO.wait_for_edge(pin_data['in_a'], GPIO.BOTH, timeout=1)
assert val is None
GPIO.cleanup()


Expand Down

0 comments on commit 8474cb3

Please sign in to comment.