mirror of
https://github.com/KevinMidboe/python-gpiozero.git
synced 2025-10-29 09:40:36 +00:00
337 lines
11 KiB
Python
337 lines
11 KiB
Python
from __future__ import (
|
|
unicode_literals,
|
|
absolute_import,
|
|
print_function,
|
|
division,
|
|
)
|
|
nstr = str
|
|
str = type('')
|
|
|
|
import io
|
|
import os
|
|
import mmap
|
|
import errno
|
|
import struct
|
|
from time import sleep
|
|
from threading import Thread, Event, Lock
|
|
from collections import Counter
|
|
|
|
from . import Pin, PINS_CLEANUP
|
|
from ..exc import (
|
|
PinInvalidPull,
|
|
PinInvalidEdges,
|
|
PinInvalidFunction,
|
|
PinFixedPull,
|
|
)
|
|
|
|
|
|
class GPIOMemory(object):
|
|
|
|
GPIO_BASE_OFFSET = 0x200000
|
|
PERI_BASE_OFFSET = {
|
|
'BCM2708': 0x20000000,
|
|
'BCM2835': 0x20000000,
|
|
'BCM2709': 0x3f000000,
|
|
'BCM2836': 0x3f000000,
|
|
}
|
|
|
|
# From BCM2835 data-sheet, p.91
|
|
GPFSEL_OFFSET = 0x00 >> 2
|
|
GPSET_OFFSET = 0x1c >> 2
|
|
GPCLR_OFFSET = 0x28 >> 2
|
|
GPLEV_OFFSET = 0x34 >> 2
|
|
GPEDS_OFFSET = 0x40 >> 2
|
|
GPREN_OFFSET = 0x4c >> 2
|
|
GPFEN_OFFSET = 0x58 >> 2
|
|
GPHEN_OFFSET = 0x64 >> 2
|
|
GPLEN_OFFSET = 0x70 >> 2
|
|
GPAREN_OFFSET = 0x7c >> 2
|
|
GPAFEN_OFFSET = 0x88 >> 2
|
|
GPPUD_OFFSET = 0x94 >> 2
|
|
GPPUDCLK_OFFSET = 0x98 >> 2
|
|
|
|
def __init__(self):
|
|
try:
|
|
self.fd = os.open('/dev/gpiomem', os.O_RDWR | os.O_SYNC)
|
|
except OSError:
|
|
try:
|
|
self.fd = os.open('/dev/mem', os.O_RDWR | os.O_SYNC)
|
|
except OSError:
|
|
raise IOError(
|
|
'unable to open /dev/gpiomem or /dev/mem; '
|
|
'upgrade your kernel or run as root')
|
|
else:
|
|
offset = self.peripheral_base() + self.GPIO_BASE_OFFSET
|
|
else:
|
|
offset = 0
|
|
self.mem = mmap.mmap(self.fd, 4096, offset=offset)
|
|
|
|
def close(self):
|
|
self.mem.close()
|
|
os.close(self.fd)
|
|
|
|
def peripheral_base(self):
|
|
try:
|
|
with io.open('/proc/device-tree/soc/ranges', 'rb') as f:
|
|
f.seek(4)
|
|
return struct.unpack(nstr('>L'), f.read(4))[0]
|
|
except IOError:
|
|
with io.open('/proc/cpuinfo', 'r') as f:
|
|
for line in f:
|
|
if line.startswith('Hardware'):
|
|
try:
|
|
return self.PERI_BASE_OFFSET[line.split(':')[1].strip()]
|
|
except KeyError:
|
|
raise IOError('unable to determine RPi revision')
|
|
raise IOError('unable to determine peripheral base')
|
|
|
|
def __getitem__(self, index):
|
|
return struct.unpack_from(nstr('<L'), self.mem, index * 4)[0]
|
|
|
|
def __setitem__(self, index, value):
|
|
struct.pack_into(nstr('<L'), self.mem, index * 4, value)
|
|
|
|
|
|
class GPIOFS(object):
|
|
|
|
GPIO_PATH = '/sys/class/gpio'
|
|
|
|
def __init__(self):
|
|
self._lock = Lock()
|
|
self._pin_refs = Counter()
|
|
|
|
def path(self, name):
|
|
return os.path.join(self.GPIO_PATH, name)
|
|
|
|
def export(self, pin):
|
|
with self._lock:
|
|
if self._pin_refs[pin] == 0:
|
|
# Set the count to 1 to indicate the GPIO is already exported
|
|
# (we'll correct this if we find it isn't, but this enables us
|
|
# to "leave the system in the state we found it")
|
|
self._pin_refs[pin] = 1
|
|
result = None
|
|
# Dirty hack to wait for udev to set permissions on
|
|
# gpioN/direction; there's no other way around this as there's
|
|
# no synchronous mechanism for setting permissions on sysfs
|
|
for i in range(10):
|
|
try:
|
|
result = io.open(self.path('gpio%d/value' % pin), 'w+b', buffering=0)
|
|
except IOError as e:
|
|
if e.errno == errno.ENOENT:
|
|
with io.open(self.path('export'), 'wb') as f:
|
|
f.write(str(pin).encode('ascii'))
|
|
# Pin wasn't exported, so correct the ref-count
|
|
self._pin_refs[pin] = 0
|
|
elif e.errno == errno.EACCES:
|
|
sleep(i / 100)
|
|
else:
|
|
raise
|
|
else:
|
|
break
|
|
if not result:
|
|
raise RuntimeError('failed to export pin %d' % pin)
|
|
else:
|
|
result = io.open(self.path('gpio%d/value' % pin), 'w+b', buffering=0)
|
|
self._pin_refs[pin] += 1
|
|
return result
|
|
|
|
def unexport(self, pin):
|
|
with self._lock:
|
|
self._pin_refs[pin] -= 1
|
|
if self._pin_refs[pin] == 0:
|
|
with io.open(self.path('unexport'), 'wb') as f:
|
|
f.write(str(pin).encode('ascii'))
|
|
|
|
|
|
class NativePin(Pin):
|
|
"""
|
|
Uses a built-in pure Python implementation to interface to the Pi's GPIO
|
|
pins. This is the default pin implementation if no third-party libraries
|
|
are discovered.
|
|
|
|
.. warning::
|
|
|
|
This implementation does *not* currently support PWM. Attempting to
|
|
use any class which requests PWM will raise an exception. This
|
|
implementation is also experimental; we make no guarantees it will
|
|
not eat your Pi for breakfast!
|
|
"""
|
|
|
|
_MEM = None
|
|
_PINS = {}
|
|
|
|
PULL_NAMES = {
|
|
0b00: 'floating',
|
|
0b01: 'down',
|
|
0b10: 'up',
|
|
0b11: 'reserved',
|
|
}
|
|
|
|
FUNCTION_NAMES = {
|
|
0b000: 'input',
|
|
0b001: 'output',
|
|
0b100: 'alt0',
|
|
0b101: 'alt1',
|
|
0b110: 'alt2',
|
|
0b111: 'alt3',
|
|
0b011: 'alt4',
|
|
0b010: 'alt5',
|
|
}
|
|
|
|
EDGE_NAMES = {
|
|
(True, True): 'both',
|
|
(True, False): 'rising',
|
|
(False, True): 'falling',
|
|
(False, False): 'none',
|
|
}
|
|
|
|
PULL_VALUES = {v: k for (k, v) in PULL_NAMES.items()}
|
|
FUNCTION_VALUES = {v: k for (k, v) in FUNCTION_NAMES.items()}
|
|
EDGE_VALUES = {v: k for (k, v) in EDGE_NAMES.items()}
|
|
|
|
def __new__(cls, number):
|
|
if not cls._PINS:
|
|
cls._MEM = GPIOMemory()
|
|
PINS_CLEANUP.append(cls._MEM.close)
|
|
if not (0 <= number < 54):
|
|
raise ValueError('invalid pin %d specified (must be 0..53)' % number)
|
|
try:
|
|
return cls._PINS[number]
|
|
except KeyError:
|
|
self = super(NativePin, cls).__new__(cls)
|
|
cls._PINS[number] = self
|
|
self._number = number
|
|
self._func_offset = self._MEM.GPFSEL_OFFSET + (number // 10)
|
|
self._func_shift = (number % 10) * 3
|
|
self._set_offset = self._MEM.GPSET_OFFSET + (number // 32)
|
|
self._set_shift = number % 32
|
|
self._clear_offset = self._MEM.GPCLR_OFFSET + (number // 32)
|
|
self._clear_shift = number % 32
|
|
self._level_offset = self._MEM.GPLEV_OFFSET + (number // 32)
|
|
self._level_shift = number % 32
|
|
self._pull_offset = self._MEM.GPPUDCLK_OFFSET + (number // 32)
|
|
self._pull_shift = number % 32
|
|
self._edge_offset = self._MEM.GPEDS_OFFSET + (number // 32)
|
|
self._edge_shift = number % 32
|
|
self._rising_offset = self._MEM.GPREN_OFFSET + (number // 32)
|
|
self._rising_shift = number % 32
|
|
self._falling_offset = self._MEM.GPFEN_OFFSET + (number // 32)
|
|
self._falling_shift = number % 32
|
|
self._when_changed = None
|
|
self._change_thread = None
|
|
self._change_event = Event()
|
|
self.function = 'input'
|
|
self.pull = 'up' if number in (2, 3) else 'floating'
|
|
self.bounce = None
|
|
self.edges = 'both'
|
|
return self
|
|
|
|
def __repr__(self):
|
|
return "GPIO%d" % self._number
|
|
|
|
@property
|
|
def number(self):
|
|
return self._number
|
|
|
|
def close(self):
|
|
self.when_changed = None
|
|
self.function = 'input'
|
|
|
|
def _get_function(self):
|
|
return self.FUNCTION_NAMES[(self._MEM[self._func_offset] >> self._func_shift) & 7]
|
|
|
|
def _set_function(self, value):
|
|
try:
|
|
value = self.FUNCTION_VALUES[value]
|
|
except KeyError:
|
|
raise PinInvalidFunction('invalid function "%s" for pin %r' % self)
|
|
self._MEM[self._func_offset] = (
|
|
self._MEM[self._func_offset]
|
|
& ~(7 << self._func_shift)
|
|
| (value << self._func_shift)
|
|
)
|
|
|
|
def _get_state(self):
|
|
return bool(self._MEM[self._level_offset] & (1 << self._level_shift))
|
|
|
|
def _set_state(self, value):
|
|
if value:
|
|
self._MEM[self._set_offset] = 1 << self._set_shift
|
|
else:
|
|
self._MEM[self._clear_offset] = 1 << self._clear_shift
|
|
|
|
def _get_pull(self):
|
|
return self.PULL_NAMES[self._pull]
|
|
|
|
def _set_pull(self, value):
|
|
if self.function != 'input':
|
|
raise PinFixedPull('cannot set pull on non-input pin %r' % self)
|
|
if value != 'up' and self._number in (2, 3):
|
|
raise PinFixedPull('%r has a physical pull-up resistor' % self)
|
|
try:
|
|
value = self.PULL_VALUES[value]
|
|
except KeyError:
|
|
raise PinInvalidPull('invalid pull direction "%s" for pin %r' % self)
|
|
self._MEM[self._MEM.GPPUD_OFFSET] = value
|
|
sleep(0.000000214)
|
|
self._MEM[self._pull_offset] = 1 << self._pull_shift
|
|
sleep(0.000000214)
|
|
self._MEM[self._MEM.GPPUD_OFFSET] = 0
|
|
self._MEM[self._pull_offset] = 0
|
|
self._pull = value
|
|
|
|
def _get_edges(self):
|
|
rising = bool(self._MEM[self._rising_offset] & (1 << self._rising_shift))
|
|
falling = bool(self._MEM[self._falling_offset] & (1 << self._falling_shift))
|
|
return self.EDGE_NAMES[(rising, falling)]
|
|
|
|
def _set_edges(self, value):
|
|
try:
|
|
rising, falling = self.EDGE_VALUES[value]
|
|
except KeyError:
|
|
raise PinInvalidEdges('invalid edge specification "%s" for pin %r' % self)
|
|
f = self.when_changed
|
|
self.when_changed = None
|
|
try:
|
|
self._MEM[self._rising_offset] = (
|
|
self._MEM[self._rising_offset]
|
|
& ~(1 << self._rising_shift)
|
|
| (rising << self._rising_shift)
|
|
)
|
|
self._MEM[self._falling_offset] = (
|
|
self._MEM[self._falling_offset]
|
|
& ~(1 << self._falling_shift)
|
|
| (falling << self._falling_shift)
|
|
)
|
|
finally:
|
|
self.when_changed = f
|
|
|
|
def _get_when_changed(self):
|
|
return self._when_changed
|
|
|
|
def _set_when_changed(self, value):
|
|
if self._when_changed is None and value is not None:
|
|
self._when_changed = value
|
|
self._change_thread = Thread(target=self._change_watch)
|
|
self._change_thread.daemon = True
|
|
self._change_event.clear()
|
|
self._change_thread.start()
|
|
elif self._when_changed is not None and value is None:
|
|
self._change_event.set()
|
|
self._change_thread.join()
|
|
self._change_thread = None
|
|
self._when_changed = None
|
|
else:
|
|
self._when_changed = value
|
|
|
|
def _change_watch(self):
|
|
offset = self._edge_offset
|
|
mask = 1 << self._edge_shift
|
|
self._MEM[offset] = mask # clear any existing detection bit
|
|
while not self._change_event.wait(0.001):
|
|
if self._MEM[offset] & mask:
|
|
self._MEM[offset] = mask
|
|
self._when_changed()
|
|
|