src/adafruit_blinka/microcontroller/generic_linux/sysfs_pin.py

   1 # SPDX-FileCopyrightText: 2021 Melissa LeBlanc-Williams for Adafruit Industries
   2 #
   3 # SPDX-License-Identifier: MIT
   4 """
   5 Much code from https://github.com/vsergeev/python-periphery/blob/master/periphery/gpio.py
   6 Copyright (c) 2015-2019 vsergeev / Ivan (Vanya) A. Sergeev
   7 License: MIT
   8 """
   9
  10 import os
  11 import errno
  12 import time
  13
  14
  15 # pylint: disable=unnecessary-pass
  16 class GPIOError(IOError):
  17     """Base class for GPIO errors."""
  18
  19     pass
  20
  21
  22 # pylint: enable=unnecessary-pass
  23
  24
  25 class Pin:
  26     """SysFS GPIO Pin Class"""
  27
  28     # Number of retries to check for GPIO export or direction write on open
  29     GPIO_OPEN_RETRIES = 10
  30     # Delay between check for GPIO export or direction write on open (100ms)
  31     GPIO_OPEN_DELAY = 0.1
  32
  33     IN = "in"
  34     OUT = "out"
  35     LOW = 0
  36     HIGH = 1
  37     PULL_NONE = 0
  38     PULL_UP = 1
  39     PULL_DOWN = 2
  40
  41     id = None
  42     _value = LOW
  43     _mode = IN
  44
  45     # Sysfs paths
  46     _sysfs_path = "/sys/class/gpio/"
  47     _channel_path = "gpiochip{}"
  48
  49     # Channel paths
  50     _export_path = "export"
  51     _unexport_path = "unexport"
  52     _pin_path = "gpio{}"
  53
  54     def __init__(self, pin_id):
  55         """Instantiate a Pin object and open the sysfs GPIO with the specified
  56         pin number.
  57
  58         Args:
  59             pin_id (int): GPIO pin number.
  60
  61         Returns:
  62             SysfsGPIO: GPIO object.
  63
  64         Raises:
  65             GPIOError: if an I/O or OS error occurs.
  66             TypeError: if `line` or `direction`  types are invalid.
  67             ValueError: if `direction` value is invalid.
  68             TimeoutError: if waiting for GPIO export times out.
  69
  70         """
  71
  72         if not isinstance(pin_id, int):
  73             raise TypeError("Invalid Pin ID, should be integer.")
  74
  75         self.id = pin_id
  76         self._fd = None
  77         self._line = None
  78         self._path = None
  79
  80     def __del__(self):
  81         self._close()
  82
  83     def __enter__(self):
  84         return self
  85
  86     def __exit__(self, t, value, traceback):
  87         self._close()
  88
  89     def init(self, mode=IN, pull=None):
  90         """Initialize the Pin"""
  91         if mode is not None:
  92             if mode == self.IN:
  93                 self._mode = self.IN
  94                 self._open(self.IN)
  95             elif mode == self.OUT:
  96                 self._mode = self.OUT
  97                 self._open(self.OUT)
  98             else:
  99                 raise RuntimeError("Invalid mode for pin: %s" % self.id)
 100
 101             if pull is not None:
 102                 if pull == self.PULL_UP:
 103                     raise NotImplementedError(
 104                         "Internal pullups not supported in periphery, "
 105                         "use physical resistor instead!"
 106                     )
 107                 if pull == self.PULL_DOWN:
 108                     raise NotImplementedError(
 109                         "Internal pulldowns not supported in periphery, "
 110                         "use physical resistor instead!"
 111                     )
 112                 raise RuntimeError("Invalid pull for pin: %s" % self.id)
 113
 114     def value(self, val=None):
 115         """Set or return the Pin Value"""
 116         if val is not None:
 117             if val == self.LOW:
 118                 self._value = val
 119                 self._write(False)
 120                 return None
 121             if val == self.HIGH:
 122                 self._value = val
 123                 self._write(True)
 124                 return None
 125             raise RuntimeError("Invalid value for pin")
 126         return self.HIGH if self._read() else self.LOW
 127
 128     # pylint: disable=too-many-branches
 129     def _open(self, direction):
 130         if not isinstance(direction, str):
 131             raise TypeError("Invalid direction type, should be string.")
 132         if direction.lower() not in ["in", "out", "high", "low"]:
 133             raise ValueError('Invalid direction, can be: "in", "out", "high", "low".')
 134         gpio_path = "/sys/class/gpio/gpio{:d}".format(self.id)
 135
 136         if not os.path.isdir(gpio_path):
 137             # Export the line
 138             try:
 139                 with open("/sys/class/gpio/export", "w", encoding="utf-8") as f_export:
 140                     f_export.write("{:d}\n".format(self.id))
 141             except IOError as e:
 142                 raise GPIOError(e.errno, "Exporting GPIO: " + e.strerror) from IOError
 143
 144             # Loop until GPIO is exported
 145             exported = False
 146             for i in range(self.GPIO_OPEN_RETRIES):
 147                 if os.path.isdir(gpio_path):
 148                     exported = True
 149                     break
 150
 151                 time.sleep(self.GPIO_OPEN_DELAY)
 152
 153             if not exported:
 154                 raise TimeoutError(
 155                     'Exporting GPIO: waiting for "{:s}" timed out'.format(gpio_path)
 156                 )
 157
 158             # Write direction, looping in case of EACCES errors due to delayed udev
 159             # permission rule application after export
 160             for i in range(self.GPIO_OPEN_RETRIES):
 161                 try:
 162                     with open(
 163                         os.path.join(gpio_path, "direction"), "w", encoding="utf-8"
 164                     ) as f_direction:
 165                         f_direction.write(direction.lower() + "\n")
 166                     break
 167                 except IOError as e:
 168                     if e.errno != errno.EACCES or (
 169                         e.errno == errno.EACCES and i == self.GPIO_OPEN_RETRIES - 1
 170                     ):
 171                         raise GPIOError(
 172                             e.errno, "Setting GPIO direction: " + e.strerror
 173                         ) from IOError
 174
 175                 time.sleep(self.GPIO_OPEN_DELAY)
 176         else:
 177             # Write direction
 178             try:
 179                 with open(
 180                     os.path.join(gpio_path, "direction"), "w", encoding="utf-8"
 181                 ) as f_direction:
 182                     f_direction.write(direction.lower() + "\n")
 183             except IOError as e:
 184                 raise GPIOError(
 185                     e.errno, "Setting GPIO direction: " + e.strerror
 186                 ) from IOError
 187
 188         # Open value
 189         try:
 190             self._fd = os.open(os.path.join(gpio_path, "value"), os.O_RDWR)
 191         except OSError as e:
 192             raise GPIOError(e.errno, "Opening GPIO: " + e.strerror) from OSError
 193
 194         self._path = gpio_path
 195
 196     # pylint: enable=too-many-branches
 197
 198     def _close(self):
 199         if self._fd is None:
 200             return
 201
 202         try:
 203             os.close(self._fd)
 204         except OSError as e:
 205             raise GPIOError(e.errno, "Closing GPIO: " + e.strerror) from OSError
 206
 207         self._fd = None
 208
 209         # Unexport the line
 210         try:
 211             unexport_fd = os.open("/sys/class/gpio/unexport", os.O_WRONLY)
 212             os.write(unexport_fd, "{:d}\n".format(self.id).encode())
 213             os.close(unexport_fd)
 214         except OSError as e:
 215             raise GPIOError(e.errno, "Unexporting GPIO: " + e.strerror) from OSError
 216
 217     def _read(self):
 218         # Read value
 219         try:
 220             buf = os.read(self._fd, 2)
 221         except OSError as e:
 222             raise GPIOError(e.errno, "Reading GPIO: " + e.strerror) from OSError
 223
 224         # Rewind
 225         try:
 226             os.lseek(self._fd, 0, os.SEEK_SET)
 227         except OSError as e:
 228             raise GPIOError(e.errno, "Rewinding GPIO: " + e.strerror) from OSError
 229
 230         if buf[0] == b"0"[0]:
 231             return False
 232         if buf[0] == b"1"[0]:
 233             return True
 234
 235         raise GPIOError(None, "Unknown GPIO value: {}".format(buf))
 236
 237     def _write(self, value):
 238         if not isinstance(value, bool):
 239             raise TypeError("Invalid value type, should be bool.")
 240
 241         # Write value
 242         try:
 243             if value:
 244                 os.write(self._fd, b"1\n")
 245             else:
 246                 os.write(self._fd, b"0\n")
 247         except OSError as e:
 248             raise GPIOError(e.errno, "Writing GPIO: " + e.strerror) from OSError
 249
 250         # Rewind
 251         try:
 252             os.lseek(self._fd, 0, os.SEEK_SET)
 253         except OSError as e:
 254             raise GPIOError(e.errno, "Rewinding GPIO: " + e.strerror) from OSError
 255
 256     @property
 257     def chip_name(self):
 258         """Return the Chip Name"""
 259         gpio_path = os.path.join(self._path, "device")
 260
 261         gpiochip_path = os.readlink(gpio_path)
 262
 263         if "/" not in gpiochip_path:
 264             raise GPIOError(
 265                 None,
 266                 'Reading gpiochip name: invalid device symlink "{:s}"'.format(
 267                     gpiochip_path
 268                 ),
 269             )
 270
 271         return gpiochip_path.split("/")[-1]
 272
 273     @property
 274     def chip_label(self):
 275         """Return the Chip Label"""
 276         gpio_path = "/sys/class/gpio/{:s}/label".format(self.chip_name)
 277
 278         try:
 279             with open(gpio_path, "r", encoding="utf-8") as f_label:
 280                 label = f_label.read()
 281         except (GPIOError, IOError) as e:
 282             if isinstance(e, IOError):
 283                 raise GPIOError(
 284                     e.errno, "Reading gpiochip label: " + e.strerror
 285                 ) from IOError
 286
 287             raise GPIOError(
 288                 None, "Reading gpiochip label: " + e.strerror
 289             ) from GPIOError
 290
 291         return label.strip()
 292
 293     # Mutable properties
 294
 295     def _get_direction(self):
 296         # Read direction
 297         try:
 298             with open(
 299                 os.path.join(self._path, "direction"), "r", encoding="utf-8"
 300             ) as f_direction:
 301                 direction = f_direction.read()
 302         except IOError as e:
 303             raise GPIOError(
 304                 e.errno, "Getting GPIO direction: " + e.strerror
 305             ) from IOError
 306
 307         return direction.strip()
 308
 309     def _set_direction(self, direction):
 310         if not isinstance(direction, str):
 311             raise TypeError("Invalid direction type, should be string.")
 312         if direction.lower() not in ["in", "out", "high", "low"]:
 313             raise ValueError('Invalid direction, can be: "in", "out", "high", "low".')
 314
 315         # Write direction
 316         try:
 317             with open(
 318                 os.path.join(self._path, "direction"), "w", encoding="utf-8"
 319             ) as f_direction:
 320                 f_direction.write(direction.lower() + "\n")
 321         except IOError as e:
 322             raise GPIOError(
 323                 e.errno, "Setting GPIO direction: " + e.strerror
 324             ) from IOError
 325
 326     direction = property(_get_direction, _set_direction)
 327
 328     def __str__(self):
 329         try:
 330             str_direction = self.direction
 331         except GPIOError:
 332             str_direction = "<error>"
 333
 334         try:
 335             str_chip_name = self.chip_name
 336         except GPIOError:
 337             str_chip_name = "<error>"
 338
 339         try:
 340             str_chip_label = self.chip_label
 341         except GPIOError:
 342             str_chip_label = "<error>"
 343
 344         output = "Pin {:d} (dev={:s}, fd={:d}, dir={:s}, chip_name='{:s}', chip_label='{:s}')"
 345         return output.format(
 346             self.id, self._path, self._fd, str_direction, str_chip_name, str_chip_label
 347         )