# Copyright (c) 2012-2017 Ben Croston . # Copyright (c) 2019-2023, NVIDIA CORPORATION. All rights reserved. # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), # to deal in the Software without restriction, including without limitation # the rights to use, copy, modify, merge, publish, distribute, sublicense, # and/or sell copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL # THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER # DEALINGS IN THE SOFTWARE. from Jetson.GPIO import gpio_event as event from Jetson.GPIO import gpio_pin_data from Jetson.GPIO import gpio_cdev import os import warnings import time # sysfs root _GPIOCHIP_ROOT = "/dev/gpiochip0" if not os.access(_GPIOCHIP_ROOT, os.W_OK): raise RuntimeError("The current user does not have permissions set to access the library functionalites. Please configure permissions or use the root user to run this. It is also possible that {} does not exist. Please check if that file is present.".format(_GPIOCHIP_ROOT)) # Pin Numbering Modes BOARD = 10 BCM = 11 TEGRA_SOC = 1000 CVM = 1001 # The constants and their offsets are implemented to prevent HIGH from being # used in place of other variables (ie. HIGH and RISING should not be # interchangeable) # Pull up/down options _PUD_OFFSET = 20 PUD_OFF = 0 + _PUD_OFFSET PUD_DOWN = 1 + _PUD_OFFSET PUD_UP = 2 + _PUD_OFFSET HIGH = 1 LOW = 0 # Edge possibilities # These values (with _EDGE_OFFSET subtracted) must match gpio_event.py:*_EDGE _EDGE_OFFSET = 30 RISING = 1 + _EDGE_OFFSET FALLING = 2 + _EDGE_OFFSET BOTH = 3 + _EDGE_OFFSET # GPIO directions. UNKNOWN constant is for gpios that are not yet setup UNKNOWN = -1 OUT = 0 IN = 1 HARD_PWM = 43 model, JETSON_INFO, _channel_data_by_mode = gpio_pin_data.get_data() RPI_INFO = JETSON_INFO # Dictionary used as a lookup table for pin to its info object (_Gpios) mapping # key: channel, value: ChannelInfo object _channel_data = {} _gpio_warnings = True _gpio_mode = None # Dictionary used as a lookup table for pin to its configuration # key: channel, value: GPIO directions (IN/OUT deprecated) _channel_configuration = {} # Dictionary used as a lookup table from GPIO chip name to chip fd _chip_fd = {} def _validate_mode_set(): if _gpio_mode is None: raise RuntimeError("Please set pin numbering mode using " "GPIO.setmode(GPIO.BOARD), GPIO.setmode(GPIO.BCM), " "GPIO.setmode(GPIO.TEGRA_SOC) or " "GPIO.setmode(GPIO.CVM)") def _make_iterable(iterable, single_length=None): if isinstance(iterable, str): iterable = [iterable] try: for x in iterable: break except: iterable = [iterable] if single_length is not None and len(iterable) == 1: iterable = iterable * single_length return iterable def _channel_to_info_lookup(channel, need_gpio, need_pwm): if channel not in _channel_data: raise ValueError("Channel %s is invalid" % str(channel)) ch_info = _channel_data[channel] if need_pwm and ch_info.pwm_chip_dir is None: raise ValueError("Channel %s is not a PWM" % str(channel)) return ch_info def _channel_to_info(channel, need_gpio=False, need_pwm=False): _validate_mode_set() return _channel_to_info_lookup(channel, need_gpio, need_pwm) def _channels_to_infos(channels, need_gpio=False, need_pwm=False): _validate_mode_set() return [_channel_to_info_lookup(c, need_gpio, need_pwm) for c in _make_iterable(channels)] def _sysfs_channel_configuration(ch_info): """Return the current configuration of a channel as reported by sysfs. Any of IN, OUT, PWM, or None may be returned.""" if ch_info.pwm_chip_dir is not None: pwm_dir = "%s/pwm%i" % (ch_info.pwm_chip_dir, ch_info.pwm_id) if os.path.exists(pwm_dir): return HARD_PWM return None def _app_channel_configuration(ch_info): """Return the current configuration of a channel as requested by this module in this process. Any of IN, OUT, or None may be returned.""" return _channel_configuration.get(ch_info.channel, None) def _chip_fd_map(ch_info): return _chip_fd.get(ch_info.gpio_chip, None) def _do_one_channel(ch_info, direction, initial, consumer): ch_info.chip_fd = _chip_fd_map(ch_info) if not ch_info.chip_fd: ch_info.chip_fd = gpio_cdev.chip_open_by_label(ch_info.gpio_chip) _chip_fd[ch_info.gpio_chip] = ch_info.chip_fd cdev_direction = gpio_cdev.GPIOHANDLE_REQUEST_OUTPUT if direction == OUT else gpio_cdev.GPIOHANDLE_REQUEST_INPUT request = gpio_cdev.request_handle(ch_info.line_offset, cdev_direction, initial, consumer) gpio_cdev.open_line(ch_info, request) _channel_configuration[ch_info.channel] = direction def _pwm_path(ch_info): return ch_info.pwm_chip_dir + '/pwm' + str(ch_info.pwm_id) def _pwm_export_path(ch_info): return ch_info.pwm_chip_dir + '/export' def _pwm_unexport_path(ch_info): return ch_info.pwm_chip_dir + '/unexport' def _pwm_period_path(ch_info): return _pwm_path(ch_info) + "/period" def _pwm_duty_cycle_path(ch_info): return _pwm_path(ch_info) + "/duty_cycle" def _pwm_enable_path(ch_info): return _pwm_path(ch_info) + "/enable" def _export_pwm(ch_info): if not os.path.exists(_pwm_path(ch_info)): with open(_pwm_export_path(ch_info), 'w') as f: f.write(str(ch_info.pwm_id)) enable_path = _pwm_enable_path(ch_info) while not os.access(enable_path, os.R_OK | os.W_OK): time.sleep(0.01) ch_info.f_duty_cycle = open(_pwm_duty_cycle_path(ch_info), 'r+') def _unexport_pwm(ch_info): ch_info.f_duty_cycle.close() with open(_pwm_unexport_path(ch_info), 'w') as f: f.write(str(ch_info.pwm_id)) def _set_pwm_period(ch_info, period_ns): with open(_pwm_period_path(ch_info), 'w') as f: f.write(str(period_ns)) def _set_pwm_duty_cycle(ch_info, duty_cycle_ns): # On boot, both period and duty cycle are both 0. In this state, the period # must be set first; any configuration change made while period==0 is # rejected. This is fine if we actually want a duty cycle of 0. Later, once # any period has been set, we will always be able to set a duty cycle of 0. # The code could be written to always read the current value, and only # write the value if the desired value is different. However, we enable # this check only for the 0 duty cycle case, to avoid having to read the # current value every time the duty cycle is set. if not duty_cycle_ns: ch_info.f_duty_cycle.seek(0) cur = ch_info.f_duty_cycle.read().strip() if cur == '0': return ch_info.f_duty_cycle.seek(0) ch_info.f_duty_cycle.write(str(duty_cycle_ns)) ch_info.f_duty_cycle.flush() def _enable_pwm(ch_info): with open(_pwm_enable_path(ch_info), 'w') as f: f.write("1") def _disable_pwm(ch_info): with open(_pwm_enable_path(ch_info), 'w') as f: f.write("0") # Clean up all resources taken by a channel, # including pwm, chip and lines def _cleanup_one(ch_info): #clean up pwm config app_cfg = _channel_configuration[ch_info.channel] if app_cfg == HARD_PWM: _disable_pwm(ch_info) _unexport_pwm(ch_info) else: event.event_cleanup(ch_info.gpio_chip, ch_info.channel) del _channel_configuration[ch_info.channel] # clean gpio config # clean up chip if ch_info.chip_fd: gpio_cdev.close_chip(ch_info.chip_fd) ch_info.chip_fd = None if ch_info.gpio_chip in _chip_fd: del _chip_fd[ch_info.gpio_chip] # clean up line if ch_info.line_handle: gpio_cdev.close_line(ch_info.line_handle) ch_info.line_handle = None def _cleanup_all(): global _gpio_mode for channel in list(_channel_configuration.keys()): ch_info = _channel_to_info(channel) _cleanup_one(ch_info) _gpio_mode = None # Function used to enable/disable warnings during setup and cleanup. # Param -> state is a bool def setwarnings(state): global _gpio_warnings _gpio_warnings = bool(state) # Function used to set the pin mumbering mode. Possible mode values are BOARD, # BCM, TEGRA_SOC and CVM def setmode(mode): global _gpio_mode, _channel_data # check if a different mode has been set if _gpio_mode and mode != _gpio_mode: raise ValueError("A different mode has already been set!") mode_map = { BOARD: 'BOARD', BCM: 'BCM', CVM: 'CVM', TEGRA_SOC: 'TEGRA_SOC', } # check if mode parameter is valid if mode not in mode_map: raise ValueError("An invalid mode was passed to setmode()!") _channel_data = _channel_data_by_mode[mode_map[mode]] _gpio_mode = mode # Function used to get the currently set pin numbering mode def getmode(): return _gpio_mode # Mutable class to represent a default function argument. # See https://stackoverflow.com/a/57628817/2767322 class _Default: def __init__(self, val): self.val = val # Function used to setup individual pins or lists/tuples of pins as # Input or Output. Param channels must an integer or list/tuple of integers, # direction must be IN or OUT, pull_up_down must be PUD_OFF, PUD_UP or # PUD_DOWN and is only valid when direction in IN, initial must be HIGH or LOW # and is only valid when direction is OUT def setup(channels, direction, pull_up_down=_Default(PUD_OFF), initial=None, consumer='Jetson-gpio'): if pull_up_down in setup.__defaults__: pull_up_down_explicit = False pull_up_down = pull_up_down.val else: pull_up_down_explicit = True ch_infos = _channels_to_infos(channels, need_gpio=True) # check direction is valid if direction != OUT and direction != IN: raise ValueError("An invalid direction was passed to setup()") # check if pullup/down is used with output if direction == OUT and pull_up_down != PUD_OFF: raise ValueError("pull_up_down parameter is not valid for outputs") # check if pullup/down value is specified and/or valid if pull_up_down_explicit: warnings.warn("Jetson.GPIO ignores setup()'s pull_up_down parameter") if (pull_up_down != PUD_OFF and pull_up_down != PUD_UP and pull_up_down != PUD_DOWN): raise ValueError("Invalid value for pull_up_down; should be one of" "PUD_OFF, PUD_UP or PUD_DOWN") for ch_info in ch_infos: if ch_info.channel in _channel_configuration: _cleanup_one(ch_info) if direction == OUT: initial = _make_iterable(initial, len(ch_infos)) if len(initial) != len(ch_infos): raise RuntimeError("Number of values != number of channels") for ch_info, init in zip(ch_infos, initial): _do_one_channel(ch_info, direction, init, consumer) else: for ch_info in ch_infos: _do_one_channel(ch_info, direction, initial, consumer) # Function used to cleanup channels at the end of the program. # The param channel can be an integer or list/tuple of integers specifying the # channels to be cleaned up. If no channel is provided, all channels are # cleaned def cleanup(channel=None): # warn if no channel is setup if _gpio_mode is None: if _gpio_warnings: warnings.warn("No channels have been set up yet - nothing to " "clean up! Try cleaning up at the end of your " "program instead!", RuntimeWarning) return # clean all channels if no channel param provided if channel is None: _cleanup_all() return ch_infos = _channels_to_infos(channel) for ch_info in ch_infos: if ch_info.channel in _channel_configuration: _cleanup_one(ch_info) # Function used to return the current value of the specified channel. # Function returns either HIGH or LOW def input(channel): ch_info = _channel_to_info(channel, need_gpio=True) cur_cfg = _app_channel_configuration(ch_info) if cur_cfg not in [IN, OUT]: raise RuntimeError("You must setup() the GPIO channel first") # _GPIOHANDLE_GET_LINE_VALUES_IOCTL, _CGpiohandleData return gpio_cdev.get_value(ch_info.line_handle) # Function used to set a value to a channel or list/tuple of channels. # Parameter channels must be an integer or list/tuple of integers. # Values must be either HIGH or LOW or list/tuple # of HIGH and LOW with the same length as the channels list/tuple def output(channels, values): ch_infos = _channels_to_infos(channels, need_gpio=True) values = _make_iterable(values, len(ch_infos)) if len(values) != len(ch_infos): raise RuntimeError("Number of values != number of channels") # check that channels have been set as output if any(_app_channel_configuration(ch_info) != OUT for ch_info in ch_infos): raise RuntimeError("The GPIO channel has not been set up as an " "OUTPUT") for ch_info, value in zip(ch_infos, values): gpio_cdev.set_value(ch_info.line_handle, value) # Function used to add threaded event detection for a specified gpio channel. # Param gpio must be an integer specifying the channel, edge must be RISING, # FALLING or BOTH. A callback function to be called when the event is detected # and an integer bounctime in milliseconds can be optionally provided. A optional # polltime in second can be provided to indicate the max time waiting for an edge. # Note that one channel only allows one event, which the duplicated event will # be ignored. def add_event_detect(channel, edge, callback=None, bouncetime=None, polltime=0.2): ch_info = _channel_to_info(channel, need_gpio=True) if (not callable(callback)) and callback is not None: raise TypeError("Callback Parameter must be callable") # channel must be setup as input if _app_channel_configuration(ch_info) != IN: raise RuntimeError("You must setup() the GPIO channel as an input " "first") # edge must be rising, falling or both if edge != RISING and edge != FALLING and edge != BOTH: raise ValueError("The edge must be set to RISING, FALLING, or BOTH") else: edge = gpio_cdev.GPIOEVENT_REQUEST_RISING_EDGE if edge == RISING else gpio_cdev.GPIOEVENT_REQUEST_FALLING_EDGE if edge == FALLING else gpio_cdev.GPIOEVENT_REQUEST_BOTH_EDGES # if bouncetime is provided, it must be int and greater than 0 if bouncetime is not None: if type(bouncetime) != int: raise TypeError("bouncetime must be an integer") elif bouncetime < 0: raise ValueError("bouncetime must be an integer greater than 0") if ch_info.line_handle: gpio_cdev.close_line(ch_info.line_handle) request = gpio_cdev.request_event(ch_info.line_offset, edge, ch_info.consumer) event.add_edge_detect(ch_info.chip_fd, ch_info.gpio_chip, channel, request, bouncetime, polltime) if callback is not None: event.add_edge_callback(ch_info.gpio_chip, channel, lambda: callback(channel)) # We should wait until the thread is up, which the device buffer cleaning takes time time.sleep(1) # Function used to remove event detection for channel # Timeout param for the max time to wait for thread (event detecion) to end def remove_event_detect(channel, timeout=0.5): ch_info = _channel_to_info(channel, need_gpio=True) event.remove_edge_detect(ch_info.gpio_chip, channel, timeout) # Function used to check if an event occurred on the specified channel. # Param channel must be an integer. # This function return True or False def event_detected(channel): ch_info = _channel_to_info(channel, need_gpio=True) # channel must be setup as input if _app_channel_configuration(ch_info) != IN: raise RuntimeError("You must setup() the GPIO channel as an " "input first") return event.edge_event_detected(ch_info.gpio_chip, channel) # Function used to add a callback function to channel, after it has been # registered for events using add_event_detect() def add_event_callback(channel, callback): ch_info = _channel_to_info(channel, need_gpio=True) if not callable(callback): raise TypeError("Parameter must be callable") if _app_channel_configuration(ch_info) != IN: raise RuntimeError("You must setup() the GPIO channel as an " "input first") if not event.gpio_event_added(ch_info.gpio_chip, channel): raise RuntimeError("Add event detection using add_event_detect first " "before adding a callback") event.add_edge_callback(ch_info.gpio_chip, channel, lambda: callback(channel)) # We should wait until the thread is up, which the device buffer cleaning takes time time.sleep(1) # Function used to wait for a edge event in blocking mode, it is also one-shoot. def wait_for_edge(channel, edge, bouncetime=None, timeout=None): ch_info = _channel_to_info(channel, need_gpio=True) # channel must be setup as input if _app_channel_configuration(ch_info) != IN: raise RuntimeError("You must setup() the GPIO channel as an input " "first") # edge provided must be rising, falling or both if edge != RISING and edge != FALLING and edge != BOTH: raise ValueError("The edge must be set to RISING, FALLING_EDGE " "or BOTH") else: edge = gpio_cdev.GPIOEVENT_REQUEST_RISING_EDGE if edge == RISING else gpio_cdev.GPIOEVENT_REQUEST_FALLING_EDGE if edge == FALLING else gpio_cdev.GPIOEVENT_REQUEST_BOTH_EDGES # if bouncetime is provided, it must be int and greater than 0 if bouncetime is not None: if type(bouncetime) != int: raise TypeError("bouncetime must be an integer") elif bouncetime < 0: raise ValueError("bouncetime must be an integer greater than 0") # if timeout is specified, it must be an int and greater than 0 if timeout is not None: if type(timeout) != int: raise TypeError("Timeout must be an integer") elif timeout < 0: raise ValueError("Timeout must greater than 0") if ch_info.line_handle: gpio_cdev.close_line(ch_info.line_handle) request = gpio_cdev.request_event(ch_info.line_offset, edge, ch_info.consumer) result = event.blocking_wait_for_edge(ch_info.chip_fd, ch_info.gpio_chip, channel, request, bouncetime, timeout) # If not error, result == channel. If timeout occurs while waiting, # result == None. If error occurs, result == -1 means channel is # registered for conflicting edge detection, result == -2 means an error # occurred while registering event or polling if not result: return None elif result == -1: raise RuntimeError("Conflicting edge detection event already exists " "for this GPIO channel") elif result == -2: raise RuntimeError("Error waiting for edge") return channel # Function used to check the currently set function of the channel specified. # Param channel must be an integers. The function returns either IN, OUT, # or UNKNOWN def gpio_function(channel): ch_info = _channel_to_info(channel) func = _app_channel_configuration(ch_info) if func is None: func = UNKNOWN return func class PWM(object): def __init__(self, channel, frequency_hz): self._ch_info = _channel_to_info(channel, need_pwm=True) app_cfg = _app_channel_configuration(self._ch_info) if app_cfg == HARD_PWM: raise ValueError("Can't create duplicate PWM objects") # Apps typically set up channels as GPIO before making them be PWM, # because RPi.GPIO does soft-PWM. We must undo the GPIO export to # allow HW PWM to run on the pin. if app_cfg in [IN, OUT]: cleanup(channel) if _gpio_warnings: sysfs_cfg = _sysfs_channel_configuration(self._ch_info) app_cfg = _app_channel_configuration(self._ch_info) # warn if channel has been setup external to current program if app_cfg is None and sysfs_cfg is not None: warnings.warn( "This channel is already in use, continuing anyway. " "Use GPIO.setwarnings(False) to disable warnings", RuntimeWarning) _export_pwm(self._ch_info) self._started = False _set_pwm_duty_cycle(self._ch_info, 0) # Anything that doesn't match new frequency_hz self._frequency_hz = -1 * frequency_hz self._reconfigure(frequency_hz, 0.0) _channel_configuration[channel] = HARD_PWM def __del__(self): if _channel_configuration.get(self._ch_info.channel, None) != HARD_PWM: # The user probably ran cleanup() on the channel already, so avoid # attempts to repeat the cleanup operations. return self.stop() _unexport_pwm(self._ch_info) del _channel_configuration[self._ch_info.channel] def start(self, duty_cycle_percent): self._reconfigure(self._frequency_hz, duty_cycle_percent, start=True) def ChangeFrequency(self, frequency_hz): self._reconfigure(frequency_hz, self._duty_cycle_percent) def ChangeDutyCycle(self, duty_cycle_percent): self._reconfigure(self._frequency_hz, duty_cycle_percent) def stop(self): if not self._started: return _disable_pwm(self._ch_info) def _reconfigure(self, frequency_hz, duty_cycle_percent, start=False): if duty_cycle_percent < 0.0 or duty_cycle_percent > 100.0: raise ValueError("") freq_change = start or (frequency_hz != self._frequency_hz) stop = self._started and freq_change if stop: self._started = False _disable_pwm(self._ch_info) if freq_change: self._frequency_hz = frequency_hz self._period_ns = int(1000000000.0 / frequency_hz) # Reset duty cycle period incase the previous duty # cycle is higher than the period _set_pwm_duty_cycle(self._ch_info, 0) _set_pwm_period(self._ch_info, self._period_ns) self._duty_cycle_percent = duty_cycle_percent self._duty_cycle_ns = int(self._period_ns * (duty_cycle_percent / 100.0)) _set_pwm_duty_cycle(self._ch_info, self._duty_cycle_ns) if stop or start: _enable_pwm(self._ch_info) self._started = True