Created
March 6, 2016 06:37
-
-
Save lanewinfield/fecc5f259e9017b88d6c to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
import os | |
import threading | |
from subprocess import call | |
import glob | |
import re | |
# Display | |
import Adafruit_GPIO.SPI as SPI | |
import Adafruit_SSD1306 | |
import Image | |
import ImageDraw | |
import ImageFont | |
# Rotary | |
import gaugette.rotary_encoder | |
import gaugette.switch | |
# Audio | |
from pygame import mixer | |
# Sonar | |
import RPi.GPIO as GPIO | |
# ------------------------------------------------- | |
# DISPLAY pin config | |
RST = 24 | |
DC = 23 | |
SPI_PORT = 0 | |
SPI_DEVICE = 0 | |
# DISPLAY initialize | |
disp = Adafruit_SSD1306.SSD1306_128_32(rst=RST, dc=DC, spi=SPI.SpiDev(SPI_PORT, SPI_DEVICE, max_speed_hz=8000000)) | |
disp.begin() | |
disp.clear() | |
disp.display() | |
display_width = disp.width | |
display_height = disp.height | |
volume_image = Image.new('1', (display_width, display_height)) | |
# Get drawing object to draw on image. | |
volume_draw = ImageDraw.Draw(volume_image) | |
# Alternatively load a different format image, resize it, and convert to 1 bit color. | |
image = Image.open('images/poop_screen_sound.gif').convert('1') | |
# Display image. | |
disp.image(image) | |
disp.display() | |
# AUDIO setup | |
allfiles = glob.glob("/home/pi/dumpdamper/audio/*") | |
filenames = [] | |
for f in allfiles: | |
filenames.append(re.search(r"[^\/]*(?=\.[^.]+($|\?))",f)) | |
audio_files = zip(allfiles, filenames) | |
print audio_files | |
mixer.pre_init(44100, -16, 2, 2048) # setup mixer to avoid sound lag | |
mixer.init() #initialize pygame | |
try: | |
sound_hello = mixer.Sound(open('confirm.wav')) | |
except: | |
raise UserWarning, "failed" | |
sound_hello.set_volume(0.5) | |
A_PIN = 7 | |
B_PIN = 9 | |
SW_PIN = 8 | |
encoder = gaugette.rotary_encoder.RotaryEncoder.Worker(A_PIN, B_PIN) | |
encoder.start() | |
switch = gaugette.switch.Switch(SW_PIN) | |
last_state = None | |
# SONAR setup | |
shutdown = False | |
GPIO.setmode(GPIO.BCM) | |
# Define GPIO to use on Pi | |
GPIO_TRIGECHO = 18 | |
# Set pins as output and input | |
GPIO.setup(GPIO_TRIGECHO,GPIO.OUT) # Initial state as output | |
# Set trigger to False (Low) | |
GPIO.output(GPIO_TRIGECHO, False) | |
def knob_handler(): | |
while shutdown!=True: | |
global last_state | |
delta = encoder.get_delta() | |
if delta!=0: | |
print "rotate %d" % delta | |
if delta>0: | |
volume_up(delta) | |
if delta<0: | |
volume_down(delta) | |
sw_state = switch.get_state() | |
if sw_state != last_state: | |
print "switch %d" % sw_state | |
last_state = sw_state | |
def measure(): | |
# This function measures a distance | |
# Pulse the trigger/echo line to initiate a measurement | |
GPIO.output(GPIO_TRIGECHO, True) | |
time.sleep(0.00001) | |
GPIO.output(GPIO_TRIGECHO, False) | |
#ensure start time is set in case of very quick return | |
start = time.time() | |
# set line to input to check for start of echo response | |
GPIO.setup(GPIO_TRIGECHO, GPIO.IN) | |
while GPIO.input(GPIO_TRIGECHO)==0: | |
start = time.time() | |
# Wait for end of echo response | |
while GPIO.input(GPIO_TRIGECHO)==1: | |
stop = time.time() | |
GPIO.setup(GPIO_TRIGECHO, GPIO.OUT) | |
GPIO.output(GPIO_TRIGECHO, False) | |
elapsed = stop-start | |
distance = (elapsed * 34300)/2.0 | |
time.sleep(0.1) | |
return distance | |
def measure_average(): | |
# This function takes n measurements ignoring any rogue values | |
# returns the average. | |
MAX_TRIES = 5 # Attempts to get consistent results | |
MEASURE_COUNT = 3 # Nof raw measures in each attempt | |
CHECK = 2.0 # tolerance in cm between measurements | |
midpoint = MEASURE_COUNT / 2 | |
for tries in range(MAX_TRIES): | |
distances = [] | |
for i in range(MEASURE_COUNT): | |
distances.append(measure()) | |
distances.sort() | |
measureOK = True | |
for i in range(MEASURE_COUNT - 1): | |
if abs(distances[i] - distances[midpoint]) > CHECK: | |
measureOK = False | |
break | |
if measureOK: | |
break | |
print "Inconsistent results. ", distances, " Retrying.." | |
if measureOK: | |
distance = sum(distances) / len(distances) | |
else: | |
print "Inconsistent after retries. Best guess value" | |
distance = distances[len(distances)/2] | |
return distance | |
def sonar_handler(): | |
print "sonar handler starting" | |
distance = measure_average() | |
print " Distance : %.1f cm" % distance | |
if shutdown!=True: | |
threading.Timer(3, sonar_handler).start() | |
def display_volume(): | |
disp.clear() | |
volumePerc = (sound_hello.get_volume()/1)*display_width | |
if volumePerc < 2: | |
volumePerc = 2 | |
print volumePerc | |
volume_draw.rectangle((0,0,display_width,display_height), outline=0, fill=0) | |
volume_draw.rectangle((0,0,volumePerc,display_height), outline=0, fill=255) | |
disp.image(volume_image) | |
disp.display(); | |
def volume_up(amt): | |
volume_change = sound_hello.get_volume() + abs(amt)*0.0078125 | |
if volume_change >= 1: | |
volume_change = 1 | |
sound_hello.set_volume(volume_change) | |
display_volume() | |
def volume_down(amt): | |
volume_change = sound_hello.get_volume() - abs(amt)*0.0078125 | |
if volume_change <= 0: | |
volume_change = 0 | |
sound_hello.set_volume(volume_change) | |
display_volume() | |
try: | |
sonar_handler() | |
knob_thread = threading.Thread(target=knob_handler) | |
knob_thread.start() | |
# knob_handler() | |
except KeyboardInterrupt: | |
# User pressed CTRL-C | |
# Reset GPIO settings | |
shutdown=True | |
knob_thread.join() | |
GPIO.cleanup() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment