So, I'm making what's turning out to be a pretty difficult project for me to code. In short, I'm making a speaker that will play audio when you get near it (thanks to a sonar), with a knob interface (rotary encoder) that allows you to change volume and/or track. And I’m writing it all in Python.
But, here's the rub: because it's a rotary encoder, I have to be constantly checking for rotation changes, and with the sonar, I have to constantly be checking to see if anybody's near. So, both these constant loops need to be running at essentially the same time and in the background.
I've done a fair amount of research into multithreading, multiprocessing, and asynchronous code and I can't seem to find a solution that works correctly.
For the rotary encoder, I'm using code I found here (Rotary Encoder library for RPI), and for the sonar, I’m attempting to use this code I found on the RPI forums.
The problem is, when I run the sonar code from above in a process or thread, it seems to misread and give a lot of crap data, so it’s not trustworthy. When it’s alone, it’s fine.
Right now my most recent attempt has me starting a thread timer for the sonar that restarts every 3 seconds.
def sonar_handler():
print "sonar handler starting"
distance = measure_average()
print " Distance : %.1f cm" % distance
threading.Timer(3, sonar_handler).start()
BUT, I have to have the knob working on a constant loop on its own thread, and when I add this into the loop, I get constant bad readings from my sonar:
def knob_handler():
while 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
try:
knob_thread = threading.Thread(target=knob_handler)
knob_thread.start()
In case it's helpful, here's the code that the sonar_handler() is referring to:
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
At the end of the day, is there an easier, more straightforward way to do this?
