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