h3n3/darkwater_python_640
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Got rid of non-python files

Browse Source
This commit is contained in:
shrkey
2016-08-13 19:23:17 +01:00
parent 9a20c84464
commit 4bcc4ff404
18 changed files with 1238 additions and 8129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
darkwater_640/Adafruit_I2C.py Normal file
View File

@@ -0,0 +1,161 @@
#!/usr/bin/python
import re
import smbus
# ===========================================================================
# Adafruit_I2C Class
# ===========================================================================
class Adafruit_I2C(object):
@staticmethod
def getPiRevision():
"Gets the version number of the Raspberry Pi board"
# Revision list available at: http://elinux.org/RPi_HardwareHistory#Board_Revision_History
try:
with open('/proc/cpuinfo', 'r') as infile:
for line in infile:
# Match a line of the form "Revision : 0002" while ignoring extra
# info in front of the revsion (like 1000 when the Pi was over-volted).
match = re.match('Revision\s+:\s+.*(\w{4})$', line)
if match and match.group(1) in ['0000', '0002', '0003']:
# Return revision 1 if revision ends with 0000, 0002 or 0003.
return 1
elif match:
# Assume revision 2 if revision ends with any other 4 chars.
return 2
# Couldn't find the revision, assume revision 0 like older code for compatibility.
return 0
except:
return 0
@staticmethod
def getPiI2CBusNumber():
# Gets the I2C bus number /dev/i2c#
return 1 if Adafruit_I2C.getPiRevision() > 1 else 0
def __init__(self, address, busnum=-1, debug=False):
self.address = address
# By default, the correct I2C bus is auto-detected using /proc/cpuinfo
# Alternatively, you can hard-code the bus version below:
# self.bus = smbus.SMBus(0); # Force I2C0 (early 256MB Pi's)
# self.bus = smbus.SMBus(1); # Force I2C1 (512MB Pi's)
self.bus = smbus.SMBus(busnum if busnum >= 0 else Adafruit_I2C.getPiI2CBusNumber())
self.debug = debug
def reverseByteOrder(self, data):
"Reverses the byte order of an int (16-bit) or long (32-bit) value"
# Courtesy Vishal Sapre
byteCount = len(hex(data)[2:].replace('L','')[::2])
val = 0
for i in range(byteCount):
val = (val << 8) | (data & 0xff)
data >>= 8
return val
def errMsg(self):
print "Error accessing 0x%02X: Check your I2C address" % self.address
return -1
def write8(self, reg, value):
"Writes an 8-bit value to the specified register/address"
try:
self.bus.write_byte_data(self.address, reg, value)
if self.debug:
print "I2C: Wrote 0x%02X to register 0x%02X" % (value, reg)
except IOError, err:
return self.errMsg()
def write16(self, reg, value):
"Writes a 16-bit value to the specified register/address pair"
try:
self.bus.write_word_data(self.address, reg, value)
if self.debug:
print ("I2C: Wrote 0x%02X to register pair 0x%02X,0x%02X" %
(value, reg, reg+1))
except IOError, err:
return self.errMsg()
def writeRaw8(self, value):
"Writes an 8-bit value on the bus"
try:
self.bus.write_byte(self.address, value)
if self.debug:
print "I2C: Wrote 0x%02X" % value
except IOError, err:
return self.errMsg()
def writeList(self, reg, list):
"Writes an array of bytes using I2C format"
try:
if self.debug:
print "I2C: Writing list to register 0x%02X:" % reg
print list
self.bus.write_i2c_block_data(self.address, reg, list)
except IOError, err:
return self.errMsg()
def readList(self, reg, length):
"Read a list of bytes from the I2C device"
try:
results = self.bus.read_i2c_block_data(self.address, reg, length)
if self.debug:
print ("I2C: Device 0x%02X returned the following from reg 0x%02X" %
(self.address, reg))
print results
return results
except IOError, err:
return self.errMsg()
def readU8(self, reg):
"Read an unsigned byte from the I2C device"
try:
result = self.bus.read_byte_data(self.address, reg)
if self.debug:
print ("I2C: Device 0x%02X returned 0x%02X from reg 0x%02X" %
(self.address, result & 0xFF, reg))
return result
except IOError, err:
return self.errMsg()
def readS8(self, reg):
"Reads a signed byte from the I2C device"
try:
result = self.bus.read_byte_data(self.address, reg)
if result > 127: result -= 256
if self.debug:
print ("I2C: Device 0x%02X returned 0x%02X from reg 0x%02X" %
(self.address, result & 0xFF, reg))
return result
except IOError, err:
return self.errMsg()
def readU16(self, reg, little_endian=True):
"Reads an unsigned 16-bit value from the I2C device"
try:
result = self.bus.read_word_data(self.address,reg)
# Swap bytes if using big endian because read_word_data assumes little
# endian on ARM (little endian) systems.
if not little_endian:
result = ((result << 8) & 0xFF00) + (result >> 8)
if (self.debug):
print "I2C: Device 0x%02X returned 0x%04X from reg 0x%02X" % (self.address, result & 0xFFFF, reg)
return result
except IOError, err:
return self.errMsg()
def readS16(self, reg, little_endian=True):
"Reads a signed 16-bit value from the I2C device"
try:
result = self.readU16(reg,little_endian)
if result > 32767: result -= 65536
return result
except IOError, err:
return self.errMsg()
if __name__ == '__main__':
try:
bus = Adafruit_I2C(address=0)
print "Default I2C bus is accessible"
except:
print "Error accessing default I2C bus"

143
darkwater_640/Adafruit_PWM_Servo_Driver.py Normal file
View File

@@ -0,0 +1,143 @@
#!/usr/bin/python
import time
import math
from Adafruit_I2C import Adafruit_I2C
# ============================================================================
# Adafruit PCA9685 16-Channel PWM Servo Driver
# ============================================================================
class PWM :
# Registers/etc.
__MODE1 = 0x00
__MODE2 = 0x01
__SUBADR1 = 0x02
__SUBADR2 = 0x03
__SUBADR3 = 0x04
__PRESCALE = 0xFE
__LED0_ON_L = 0x06
__LED0_ON_H = 0x07
__LED0_OFF_L = 0x08
__LED0_OFF_H = 0x09
__ALL_LED_ON_L = 0xFA
__ALL_LED_ON_H = 0xFB
__ALL_LED_OFF_L = 0xFC
__ALL_LED_OFF_H = 0xFD
# Bits
__RESTART = 0x80
__SLEEP = 0x10
__ALLCALL = 0x01
__INVRT = 0x10
__OUTDRV = 0x04
general_call_i2c = Adafruit_I2C(0x00)
@classmethod
def softwareReset(cls):
"Sends a software reset (SWRST) command to all the servo drivers on the bus"
cls.general_call_i2c.writeRaw8(0x06) # SWRST
def __init__(self, address=0x40, debug=False):
self.i2c = Adafruit_I2C(address)
self.i2c.debug = debug
self.address = address
self.debug = debug
if (self.debug):
print "Reseting PCA9685 MODE1 (without SLEEP) and MODE2"
self.setAllPWM(0, 0)
self.i2c.write8(self.__MODE2, self.__OUTDRV)
self.i2c.write8(self.__MODE1, self.__ALLCALL)
time.sleep(0.005) # wait for oscillator
mode1 = self.i2c.readU8(self.__MODE1)
mode1 = mode1 & ~self.__SLEEP # wake up (reset sleep)
self.i2c.write8(self.__MODE1, mode1)
time.sleep(0.005) # wait for oscillator
def setPWMFreq(self, freq, correctionFactor=1.0):
"Sets the PWM frequency"
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq)
prescaleval -= 1.0
if (self.debug):
print "Setting PWM frequency to %d Hz" % freq
print "Estimated pre-scale: %d" % prescaleval
prescale = round(prescaleval * correctionFactor + 0.5)
if (self.debug):
print "Final pre-scale: %d" % prescale
oldmode = self.i2c.readU8(self.__MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self.i2c.write8(self.__MODE1, newmode) # go to sleep
self.i2c.write8(self.__PRESCALE, int(math.floor(prescale)))
self.i2c.write8(self.__MODE1, oldmode)
time.sleep(0.005)
self.i2c.write8(self.__MODE1, oldmode | 0x80)
def setPWMFreqMin(self, freq, correctionFactor=1.0):
"Sets the PWM frequency"
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq)
prescaleval -= 1.0
if (self.debug):
print "Setting PWM frequency to %d Hz" % freq
print "Estimated pre-scale: %d" % prescaleval
prescale = math.floor(prescaleval * correctionFactor + 0.5)
if (self.debug):
print "Final pre-scale: %d" % prescale
oldmode = self.i2c.readU8(self.__MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self.i2c.write8(self.__MODE1, newmode) # go to sleep
self.i2c.write8(self.__PRESCALE, int(math.floor(prescale)))
self.i2c.write8(self.__MODE1, oldmode)
time.sleep(0.005)
self.i2c.write8(self.__MODE1, oldmode | 0x80)
def setPWMFreqMax(self, freq, correctionFactor=1.0):
"Sets the PWM frequency"
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq)
prescaleval -= 1.0
if (self.debug):
print "Setting PWM frequency to %d Hz" % freq
print "Estimated pre-scale: %d" % prescaleval
prescale = math.ceil(prescaleval * correctionFactor + 0.5)
if (self.debug):
print "Final pre-scale: %d" % prescale
oldmode = self.i2c.readU8(self.__MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self.i2c.write8(self.__MODE1, newmode) # go to sleep
self.i2c.write8(self.__PRESCALE, int(math.floor(prescale)))
self.i2c.write8(self.__MODE1, oldmode)
time.sleep(0.005)
self.i2c.write8(self.__MODE1, oldmode | 0x80)
def getPWMFreq(self):
prescale = self.i2c.readU8(self.__PRESCALE)
calcfreq = 25000000.0 / 4096.0 / ( float(prescale) + 1 )
if (self.debug):
print "Got pre-scale: %d" % prescale
print "Calculated Frequency: %d" % calcfreq
return calcfreq
def setPWM(self, channel, on, off):
"Sets a single PWM channel"
self.i2c.write8(self.__LED0_ON_L+4*channel, on & 0xFF)
self.i2c.write8(self.__LED0_ON_H+4*channel, on >> 8)
self.i2c.write8(self.__LED0_OFF_L+4*channel, off & 0xFF)
self.i2c.write8(self.__LED0_OFF_H+4*channel, off >> 8)
def setAllPWM(self, on, off):
"Sets a all PWM channels"
self.i2c.write8(self.__ALL_LED_ON_L, on & 0xFF)
self.i2c.write8(self.__ALL_LED_ON_H, on >> 8)
self.i2c.write8(self.__ALL_LED_OFF_L, off & 0xFF)
self.i2c.write8(self.__ALL_LED_OFF_H, off >> 8)

426
darkwater_640/GPIO.py Normal file
View File

@@ -0,0 +1,426 @@
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
#
# 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.
import Adafruit_GPIO.Platform as Platform
OUT = 0
IN = 1
HIGH = True
LOW = False
RISING = 1
FALLING = 2
BOTH = 3
PUD_OFF = 0
PUD_DOWN = 1
PUD_UP = 2
class BaseGPIO(object):
"""Base class for implementing simple digital IO for a platform.
Implementors are expected to subclass from this and provide an implementation
of the setup, output, and input functions."""
def setup(self, pin, mode, pull_up_down=PUD_OFF):
"""Set the input or output mode for a specified pin. Mode should be
either OUT or IN."""
raise NotImplementedError
def output(self, pin, value):
"""Set the specified pin the provided high/low value. Value should be
either HIGH/LOW or a boolean (true = high)."""
raise NotImplementedError
def input(self, pin):
"""Read the specified pin and return HIGH/true if the pin is pulled high,
or LOW/false if pulled low."""
raise NotImplementedError
def set_high(self, pin):
"""Set the specified pin HIGH."""
self.output(pin, HIGH)
def set_low(self, pin):
"""Set the specified pin LOW."""
self.output(pin, LOW)
def is_high(self, pin):
"""Return true if the specified pin is pulled high."""
return self.input(pin) == HIGH
def is_low(self, pin):
"""Return true if the specified pin is pulled low."""
return self.input(pin) == LOW
# Basic implementation of multiple pin methods just loops through pins and
# processes each one individually. This is not optimal, but derived classes can
# provide a more optimal implementation that deals with groups of pins
# simultaneously.
# See MCP230xx or PCF8574 classes for examples of optimized implementations.
def output_pins(self, pins):
"""Set multiple pins high or low at once. Pins should be a dict of pin
name to pin value (HIGH/True for 1, LOW/False for 0). All provided pins
will be set to the given values.
"""
# General implementation just loops through pins and writes them out
# manually. This is not optimized, but subclasses can choose to implement
# a more optimal batch output implementation. See the MCP230xx class for
# example of optimized implementation.
for pin, value in iter(pins.items()):
self.output(pin, value)
def setup_pins(self, pins):
"""Setup multiple pins as inputs or outputs at once. Pins should be a
dict of pin name to pin type (IN or OUT).
"""
# General implementation that can be optimized by derived classes.
for pin, value in iter(pins.items()):
self.setup(pin, value)
def input_pins(self, pins):
"""Read multiple pins specified in the given list and return list of pin values
GPIO.HIGH/True if the pin is pulled high, or GPIO.LOW/False if pulled low.
"""
# General implementation that can be optimized by derived classes.
return [self.input(pin) for pin in pins]
def add_event_detect(self, pin, edge):
"""Enable edge detection events for a particular GPIO channel. Pin
should be type IN. Edge must be RISING, FALLING or BOTH.
"""
raise NotImplementedError
def remove_event_detect(self, pin):
"""Remove edge detection for a particular GPIO channel. Pin should be
type IN.
"""
raise NotImplementedError
def add_event_callback(self, pin, callback):
"""Add a callback for an event already defined using add_event_detect().
Pin should be type IN.
"""
raise NotImplementedError
def event_detected(self, pin):
"""Returns True if an edge has occured on a given GPIO. You need to
enable edge detection using add_event_detect() first. Pin should be
type IN.
"""
raise NotImplementedError
def wait_for_edge(self, pin, edge):
"""Wait for an edge. Pin should be type IN. Edge must be RISING,
FALLING or BOTH."""
raise NotImplementedError
def cleanup(self, pin=None):
"""Clean up GPIO event detection for specific pin, or all pins if none
is specified.
"""
raise NotImplementedError
# helper functions useful to derived classes
def _validate_pin(self, pin):
# Raise an exception if pin is outside the range of allowed values.
if pin < 0 or pin >= self.NUM_GPIO:
raise ValueError('Invalid GPIO value, must be between 0 and {0}.'.format(self.NUM_GPIO))
def _bit2(self, src, bit, val):
bit = 1 << bit
return (src | bit) if val else (src & ~bit)
class RPiGPIOAdapter(BaseGPIO):
"""GPIO implementation for the Raspberry Pi using the RPi.GPIO library."""
def __init__(self, rpi_gpio, mode=None):
self.rpi_gpio = rpi_gpio
# Suppress warnings about GPIO in use.
rpi_gpio.setwarnings(False)
# Setup board pin mode.
if mode == rpi_gpio.BOARD or mode == rpi_gpio.BCM:
rpi_gpio.setmode(mode)
elif mode is not None:
raise ValueError('Unexpected value for mode. Must be BOARD or BCM.')
else:
# Default to BCM numbering if not told otherwise.
rpi_gpio.setmode(rpi_gpio.BCM)
# Define mapping of Adafruit GPIO library constants to RPi.GPIO constants.
self._dir_mapping = { OUT: rpi_gpio.OUT,
IN: rpi_gpio.IN }
self._pud_mapping = { PUD_OFF: rpi_gpio.PUD_OFF,
PUD_DOWN: rpi_gpio.PUD_DOWN,
PUD_UP: rpi_gpio.PUD_UP }
self._edge_mapping = { RISING: rpi_gpio.RISING,
FALLING: rpi_gpio.FALLING,
BOTH: rpi_gpio.BOTH }
def setup(self, pin, mode, pull_up_down=PUD_OFF):
"""Set the input or output mode for a specified pin. Mode should be
either OUTPUT or INPUT.
"""
self.rpi_gpio.setup(pin, self._dir_mapping[mode],
pull_up_down=self._pud_mapping[pull_up_down])
def output(self, pin, value):
"""Set the specified pin the provided high/low value. Value should be
either HIGH/LOW or a boolean (true = high).
"""
self.rpi_gpio.output(pin, value)
def input(self, pin):
"""Read the specified pin and return HIGH/true if the pin is pulled high,
or LOW/false if pulled low.
"""
return self.rpi_gpio.input(pin)
def input_pins(self, pins):
"""Read multiple pins specified in the given list and return list of pin values
GPIO.HIGH/True if the pin is pulled high, or GPIO.LOW/False if pulled low.
"""
# maybe rpi has a mass read... it would be more efficient to use it if it exists
return [self.rpi_gpio.input(pin) for pin in pins]
def add_event_detect(self, pin, edge, callback=None, bouncetime=-1):
"""Enable edge detection events for a particular GPIO channel. Pin
should be type IN. Edge must be RISING, FALLING or BOTH. Callback is a
function for the event. Bouncetime is switch bounce timeout in ms for
callback
"""
kwargs = {}
if callback:
kwargs['callback']=callback
if bouncetime > 0:
kwargs['bouncetime']=bouncetime
self.rpi_gpio.add_event_detect(pin, self._edge_mapping[edge], **kwargs)
def remove_event_detect(self, pin):
"""Remove edge detection for a particular GPIO channel. Pin should be
type IN.
"""
self.rpi_gpio.remove_event_detect(pin)
def add_event_callback(self, pin, callback):
"""Add a callback for an event already defined using add_event_detect().
Pin should be type IN.
"""
self.rpi_gpio.add_event_callback(pin, callback)
def event_detected(self, pin):
"""Returns True if an edge has occured on a given GPIO. You need to
enable edge detection using add_event_detect() first. Pin should be
type IN.
"""
return self.rpi_gpio.event_detected(pin)
def wait_for_edge(self, pin, edge):
"""Wait for an edge. Pin should be type IN. Edge must be RISING,
FALLING or BOTH.
"""
self.rpi_gpio.wait_for_edge(pin, self._edge_mapping[edge])
def cleanup(self, pin=None):
"""Clean up GPIO event detection for specific pin, or all pins if none
is specified.
"""
if pin is None:
self.rpi_gpio.cleanup()
else:
self.rpi_gpio.cleanup(pin)
class AdafruitBBIOAdapter(BaseGPIO):
"""GPIO implementation for the Beaglebone Black using the Adafruit_BBIO
library.
"""
def __init__(self, bbio_gpio):
self.bbio_gpio = bbio_gpio
# Define mapping of Adafruit GPIO library constants to RPi.GPIO constants.
self._dir_mapping = { OUT: bbio_gpio.OUT,
IN: bbio_gpio.IN }
self._pud_mapping = { PUD_OFF: bbio_gpio.PUD_OFF,
PUD_DOWN: bbio_gpio.PUD_DOWN,
PUD_UP: bbio_gpio.PUD_UP }
self._edge_mapping = { RISING: bbio_gpio.RISING,
FALLING: bbio_gpio.FALLING,
BOTH: bbio_gpio.BOTH }
def setup(self, pin, mode, pull_up_down=PUD_OFF):
"""Set the input or output mode for a specified pin. Mode should be
either OUTPUT or INPUT.
"""
self.bbio_gpio.setup(pin, self._dir_mapping[mode],
pull_up_down=self._pud_mapping[pull_up_down])
def output(self, pin, value):
"""Set the specified pin the provided high/low value. Value should be
either HIGH/LOW or a boolean (true = high).
"""
self.bbio_gpio.output(pin, value)
def input(self, pin):
"""Read the specified pin and return HIGH/true if the pin is pulled high,
or LOW/false if pulled low.
"""
return self.bbio_gpio.input(pin)
def input_pins(self, pins):
"""Read multiple pins specified in the given list and return list of pin values
GPIO.HIGH/True if the pin is pulled high, or GPIO.LOW/False if pulled low.
"""
# maybe bbb has a mass read... it would be more efficient to use it if it exists
return [self.bbio_gpio.input(pin) for pin in pins]
def add_event_detect(self, pin, edge, callback=None, bouncetime=-1):
"""Enable edge detection events for a particular GPIO channel. Pin
should be type IN. Edge must be RISING, FALLING or BOTH. Callback is a
function for the event. Bouncetime is switch bounce timeout in ms for
callback
"""
kwargs = {}
if callback:
kwargs['callback']=callback
if bouncetime > 0:
kwargs['bouncetime']=bouncetime
self.bbio_gpio.add_event_detect(pin, self._edge_mapping[edge], **kwargs)
def remove_event_detect(self, pin):
"""Remove edge detection for a particular GPIO channel. Pin should be
type IN.
"""
self.bbio_gpio.remove_event_detect(pin)
def add_event_callback(self, pin, callback, bouncetime=-1):
"""Add a callback for an event already defined using add_event_detect().
Pin should be type IN. Bouncetime is switch bounce timeout in ms for
callback
"""
kwargs = {}
if bouncetime > 0:
kwargs['bouncetime']=bouncetime
self.bbio_gpio.add_event_callback(pin, callback, **kwargs)
def event_detected(self, pin):
"""Returns True if an edge has occured on a given GPIO. You need to
enable edge detection using add_event_detect() first. Pin should be
type IN.
"""
return self.bbio_gpio.event_detected(pin)
def wait_for_edge(self, pin, edge):
"""Wait for an edge. Pin should be type IN. Edge must be RISING,
FALLING or BOTH.
"""
self.bbio_gpio.wait_for_edge(pin, self._edge_mapping[edge])
def cleanup(self, pin=None):
"""Clean up GPIO event detection for specific pin, or all pins if none
is specified.
"""
if pin is None:
self.bbio_gpio.cleanup()
else:
self.bbio_gpio.cleanup(pin)
class AdafruitMinnowAdapter(BaseGPIO):
"""GPIO implementation for the Minnowboard + MAX using the mraa library"""
def __init__(self,mraa_gpio):
self.mraa_gpio = mraa_gpio
# Define mapping of Adafruit GPIO library constants to mraa constants
self._dir_mapping = { OUT: self.mraa_gpio.DIR_OUT,
IN: self.mraa_gpio.DIR_IN }
self._pud_mapping = { PUD_OFF: self.mraa_gpio.MODE_STRONG,
PUD_UP: self.mraa_gpio.MODE_HIZ,
PUD_DOWN: self.mraa_gpio.MODE_PULLDOWN }
self._edge_mapping = { RISING: self.mraa_gpio.EDGE_RISING,
FALLING: self.mraa_gpio.EDGE_FALLING,
BOTH: self.mraa_gpio.EDGE_BOTH }
def setup(self,pin,mode):
"""Set the input or output mode for a specified pin. Mode should be
either DIR_IN or DIR_OUT.
"""
self.mraa_gpio.Gpio.dir(self.mraa_gpio.Gpio(pin),self._dir_mapping[mode])
def output(self,pin,value):
"""Set the specified pin the provided high/low value. Value should be
either 1 (ON or HIGH), or 0 (OFF or LOW) or a boolean.
"""
self.mraa_gpio.Gpio.write(self.mraa_gpio.Gpio(pin), value)
def input(self,pin):
"""Read the specified pin and return HIGH/true if the pin is pulled high,
or LOW/false if pulled low.
"""
return self.mraa_gpio.Gpio.read(self.mraa_gpio.Gpio(pin))
def add_event_detect(self, pin, edge, callback=None, bouncetime=-1):
"""Enable edge detection events for a particular GPIO channel. Pin
should be type IN. Edge must be RISING, FALLING or BOTH. Callback is a
function for the event. Bouncetime is switch bounce timeout in ms for
callback
"""
kwargs = {}
if callback:
kwargs['callback']=callback
if bouncetime > 0:
kwargs['bouncetime']=bouncetime
self.mraa_gpio.Gpio.isr(self.mraa_gpio.Gpio(pin), self._edge_mapping[edge], **kwargs)
def remove_event_detect(self, pin):
"""Remove edge detection for a particular GPIO channel. Pin should be
type IN.
"""
self.mraa_gpio.Gpio.isrExit(self.mraa_gpio.Gpio(pin))
def wait_for_edge(self, pin, edge):
"""Wait for an edge. Pin should be type IN. Edge must be RISING,
FALLING or BOTH.
"""
self.bbio_gpio.wait_for_edge(self.mraa_gpio.Gpio(pin), self._edge_mapping[edge])
def get_platform_gpio(**keywords):
"""Attempt to return a GPIO instance for the platform which the code is being
executed on. Currently supports only the Raspberry Pi using the RPi.GPIO
library and Beaglebone Black using the Adafruit_BBIO library. Will throw an
exception if a GPIO instance can't be created for the current platform. The
returned GPIO object is an instance of BaseGPIO.
"""
plat = Platform.platform_detect()
if plat == Platform.RASPBERRY_PI:
import RPi.GPIO
return RPiGPIOAdapter(RPi.GPIO, **keywords)
elif plat == Platform.BEAGLEBONE_BLACK:
import Adafruit_BBIO.GPIO
return AdafruitBBIOAdapter(Adafruit_BBIO.GPIO, **keywords)
elif plat == Platform.MINNOWBOARD:
import mraa
return AdafruitMinnowAdapter(mraa, **keywords)
elif plat == Platform.UNKNOWN:
raise RuntimeError('Could not determine platform.')

200
darkwater_640/I2C.py Normal file
View File

@@ -0,0 +1,200 @@
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
# Based on Adafruit_I2C.py created by Kevin Townsend.
#
# 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.
import logging
import subprocess
import Adafruit_GPIO.Platform as Platform
def reverseByteOrder(data):
"""Reverses the byte order of an int (16-bit) or long (32-bit) value."""
# Courtesy Vishal Sapre
byteCount = len(hex(data)[2:].replace('L','')[::2])
val = 0
for i in range(byteCount):
val = (val << 8) | (data & 0xff)
data >>= 8
return val
def get_default_bus():
"""Return the default bus number based on the device platform. For a
Raspberry Pi either bus 0 or 1 (based on the Pi revision) will be returned.
For a Beaglebone Black the first user accessible bus, 1, will be returned.
"""
plat = Platform.platform_detect()
if plat == Platform.RASPBERRY_PI:
if Platform.pi_revision() == 1:
# Revision 1 Pi uses I2C bus 0.
return 0
else:
# Revision 2 Pi uses I2C bus 1.
return 1
elif plat == Platform.BEAGLEBONE_BLACK:
# Beaglebone Black has multiple I2C buses, default to 1 (P9_19 and P9_20).
return 1
else:
raise RuntimeError('Could not determine default I2C bus for platform.')
def get_i2c_device(address, busnum=None, i2c_interface=None, **kwargs):
"""Return an I2C device for the specified address and on the specified bus.
If busnum isn't specified, the default I2C bus for the platform will attempt
to be detected.
"""
if busnum is None:
busnum = get_default_bus()
return Device(address, busnum, i2c_interface, **kwargs)
def require_repeated_start():
"""Enable repeated start conditions for I2C register reads. This is the
normal behavior for I2C, however on some platforms like the Raspberry Pi
there are bugs which disable repeated starts unless explicitly enabled with
this function. See this thread for more details:
http://www.raspberrypi.org/forums/viewtopic.php?f=44&t=15840
"""
plat = Platform.platform_detect()
if plat == Platform.RASPBERRY_PI:
# On the Raspberry Pi there is a bug where register reads don't send a
# repeated start condition like the kernel smbus I2C driver functions
# define. As a workaround this bit in the BCM2708 driver sysfs tree can
# be changed to enable I2C repeated starts.
subprocess.check_call('chmod 666 /sys/module/i2c_bcm2708/parameters/combined', shell=True)
subprocess.check_call('echo -n 1 > /sys/module/i2c_bcm2708/parameters/combined', shell=True)
# Other platforms are a no-op because they (presumably) have the correct
# behavior and send repeated starts.
class Device(object):
"""Class for communicating with an I2C device using the adafruit-pureio pure
python smbus library, or other smbus compatible I2C interface. Allows reading
and writing 8-bit, 16-bit, and byte array values to registers
on the device."""
def __init__(self, address, busnum, i2c_interface=None):
"""Create an instance of the I2C device at the specified address on the
specified I2C bus number."""
self._address = address
if i2c_interface is None:
# Use pure python I2C interface if none is specified.
import Adafruit_PureIO.smbus
self._bus = Adafruit_PureIO.smbus.SMBus(busnum)
else:
# Otherwise use the provided class to create an smbus interface.
self._bus = i2c_interface(busnum)
self._logger = logging.getLogger('Adafruit_I2C.Device.Bus.{0}.Address.{1:#0X}' \
.format(busnum, address))
def writeRaw8(self, value):
"""Write an 8-bit value on the bus (without register)."""
value = value & 0xFF
self._bus.write_byte(self._address, value)
self._logger.debug("Wrote 0x%02X",
value)
def write8(self, register, value):
"""Write an 8-bit value to the specified register."""
value = value & 0xFF
self._bus.write_byte_data(self._address, register, value)
self._logger.debug("Wrote 0x%02X to register 0x%02X",
value, register)
def write16(self, register, value):
"""Write a 16-bit value to the specified register."""
value = value & 0xFFFF
self._bus.write_word_data(self._address, register, value)
self._logger.debug("Wrote 0x%04X to register pair 0x%02X, 0x%02X",
value, register, register+1)
def writeList(self, register, data):
"""Write bytes to the specified register."""
self._bus.write_i2c_block_data(self._address, register, data)
self._logger.debug("Wrote to register 0x%02X: %s",
register, data)
def readList(self, register, length):
"""Read a length number of bytes from the specified register. Results
will be returned as a bytearray."""
results = self._bus.read_i2c_block_data(self._address, register, length)
self._logger.debug("Read the following from register 0x%02X: %s",
register, results)
return results
def readRaw8(self):
"""Read an 8-bit value on the bus (without register)."""
result = self._bus.read_byte(self._address) & 0xFF
self._logger.debug("Read 0x%02X",
result)
return result
def readU8(self, register):
"""Read an unsigned byte from the specified register."""
result = self._bus.read_byte_data(self._address, register) & 0xFF
self._logger.debug("Read 0x%02X from register 0x%02X",
result, register)
return result
def readS8(self, register):
"""Read a signed byte from the specified register."""
result = self.readU8(register)
if result > 127:
result -= 256
return result
def readU16(self, register, little_endian=True):
"""Read an unsigned 16-bit value from the specified register, with the
specified endianness (default little endian, or least significant byte
first)."""
result = self._bus.read_word_data(self._address,register) & 0xFFFF
self._logger.debug("Read 0x%04X from register pair 0x%02X, 0x%02X",
result, register, register+1)
# Swap bytes if using big endian because read_word_data assumes little
# endian on ARM (little endian) systems.
if not little_endian:
result = ((result << 8) & 0xFF00) + (result >> 8)
return result
def readS16(self, register, little_endian=True):
"""Read a signed 16-bit value from the specified register, with the
specified endianness (default little endian, or least significant byte
first)."""
result = self.readU16(register, little_endian)
if result > 32767:
result -= 65536
return result
def readU16LE(self, register):
"""Read an unsigned 16-bit value from the specified register, in little
endian byte order."""
return self.readU16(register, little_endian=True)
def readU16BE(self, register):
"""Read an unsigned 16-bit value from the specified register, in big
endian byte order."""
return self.readU16(register, little_endian=False)
def readS16LE(self, register):
"""Read a signed 16-bit value from the specified register, in little
endian byte order."""
return self.readS16(register, little_endian=True)
def readS16BE(self, register):
"""Read a signed 16-bit value from the specified register, in big
endian byte order."""
return self.readS16(register, little_endian=False)

167
darkwater_640/PCA9685.py Normal file
View File

@@ -0,0 +1,167 @@
# Copyright (c) 2016 Adafruit Industries
# Author: Tony DiCola
#
# Updated by: Dark Water
#
# 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 __future__ import division
import logging
import time
import math
# Registers/etc:
PCA9685_ADDRESS = 0x60
MODE1 = 0x00
MODE2 = 0x01
SUBADR1 = 0x02
SUBADR2 = 0x03
SUBADR3 = 0x04
PRESCALE = 0xFE
LED0_ON_L = 0x06
LED0_ON_H = 0x07
LED0_OFF_L = 0x08
LED0_OFF_H = 0x09
ALL_LED_ON_L = 0xFA
ALL_LED_ON_H = 0xFB
ALL_LED_OFF_L = 0xFC
ALL_LED_OFF_H = 0xFD
# Bits:
RESTART = 0x80
SLEEP = 0x10
ALLCALL = 0x01
INVRT = 0x10
OUTDRV = 0x04
logger = logging.getLogger(__name__)
def software_reset(i2c=None, **kwargs):
"""Sends a software reset (SWRST) command to all servo drivers on the bus."""
# Setup I2C interface for device 0x00 to talk to all of them.
if i2c is None:
import Adafruit_GPIO.I2C as I2C
i2c = I2C
self._device = i2c.get_i2c_device(0x00, **kwargs)
self._device.writeRaw8(0x06) # SWRST
class PCA9685(object):
"""PCA9685 PWM LED/servo controller."""
def __init__(self, address=PCA9685_ADDRESS, i2c=None, **kwargs):
"""Initialize the PCA9685."""
# Setup I2C interface for the device.
if i2c is None:
import Adafruit_GPIO.I2C as I2C
i2c = I2C
self._device = i2c.get_i2c_device(address, **kwargs)
self.set_all_pwm(0, 0)
self._device.write8(MODE2, OUTDRV)
self._device.write8(MODE1, ALLCALL)
time.sleep(0.005) # wait for oscillator
mode1 = self._device.readU8(MODE1)
mode1 = mode1 & ~SLEEP # wake up (reset sleep)
self._device.write8(MODE1, mode1)
time.sleep(0.005) # wait for oscillator
def set_pwm_freq(self, freq_hz):
"""Set the PWM frequency to the provided value in hertz."""
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq_hz)
prescaleval -= 1.0
logger.debug('Setting PWM frequency to {0} Hz'.format(freq_hz))
logger.debug('Estimated pre-scale: {0}'.format(prescaleval))
prescale = int(math.floor(prescaleval + 0.5))
logger.debug('Final pre-scale: {0}'.format(prescale))
oldmode = self._device.readU8(MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self._device.write8(MODE1, newmode) # go to sleep
self._device.write8(PRESCALE, prescale)
self._device.write8(MODE1, oldmode)
time.sleep(0.005)
self._device.write8(MODE1, oldmode | 0x80)
def set_pwm_freq_min(self, freq_hz, correctionFactor=1.0):
"Sets the PWM frequency"
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq_hz)
prescaleval -= 1.0
if (self.debug):
print "Setting PWM frequency to %d Hz" % freq_hz
print "Estimated pre-scale: %d" % prescaleval
prescale = math.floor(prescaleval * correctionFactor + 0.5)
if (self.debug):
print "Final pre-scale: %d" % prescale
oldmode = self.i2c.readU8(self.__MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self.i2c.write8(self.__MODE1, newmode) # go to sleep
self.i2c.write8(self.__PRESCALE, int(math.floor(prescale)))
self.i2c.write8(self.__MODE1, oldmode)
time.sleep(0.005)
self.i2c.write8(self.__MODE1, oldmode | 0x80)
def set_pwm_freq_max(self, freq_hz, correctionFactor=1.0):
"Sets the PWM frequency"
prescaleval = 25000000.0 # 25MHz
prescaleval /= 4096.0 # 12-bit
prescaleval /= float(freq_hz)
prescaleval -= 1.0
if (self.debug):
print "Setting PWM frequency to %d Hz" % freq_hz
print "Estimated pre-scale: %d" % prescaleval
prescale = math.ceil(prescaleval * correctionFactor + 0.5)
if (self.debug):
print "Final pre-scale: %d" % prescale
oldmode = self.i2c.readU8(self.__MODE1);
newmode = (oldmode & 0x7F) | 0x10 # sleep
self.i2c.write8(self.__MODE1, newmode) # go to sleep
self.i2c.write8(self.__PRESCALE, int(math.floor(prescale)))
self.i2c.write8(self.__MODE1, oldmode)
time.sleep(0.005)
self.i2c.write8(self.__MODE1, oldmode | 0x80)
def get_pwm_freq(self):
prescale = self.i2c.readU8(self.__PRESCALE)
calcfreq = 25000000.0 / 4096.0 / ( float(prescale) + 1 )
if (self.debug):
print "Got pre-scale: %d" % prescale
print "Calculated Frequency: %d" % calcfreq
return calcfreq
def set_pwm(self, channel, on, off):
"""Sets a single PWM channel."""
self._device.write8(LED0_ON_L+4*channel, on & 0xFF)
self._device.write8(LED0_ON_H+4*channel, on >> 8)
self._device.write8(LED0_OFF_L+4*channel, off & 0xFF)
self._device.write8(LED0_OFF_H+4*channel, off >> 8)
def set_all_pwm(self, on, off):
"""Sets all PWM channels."""
self._device.write8(ALL_LED_ON_L, on & 0xFF)
self._device.write8(ALL_LED_ON_H, on >> 8)
self._device.write8(ALL_LED_OFF_L, off & 0xFF)
self._device.write8(ALL_LED_OFF_H, off >> 8)

106
darkwater_640/Platform.py Normal file
View File

@@ -0,0 +1,106 @@
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
# 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.
import platform
import re
# Platform identification constants.
UNKNOWN = 0
RASPBERRY_PI = 1
BEAGLEBONE_BLACK = 2
MINNOWBOARD = 3
def platform_detect():
"""Detect if running on the Raspberry Pi or Beaglebone Black and return the
platform type. Will return RASPBERRY_PI, BEAGLEBONE_BLACK, or UNKNOWN."""
# Handle Raspberry Pi
pi = pi_version()
if pi is not None:
return RASPBERRY_PI
# Handle Beaglebone Black
# TODO: Check the Beaglebone Black /proc/cpuinfo value instead of reading
# the platform.
plat = platform.platform()
if plat.lower().find('armv7l-with-debian') > -1:
return BEAGLEBONE_BLACK
elif plat.lower().find('armv7l-with-ubuntu') > -1:
return BEAGLEBONE_BLACK
elif plat.lower().find('armv7l-with-glibc2.4') > -1:
return BEAGLEBONE_BLACK
# Handle Minnowboard
# Assumption is that mraa is installed
try:
import mraa
if mraa.getPlatformName()=='MinnowBoard MAX':
return MINNOWBOARD
except ImportError:
pass
# Couldn't figure out the platform, just return unknown.
return UNKNOWN
def pi_revision():
"""Detect the revision number of a Raspberry Pi, useful for changing
functionality like default I2C bus based on revision."""
# Revision list available at: http://elinux.org/RPi_HardwareHistory#Board_Revision_History
with open('/proc/cpuinfo', 'r') as infile:
for line in infile:
# Match a line of the form "Revision : 0002" while ignoring extra
# info in front of the revsion (like 1000 when the Pi was over-volted).
match = re.match('Revision\s+:\s+.*(\w{4})$', line, flags=re.IGNORECASE)
if match and match.group(1) in ['0000', '0002', '0003']:
# Return revision 1 if revision ends with 0000, 0002 or 0003.
return 1
elif match:
# Assume revision 2 if revision ends with any other 4 chars.
return 2
# Couldn't find the revision, throw an exception.
raise RuntimeError('Could not determine Raspberry Pi revision.')
def pi_version():
"""Detect the version of the Raspberry Pi. Returns either 1, 2 or
None depending on if it's a Raspberry Pi 1 (model A, B, A+, B+),
Raspberry Pi 2 (model B+), or not a Raspberry Pi.
"""
# Check /proc/cpuinfo for the Hardware field value.
# 2708 is pi 1
# 2709 is pi 2
# Anything else is not a pi.
with open('/proc/cpuinfo', 'r') as infile:
cpuinfo = infile.read()
# Match a line like 'Hardware : BCM2709'
match = re.search('^Hardware\s+:\s+(\w+)$', cpuinfo,
flags=re.MULTILINE | re.IGNORECASE)
if not match:
# Couldn't find the hardware, assume it isn't a pi.
return None
if match.group(1) == 'BCM2708':
# Pi 1
return 1
elif match.group(1) == 'BCM2709':
# Pi 2
return 2
else:
# Something else, not a pi.
return None

336
darkwater_640/SPI.py Normal file
View File

@@ -0,0 +1,336 @@
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
#
# 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.
import operator
import time
import Adafruit_GPIO as GPIO
MSBFIRST = 0
LSBFIRST = 1
class SpiDev(object):
"""Hardware-based SPI implementation using the spidev interface."""
def __init__(self, port, device, max_speed_hz=500000):
"""Initialize an SPI device using the SPIdev interface. Port and device
identify the device, for example the device /dev/spidev1.0 would be port
1 and device 0.
"""
import spidev
self._device = spidev.SpiDev()
self._device.open(port, device)
self._device.max_speed_hz=max_speed_hz
# Default to mode 0.
self._device.mode = 0
def set_clock_hz(self, hz):
"""Set the speed of the SPI clock in hertz. Note that not all speeds
are supported and a lower speed might be chosen by the hardware.
"""
self._device.max_speed_hz=hz
def set_mode(self, mode):
"""Set SPI mode which controls clock polarity and phase. Should be a
numeric value 0, 1, 2, or 3. See wikipedia page for details on meaning:
http://en.wikipedia.org/wiki/Serial_Peripheral_Interface_Bus
"""
if mode < 0 or mode > 3:
raise ValueError('Mode must be a value 0, 1, 2, or 3.')
self._device.mode = mode
def set_bit_order(self, order):
"""Set order of bits to be read/written over serial lines. Should be
either MSBFIRST for most-significant first, or LSBFIRST for
least-signifcant first.
"""
if order == MSBFIRST:
self._device.lsbfirst = False
elif order == LSBFIRST:
self._device.lsbfirst = True
else:
raise ValueError('Order must be MSBFIRST or LSBFIRST.')
def close(self):
"""Close communication with the SPI device."""
self._device.close()
def write(self, data):
"""Half-duplex SPI write. The specified array of bytes will be clocked
out the MOSI line.
"""
self._device.writebytes(data)
def read(self, length):
"""Half-duplex SPI read. The specified length of bytes will be clocked
in the MISO line and returned as a bytearray object.
"""
return bytearray(self._device.readbytes(length))
def transfer(self, data):
"""Full-duplex SPI read and write. The specified array of bytes will be
clocked out the MOSI line, while simultaneously bytes will be read from
the MISO line. Read bytes will be returned as a bytearray object.
"""
return bytearray(self._device.xfer2(data))
class SpiDevMraa(object):
"""Hardware SPI implementation with the mraa library on Minnowboard"""
def __init__(self, port, device, max_speed_hz=500000):
import mraa
self._device = mraa.Spi(0)
self._device.mode(0)
def set_clock_hz(self, hz):
"""Set the speed of the SPI clock in hertz. Note that not all speeds
are supported and a lower speed might be chosen by the hardware.
"""
self._device.frequency(hz)
def set_mode(self,mode):
"""Set SPI mode which controls clock polarity and phase. Should be a
numeric value 0, 1, 2, or 3. See wikipedia page for details on meaning:
http://en.wikipedia.org/wiki/Serial_Peripheral_Interface_Bus
"""
if mode < 0 or mode > 3:
raise ValueError('Mode must be a value 0, 1, 2, or 3.')
self._device.mode(mode)
def set_mode(self,mode):
"""Set SPI mode which controls clock polarity and phase. Should be a
numeric value 0, 1, 2, or 3. See wikipedia page for details on meaning:
http://en.wikipedia.org/wiki/Serial_Peripheral_Interface_Bus
"""
if mode < 0 or mode > 3:
raise ValueError('Mode must be a value 0, 1, 2, or 3.')
self._device.mode(mode)
def set_bit_order(self, order):
"""Set order of bits to be read/written over serial lines. Should be
either MSBFIRST for most-significant first, or LSBFIRST for
least-signifcant first.
"""
if order == MSBFIRST:
self._device.lsbmode(False)
elif order == LSBFIRST:
self._device.lsbmode(True)
else:
raise ValueError('Order must be MSBFIRST or LSBFIRST.')
def close(self):
"""Close communication with the SPI device."""
self._device.Spi()
def write(self, data):
"""Half-duplex SPI write. The specified array of bytes will be clocked
out the MOSI line.
"""
self._device.write(bytearray(data))
class BitBang(object):
"""Software-based implementation of the SPI protocol over GPIO pins."""
def __init__(self, gpio, sclk, mosi=None, miso=None, ss=None):
"""Initialize bit bang (or software) based SPI. Must provide a BaseGPIO
class, the SPI clock, and optionally MOSI, MISO, and SS (slave select)
pin numbers. If MOSI is set to None then writes will be disabled and fail
with an error, likewise for MISO reads will be disabled. If SS is set to
None then SS will not be asserted high/low by the library when
transfering data.
"""
self._gpio = gpio
self._sclk = sclk
self._mosi = mosi
self._miso = miso
self._ss = ss
# Set pins as outputs/inputs.
gpio.setup(sclk, GPIO.OUT)
if mosi is not None:
gpio.setup(mosi, GPIO.OUT)
if miso is not None:
gpio.setup(miso, GPIO.IN)
if ss is not None:
gpio.setup(ss, GPIO.OUT)
# Assert SS high to start with device communication off.
gpio.set_high(ss)
# Assume mode 0.
self.set_mode(0)
# Assume most significant bit first order.
self.set_bit_order(MSBFIRST)
def set_clock_hz(self, hz):
"""Set the speed of the SPI clock. This is unsupported with the bit
bang SPI class and will be ignored.
"""
pass
def set_mode(self, mode):
"""Set SPI mode which controls clock polarity and phase. Should be a
numeric value 0, 1, 2, or 3. See wikipedia page for details on meaning:
http://en.wikipedia.org/wiki/Serial_Peripheral_Interface_Bus
"""
if mode < 0 or mode > 3:
raise ValueError('Mode must be a value 0, 1, 2, or 3.')
if mode & 0x02:
# Clock is normally high in mode 2 and 3.
self._clock_base = GPIO.HIGH
else:
# Clock is normally low in mode 0 and 1.
self._clock_base = GPIO.LOW
if mode & 0x01:
# Read on trailing edge in mode 1 and 3.
self._read_leading = False
else:
# Read on leading edge in mode 0 and 2.
self._read_leading = True
# Put clock into its base state.
self._gpio.output(self._sclk, self._clock_base)
def set_bit_order(self, order):
"""Set order of bits to be read/written over serial lines. Should be
either MSBFIRST for most-significant first, or LSBFIRST for
least-signifcant first.
"""
# Set self._mask to the bitmask which points at the appropriate bit to
# read or write, and appropriate left/right shift operator function for
# reading/writing.
if order == MSBFIRST:
self._mask = 0x80
self._write_shift = operator.lshift
self._read_shift = operator.rshift
elif order == LSBFIRST:
self._mask = 0x01
self._write_shift = operator.rshift
self._read_shift = operator.lshift
else:
raise ValueError('Order must be MSBFIRST or LSBFIRST.')
def close(self):
"""Close the SPI connection. Unused in the bit bang implementation."""
pass
def write(self, data, assert_ss=True, deassert_ss=True):
"""Half-duplex SPI write. If assert_ss is True, the SS line will be
asserted low, the specified bytes will be clocked out the MOSI line, and
if deassert_ss is True the SS line be put back high.
"""
# Fail MOSI is not specified.
if self._mosi is None:
raise RuntimeError('Write attempted with no MOSI pin specified.')
if assert_ss and self._ss is not None:
self._gpio.set_low(self._ss)
for byte in data:
for i in range(8):
# Write bit to MOSI.
if self._write_shift(byte, i) & self._mask:
self._gpio.set_high(self._mosi)
else:
self._gpio.set_low(self._mosi)
# Flip clock off base.
self._gpio.output(self._sclk, not self._clock_base)
# Return clock to base.
self._gpio.output(self._sclk, self._clock_base)
if deassert_ss and self._ss is not None:
self._gpio.set_high(self._ss)
def read(self, length, assert_ss=True, deassert_ss=True):
"""Half-duplex SPI read. If assert_ss is true, the SS line will be
asserted low, the specified length of bytes will be clocked in the MISO
line, and if deassert_ss is true the SS line will be put back high.
Bytes which are read will be returned as a bytearray object.
"""
if self._miso is None:
raise RuntimeError('Read attempted with no MISO pin specified.')
if assert_ss and self._ss is not None:
self._gpio.set_low(self._ss)
result = bytearray(length)
for i in range(length):
for j in range(8):
# Flip clock off base.
self._gpio.output(self._sclk, not self._clock_base)
# Handle read on leading edge of clock.
if self._read_leading:
if self._gpio.is_high(self._miso):
# Set bit to 1 at appropriate location.
result[i] |= self._read_shift(self._mask, j)
else:
# Set bit to 0 at appropriate location.
result[i] &= ~self._read_shift(self._mask, j)
# Return clock to base.
self._gpio.output(self._sclk, self._clock_base)
# Handle read on trailing edge of clock.
if not self._read_leading:
if self._gpio.is_high(self._miso):
# Set bit to 1 at appropriate location.
result[i] |= self._read_shift(self._mask, j)
else:
# Set bit to 0 at appropriate location.
result[i] &= ~self._read_shift(self._mask, j)
if deassert_ss and self._ss is not None:
self._gpio.set_high(self._ss)
return result
def transfer(self, data, assert_ss=True, deassert_ss=True):
"""Full-duplex SPI read and write. If assert_ss is true, the SS line
will be asserted low, the specified bytes will be clocked out the MOSI
line while bytes will also be read from the MISO line, and if
deassert_ss is true the SS line will be put back high. Bytes which are
read will be returned as a bytearray object.
"""
if self._mosi is None:
raise RuntimeError('Write attempted with no MOSI pin specified.')
if self._mosi is None:
raise RuntimeError('Read attempted with no MISO pin specified.')
if assert_ss and self._ss is not None:
self._gpio.set_low(self._ss)
result = bytearray(len(data))
for i in range(len(data)):
for j in range(8):
# Write bit to MOSI.
if self._write_shift(data[i], j) & self._mask:
self._gpio.set_high(self._mosi)
else:
self._gpio.set_low(self._mosi)
# Flip clock off base.
self._gpio.output(self._sclk, not self._clock_base)
# Handle read on leading edge of clock.
if self._read_leading:
if self._gpio.is_high(self._miso):
# Set bit to 1 at appropriate location.
result[i] |= self._read_shift(self._mask, j)
else:
# Set bit to 0 at appropriate location.
result[i] &= ~self._read_shift(self._mask, j)
# Return clock to base.
self._gpio.output(self._sclk, self._clock_base)
# Handle read on trailing edge of clock.
if not self._read_leading:
if self._gpio.is_high(self._miso):
# Set bit to 1 at appropriate location.
result[i] |= self._read_shift(self._mask, j)
else:
# Set bit to 0 at appropriate location.
result[i] &= ~self._read_shift(self._mask, j)
if deassert_ss and self._ss is not None:
self._gpio.set_high(self._ss)
return result

1
darkwater_640/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .dw640HAT import dw_DCMotor, dw_MotorCONTROL

275
darkwater_640/darkwater_640.py Normal file
View File

@@ -0,0 +1,275 @@
#!/usr/bin/python
import RPi.GPIO as GPIO
from Adafruit_PWM_Servo_Driver import PWM
import time
# class Adafruit_StepperMotor:
# MICROSTEPS = 8
# MICROSTEP_CURVE = [0, 50, 98, 142, 180, 212, 236, 250, 255]
#
# #MICROSTEPS = 16
# # a sinusoidal curve NOT LINEAR!
# #MICROSTEP_CURVE = [0, 25, 50, 74, 98, 120, 141, 162, 180, 197, 212, 225, 236, 244, 250, 253, 255]
#
# def __init__(self, controller, num, steps=200):
# self.MC = controller
# self.revsteps = steps
# self.motornum = num
# self.sec_per_step = 0.1
# self.steppingcounter = 0
# self.currentstep = 0
#
# num -= 1
#
# if (num == 0):
# self.PWMA = 8
# self.AIN2 = 9
# self.AIN1 = 10
# self.PWMB = 13
# self.BIN2 = 12
# self.BIN1 = 11
# elif (num == 1):
# self.PWMA = 2
# self.AIN2 = 3
# self.AIN1 = 4
# self.PWMB = 7
# self.BIN2 = 6
# self.BIN1 = 5
# else:
# raise NameError('MotorHAT Stepper must be between 1 and 2 inclusive')
#
# def setSpeed(self, rpm):
# self.sec_per_step = 60.0 / (self.revsteps * rpm)
# self.steppingcounter = 0
#
# def oneStep(self, dir, style):
# pwm_a = pwm_b = 255
#
# # first determine what sort of stepping procedure we're up to
# if (style == Adafruit_MotorHAT.SINGLE):
# if ((self.currentstep/(self.MICROSTEPS/2)) % 2):
# # we're at an odd step, weird
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += self.MICROSTEPS/2
# else:
# self.currentstep -= self.MICROSTEPS/2
# else:
# # go to next even step
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += self.MICROSTEPS
# else:
# self.currentstep -= self.MICROSTEPS
# if (style == Adafruit_MotorHAT.DOUBLE):
# if not (self.currentstep/(self.MICROSTEPS/2) % 2):
# # we're at an even step, weird
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += self.MICROSTEPS/2
# else:
# self.currentstep -= self.MICROSTEPS/2
# else:
# # go to next odd step
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += self.MICROSTEPS
# else:
# self.currentstep -= self.MICROSTEPS
# if (style == Adafruit_MotorHAT.INTERLEAVE):
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += self.MICROSTEPS/2
# else:
# self.currentstep -= self.MICROSTEPS/2
#
# if (style == Adafruit_MotorHAT.MICROSTEP):
# if (dir == Adafruit_MotorHAT.FORWARD):
# self.currentstep += 1
# else:
# self.currentstep -= 1
#
# # go to next 'step' and wrap around
# self.currentstep += self.MICROSTEPS * 4
# self.currentstep %= self.MICROSTEPS * 4
#
# pwm_a = pwm_b = 0
# if (self.currentstep >= 0) and (self.currentstep < self.MICROSTEPS):
# pwm_a = self.MICROSTEP_CURVE[self.MICROSTEPS - self.currentstep]
# pwm_b = self.MICROSTEP_CURVE[self.currentstep]
# elif (self.currentstep >= self.MICROSTEPS) and (self.currentstep < self.MICROSTEPS*2):
# pwm_a = self.MICROSTEP_CURVE[self.currentstep - self.MICROSTEPS]
# pwm_b = self.MICROSTEP_CURVE[self.MICROSTEPS*2 - self.currentstep]
# elif (self.currentstep >= self.MICROSTEPS*2) and (self.currentstep < self.MICROSTEPS*3):
# pwm_a = self.MICROSTEP_CURVE[self.MICROSTEPS*3 - self.currentstep]
# pwm_b = self.MICROSTEP_CURVE[self.currentstep - self.MICROSTEPS*2]
# elif (self.currentstep >= self.MICROSTEPS*3) and (self.currentstep < self.MICROSTEPS*4):
# pwm_a = self.MICROSTEP_CURVE[self.currentstep - self.MICROSTEPS*3]
# pwm_b = self.MICROSTEP_CURVE[self.MICROSTEPS*4 - self.currentstep]
#
#
# # go to next 'step' and wrap around
# self.currentstep += self.MICROSTEPS * 4
# self.currentstep %= self.MICROSTEPS * 4
#
# # only really used for microstepping, otherwise always on!
# self.MC._pwm.setPWM(self.PWMA, 0, pwm_a*16)
# self.MC._pwm.setPWM(self.PWMB, 0, pwm_b*16)
#
# # set up coil energizing!
# coils = [0, 0, 0, 0]
#
# if (style == Adafruit_MotorHAT.MICROSTEP):
# if (self.currentstep >= 0) and (self.currentstep < self.MICROSTEPS):
# coils = [1, 1, 0, 0]
# elif (self.currentstep >= self.MICROSTEPS) and (self.currentstep < self.MICROSTEPS*2):
# coils = [0, 1, 1, 0]
# elif (self.currentstep >= self.MICROSTEPS*2) and (self.currentstep < self.MICROSTEPS*3):
# coils = [0, 0, 1, 1]
# elif (self.currentstep >= self.MICROSTEPS*3) and (self.currentstep < self.MICROSTEPS*4):
# coils = [1, 0, 0, 1]
# else:
# step2coils = [ [1, 0, 0, 0],
# [1, 1, 0, 0],
# [0, 1, 0, 0],
# [0, 1, 1, 0],
# [0, 0, 1, 0],
# [0, 0, 1, 1],
# [0, 0, 0, 1],
# [1, 0, 0, 1] ]
# coils = step2coils[self.currentstep/(self.MICROSTEPS/2)]
#
# #print "coils state = " + str(coils)
# self.MC.setPin(self.AIN2, coils[0])
# self.MC.setPin(self.BIN1, coils[1])
# self.MC.setPin(self.AIN1, coils[2])
# self.MC.setPin(self.BIN2, coils[3])
#
# return self.currentstep
#
# def step(self, steps, direction, stepstyle):
# s_per_s = self.sec_per_step
# lateststep = 0
#
# if (stepstyle == Adafruit_MotorHAT.INTERLEAVE):
# s_per_s = s_per_s / 2.0
# if (stepstyle == Adafruit_MotorHAT.MICROSTEP):
# s_per_s /= self.MICROSTEPS
# steps *= self.MICROSTEPS
#
# print s_per_s, " sec per step"
#
# for s in range(steps):
# lateststep = self.oneStep(direction, stepstyle)
# time.sleep(s_per_s)
#
# if (stepstyle == Adafruit_MotorHAT.MICROSTEP):
# # this is an edge case, if we are in between full steps, lets just keep going
# # so we end on a full step
# while (lateststep != 0) and (lateststep != self.MICROSTEPS):
# lateststep = self.oneStep(dir, stepstyle)
# time.sleep(s_per_s)
class dw_DCMotor:
def __init__(self, controller, num):
self.speed = 0
self.MC = controller
self.motornum = num
modepin = in1 = in2 = 0
if (num == 0):
in2 = 2 #phase
in1 = 3 #enable
elif (num == 1):
in2 = 4 #phase
in1 = 5 #enable
elif (num == 2):
in2 = 15 #phase
in1 = 14 #enable
elif (num == 3):
in2 = 13 #phase
in1 = 12 #enable
elif (num == 4):
in2 = 8 #phase
in1 = 9 #enable
elif (num == 5):
in2 = 10 #phase
in1 = 11 #enable
else:
raise NameError('MotorHAT Motor must be between 1 and 6 inclusive')
self.PHpin = in2
self.ENpin = in1
# switch off
self.run(dw_MotorCONTROL.RELEASE, 0)
def setMotorSpeed(self, value):
if(value > 0) and (value <= 255):
self.run(dw_MotorCONTROL.FORWARD, value)
if(value == 0):
self.run(dw_MotorCONTROL.RELEASE, value)
if(value < 0) and (value >= -255):
self.run(dw_MotorCONTROL.BACKWARD, abs(value))
def run(self, command, speed = 0):
if not self.MC:
return
if (command == dw_MotorCONTROL.FORWARD):
self.MC.setPin(self.PHpin, 0)
self.MC._pwm.setPWM(self.ENpin, 0, speed*16)
if (command == dw_MotorCONTROL.BACKWARD):
self.MC.setPin(self.PHpin, 1)
self.MC._pwm.setPWM(self.ENpin, 0, speed*16)
if (command == dw_MotorCONTROL.RELEASE):
self.MC.setPin(self.PHpin, 0)
self.MC.setPin(self.ENpin, 0)
def setSpeed(self, speed):
if (speed < 0):
speed = 0
if (speed > 255):
speed = 255
self.MC._pwm.setPWM(self.ENpin, 0, speed*16)
class dw_MotorCONTROL:
FORWARD = 1
BACKWARD = 2
BRAKE = 3
RELEASE = 4
SINGLE = 1
DOUBLE = 2
INTERLEAVE = 3
MICROSTEP = 4
ININ = 0
PHASE = 1
def __init__(self, addr = 0x60, freq = 1600):
self._i2caddr = addr # default addr on HAT
self._frequency = freq # default @1600Hz PWM freq
# self.steppers = [ Adafruit_StepperMotor(self, 1), Adafruit_StepperMotor(self, 2) ]
self._pwm = PWM(addr, debug=False)
self._pwm.setPWMFreq(self._frequency)
# Just gonna default to high for now
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(17, GPIO.OUT)
GPIO.output(17, GPIO.HIGH)
self.motors = [ dw_DCMotor(self, m) for m in range(6) ]
def setPin(self, pin, value):
if (pin < 0) or (pin > 15):
raise NameError('PWM pin must be between 0 and 15 inclusive')
if (value != 0) and (value != 1):
raise NameError('Pin value must be 0 or 1!')
if (value == 0):
self._pwm.setPWM(pin, 0, 4096)
if (value == 1):
self._pwm.setPWM(pin, 4096, 0)
def getMotor(self, num):
if (num < 1) or (num > 6):
raise NameError('MotorHAT Motor must be between 1 and 6 inclusive')
return self.motors[num-1]
def allOff(self):
for y in range(5):
self.motors[y].run(dw_MotorCONTROL.RELEASE, 0)