src/adafruit_blinka/microcontroller/generic_linux/sysfs_pin.py

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