savetheclocktower/README.md

## README.md

      
    Raw
  

              README.md
            
          
    Using a rotary encoder as a volume control

On my RetroPie machine I wanted a hardware volume knob — the games I play use a handful of emulators, and there's no unified software interface for controlling the volume. The speakers I got for my cabinet are great, but don't have their own hardware volume knob. So with a bunch of googling and trial and error, I figured out what I need to pull this off: a rotary encoder and a daemon that listens for the signals it sends.
Rotary encoder

A rotary encoder is like the standard potentiometer (i.e., analog volume knob) we all know, except (a) you can keep turning it in either direction for as long as you want, and thus (b) it talks to the RPi differently than a potentiometer would.
I picked up this one from Adafruit, but there are plenty others available. This rotary encoder also lets you push the knob in and treats that like a button press, so I figured that would be useful for toggling mute on and off.
So we've got 5 wires to hook up: three for the knob part (A, B, and ground), and two for the button part (common and ground). Here's how I hooked them up (reference):


Description
BCM #
Board #


knob A
GPIO 26
37


knob B
GPIO 19
35


knob ground
ground pin below GPIO 26
39


button common
GPIO 13
33


button ground
ground pin opposite GPIO 13
34


You can use whichever pins you want; just update the script if you change them.
Volume daemon

Since the GPIO pins are just for arbitrary signals, you need a script on the Pi that knows what to do with them. The builtin GPIO library for the Pi will do nicely for this.
You'll see the script below, but here's how it works: it listens on the specified pins, and when the knob is turned one way or another, it uses the states of the A and B pins to figure out whether the knob was turned to the left or to the right. That way it knows whether to increase or decrease the system volume in response, which it does with the command-line program amixer.
First, make sure amixer is present and install it if it isn’t.
which amixer || sudo apt-get install alsa-utils
(If you’re not using the default analog audio output, consult @thijstriemstra’s comment below for some additional steps that you may or may not need to do.)
Create a bin directory in your pi folder if it doesn't exist already, then drop the script below into it.
mkdir ~/bin

If it's not there yet, I'd also put /home/pi/bin somewhere in your PATH:
echo $PATH
# don't see "/home/pi/bin" there? then run...
echo "export PATH=$HOME/bin:$PATH" >> ~/.bashrc
# ...and restart your shell

Then drop the script into the /home/pi/bin folder:
nano ~/bin/monitor-volume # (or however you do it)
chmod +x ~/bin/monitor-volume

You can run this script in the foreground just to test it out. Edit the script and temporarily change DEBUG to True so that you can see what's going on, then simply run it with monitor-volume. When you turn or press the knob, the script should report what it's doing. (Make sure that the volume increases rather than decreases when you turn it to the right, and if it doesn't, swap your A and B pins, or just swap their numbers in the script.)
You should also play around with the constants defined at the top of the script. Naturally, if you picked other pins, you'll want to tell the script which GPIO pins to use. If your rotary encoder doesn't act like a button, or if you didn't hook up the button and don't care about it, you can set GPIO_BUTTON to None. But you may also want to change the minimum and maximum volumes (they're percentages, so they should be between 1 and 100) and the increment (how many percentage points the volume increases/decreases with each "click" of the knob).
If it's working the way you want, you can proceed to the next step: running monitor-volume automatically in the background whenever your Pi starts.
Creating a systemd service

NOTE: If you're on a version of Raspbian before Jessie, these instructions won't work for you. Hopefully someone can pipe up with a version of this for init.d in the comments.
systemd is the new way of managing startup daemons in Raspbian Jessie. Because it's new, there's not much RPi-specific documentation on it, and to find out how to use it you have to sift through a bunch of Google results from people who hate systemd and wish it didn't exist. After much trial and error, here's what worked for me:
nano ~/monitor-volume.service
# paste in the contents of monitor-volume.service, save/exit nano
chmod +x ~/monitor-volume.service
sudo mv ~/monitor-volume.service /etc/systemd/system
sudo systemctl enable monitor-volume
sudo systemctl start monitor-volume

If that worked right, then you just told Raspbian to start up that script in the background on every boot (enable), and also to start it right now (start). At this point, and on every boot after this, your volume knob should Just Work.
FAQ

This didn't work!
I got this working on an RPi3 running RetroPie 3.x, so all I can say is “works on my machine.” Some things to try:


You might not have Python 3; if which python3 turns up nothing, try this:
sudo apt-get install python3 python3-rpi.gpio


I've heard that in earlier versions of Raspbian, the pi user isn't automatically allowed to access the GPIO pins, so you need to run scripts like this as root.  If you're running into permissions errors when you try to run the script from your shell, then that's your problem, most likely. There's no particular reason why you shouldn't run this script as root, except on the general principle that you shouldn't really trust code that you didn't write. Make good choices and have backups.


I might have made a typo in the gist. Wouldn't be the first time.


If you run into trouble, leave a comment and the internet can help you figure it out.

  
## monitor-volume
#!/usr/bin/env python3

"""
The daemon responsible for changing the volume in response to a turn or press
of the volume knob.

The volume knob is a rotary encoder. It turns infinitely in either direction.
Turning it to the right will increase the volume; turning it to the left will
decrease the volume. The knob can also be pressed like a button in order to
turn muting on or off.

The knob uses two GPIO pins and we need some extra logic to decode it. The
button we can just treat like an ordinary button. Rather than poll
constantly, we use threads and interrupts to listen on all three pins in one
script.
"""

import os
import signal
import subprocess
import sys
import threading

from RPi import GPIO
from queue import Queue

DEBUG = False

# SETTINGS
# ========

# The two pins that the encoder uses (BCM numbering).
GPIO_A = 26
GPIO_B = 19

# The pin that the knob's button is hooked up to. If you have no button, set
# this to None.
GPIO_BUTTON = 13

# The minimum and maximum volumes, as percentages.
#
# The default max is less than 100 to prevent distortion. The default min is
# greater than zero because if your system is like mine, sound gets
# completely inaudible _long_ before 0%. If you've got a hardware amp or
# serious speakers or something, your results will vary.
VOLUME_MIN = 60
VOLUME_MAX = 96

# The amount you want one click of the knob to increase or decrease the
# volume. I don't think that non-integer values work here, but you're welcome
# to try.
VOLUME_INCREMENT = 1

# (END SETTINGS)
#


# When the knob is turned, the callback happens in a separate thread. If
# those turn callbacks fire erratically or out of order, we'll get confused
# about which direction the knob is being turned, so we'll use a queue to
# enforce FIFO. The callback will push onto a queue, and all the actual
# volume-changing will happen in the main thread.
QUEUE = Queue()

# When we put something in the queue, we'll use an event to signal to the
# main thread that there's something in there. Then the main thread will
# process the queue and reset the event. If the knob is turned very quickly,
# this event loop will fall behind, but that's OK because it consumes the
# queue completely each time through the loop, so it's guaranteed to catch up.
EVENT = threading.Event()

def debug(str):
  if not DEBUG:
    return
  print(str)

class RotaryEncoder:
  """
  A class to decode mechanical rotary encoder pulses.

  Ported to RPi.GPIO from the pigpio sample here:
  http://abyz.co.uk/rpi/pigpio/examples.html
  """

  def __init__(self, gpioA, gpioB, callback=None, buttonPin=None, buttonCallback=None):
    """
    Instantiate the class. Takes three arguments: the two pin numbers to
    which the rotary encoder is connected, plus a callback to run when the
    switch is turned.

    The callback receives one argument: a `delta` that will be either 1 or -1.
    One of them means that the dial is being turned to the right; the other
    means that the dial is being turned to the left. I'll be damned if I know
    yet which one is which.
    """

    self.lastGpio = None
    self.gpioA    = gpioA
    self.gpioB    = gpioB
    self.callback = callback

    self.gpioButton     = buttonPin
    self.buttonCallback = buttonCallback

    self.levA = 0
    self.levB = 0

    GPIO.setmode(GPIO.BCM)
    GPIO.setup(self.gpioA, GPIO.IN, pull_up_down=GPIO.PUD_UP)
    GPIO.setup(self.gpioB, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    GPIO.add_event_detect(self.gpioA, GPIO.BOTH, self._callback)
    GPIO.add_event_detect(self.gpioB, GPIO.BOTH, self._callback)

    if self.gpioButton:
      GPIO.setup(self.gpioButton, GPIO.IN, pull_up_down=GPIO.PUD_UP)
      GPIO.add_event_detect(self.gpioButton, GPIO.FALLING, self._buttonCallback, bouncetime=500)


  def destroy(self):
    GPIO.remove_event_detect(self.gpioA)
    GPIO.remove_event_detect(self.gpioB)
    GPIO.cleanup()

  def _buttonCallback(self, channel):
    self.buttonCallback(GPIO.input(channel))

  def _callback(self, channel):
    level = GPIO.input(channel)
    if channel == self.gpioA:
      self.levA = level
    else:
      self.levB = level

    # Debounce.
    if channel == self.lastGpio:
      return

    # When both inputs are at 1, we'll fire a callback. If A was the most
    # recent pin set high, it'll be forward, and if B was the most recent pin
    # set high, it'll be reverse.
    self.lastGpio = channel
    if channel == self.gpioA and level == 1:
      if self.levB == 1:
        self.callback(1)
    elif channel == self.gpioB and level == 1:
      if self.levA == 1:
        self.callback(-1)

class VolumeError(Exception):
  pass

class Volume:
  """
  A wrapper API for interacting with the volume settings on the RPi.
  """
  MIN = VOLUME_MIN
  MAX = VOLUME_MAX
  INCREMENT = VOLUME_INCREMENT

  def __init__(self):
    # Set an initial value for last_volume in case we're muted when we start.
    self.last_volume = self.MIN
    self._sync()

  def up(self):
    """
    Increases the volume by one increment.
    """
    return self.change(self.INCREMENT)

  def down(self):
    """
    Decreases the volume by one increment.
    """
    return self.change(-self.INCREMENT)

  def change(self, delta):
    v = self.volume + delta
    v = self._constrain(v)
    return self.set_volume(v)

  def set_volume(self, v):
    """
    Sets volume to a specific value.
    """
    self.volume = self._constrain(v)
    output = self.amixer("set 'PCM' unmute {}%".format(v))
    self._sync(output)
    return self.volume

  def toggle(self):
    """
    Toggles muting between on and off.
    """
    if self.is_muted:
      output = self.amixer("set 'PCM' unmute")
    else:
      # We're about to mute ourselves, so we should remember the last volume
      # value we had because we'll want to restore it later.
      self.last_volume = self.volume
      output = self.amixer("set 'PCM' mute")

    self._sync(output)
    if not self.is_muted:
      # If we just unmuted ourselves, we should restore whatever volume we
      # had previously.
      self.set_volume(self.last_volume)
    return self.is_muted

  def status(self):
    if self.is_muted:
      return "{}% (muted)".format(self.volume)
    return "{}%".format(self.volume)

  # Read the output of `amixer` to get the system volume and mute state.
  #
  # This is designed not to do much work because it'll get called with every
  # click of the knob in either direction, which is why we're doing simple
  # string scanning and not regular expressions.
  def _sync(self, output=None):
    if output is None:
      output = self.amixer("get 'PCM'")

    lines = output.readlines()
    if DEBUG:
      strings = [line.decode('utf8') for line in lines]
      debug("OUTPUT:")
      debug("".join(strings))
    last = lines[-1].decode('utf-8')

    # The last line of output will have two values in square brackets. The
    # first will be the volume (e.g., "[95%]") and the second will be the
    # mute state ("[off]" or "[on]").
    i1 = last.rindex('[') + 1
    i2 = last.rindex(']')

    self.is_muted = last[i1:i2] == 'off'

    i1 = last.index('[') + 1
    i2 = last.index('%')
    # In between these two will be the percentage value.
    pct = last[i1:i2]

    self.volume = int(pct)

  # Ensures the volume value is between our minimum and maximum.
  def _constrain(self, v):
    if v < self.MIN:
      return self.MIN
    if v > self.MAX:
      return self.MAX
    return v

  def amixer(self, cmd):
    p = subprocess.Popen("amixer {}".format(cmd), shell=True, stdout=subprocess.PIPE)
    code = p.wait()
    if code != 0:
      raise VolumeError("Unknown error")
      sys.exit(0)

    return p.stdout


if __name__ == "__main__":

  gpioA = GPIO_A
  gpioB = GPIO_B
  gpioButton = GPIO_BUTTON

  v = Volume()

  def on_press(value):
    v.toggle()
    print("Toggled mute to: {}".format(v.is_muted))
    EVENT.set()

  # This callback runs in the background thread. All it does is put turn
  # events into a queue and flag the main thread to process them. The
  # queueing ensures that we won't miss anything if the knob is turned
  # extremely quickly.
  def on_turn(delta):
    QUEUE.put(delta)
    EVENT.set()

  def consume_queue():
    while not QUEUE.empty():
      delta = QUEUE.get()
      handle_delta(delta)

  def handle_delta(delta):
    if v.is_muted:
      debug("Unmuting")
      v.toggle()
    if delta == 1:
      vol = v.up()
    else:
      vol = v.down()
    print("Set volume to: {}".format(vol))

  def on_exit(a, b):
    print("Exiting...")
    encoder.destroy()
    sys.exit(0)

  debug("Volume knob using pins {} and {}".format(gpioA, gpioB))

  if gpioButton != None:
    debug("Volume button using pin {}".format(gpioButton))

  debug("Initial volume: {}".format(v.volume))

  encoder = RotaryEncoder(GPIO_A, GPIO_B, callback=on_turn, buttonPin=GPIO_BUTTON, buttonCallback=on_press)
  signal.signal(signal.SIGINT, on_exit)

  while True:
    # This is the best way I could come up with to ensure that this script
    # runs indefinitely without wasting CPU by polling. The main thread will
    # block quietly while waiting for the event to get flagged. When the knob
    # is turned we're able to respond immediately, but when it's not being
    # turned we're not looping at all.
    #
    # The 1200-second (20 minute) timeout is a hack; for some reason, if I
    # don't specify a timeout, I'm unable to get the SIGINT handler above to
    # work properly. But if there is a timeout set, even if it's a very long
    # timeout, then Ctrl-C works as intended. No idea why.
    EVENT.wait(1200)
    consume_queue()
    EVENT.clear()

## monitor-volume.service
[Unit]
Description=Volume knob monitor

[Service]
User=pi
Group=pi
ExecStart=/home/pi/bin/monitor-volume

[Install]
WantedBy=multi-user.target
Description	BCM #	Board #
knob A	GPIO 26	37
knob B	GPIO 19	35
knob ground	ground pin below GPIO 26	39
button common	GPIO 13	33
button ground	ground pin opposite GPIO 13	34
	#!/usr/bin/env python3

	"""
	The daemon responsible for changing the volume in response to a turn or press
	of the volume knob.

	The volume knob is a rotary encoder. It turns infinitely in either direction.
	Turning it to the right will increase the volume; turning it to the left will
	decrease the volume. The knob can also be pressed like a button in order to
	turn muting on or off.

	The knob uses two GPIO pins and we need some extra logic to decode it. The
	button we can just treat like an ordinary button. Rather than poll
	constantly, we use threads and interrupts to listen on all three pins in one
	script.
	"""

	import os
	import signal
	import subprocess
	import sys
	import threading

	from RPi import GPIO
	from queue import Queue

	DEBUG = False

	# SETTINGS
	# ========

	# The two pins that the encoder uses (BCM numbering).
	GPIO_A = 26
	GPIO_B = 19

	# The pin that the knob's button is hooked up to. If you have no button, set
	# this to None.
	GPIO_BUTTON = 13

	# The minimum and maximum volumes, as percentages.
	#
	# The default max is less than 100 to prevent distortion. The default min is
	# greater than zero because if your system is like mine, sound gets
	# completely inaudible _long_ before 0%. If you've got a hardware amp or
	# serious speakers or something, your results will vary.
	VOLUME_MIN = 60
	VOLUME_MAX = 96

	# The amount you want one click of the knob to increase or decrease the
	# volume. I don't think that non-integer values work here, but you're welcome
	# to try.
	VOLUME_INCREMENT = 1

	# (END SETTINGS)
	#


	# When the knob is turned, the callback happens in a separate thread. If
	# those turn callbacks fire erratically or out of order, we'll get confused
	# about which direction the knob is being turned, so we'll use a queue to
	# enforce FIFO. The callback will push onto a queue, and all the actual
	# volume-changing will happen in the main thread.
	QUEUE = Queue()

	# When we put something in the queue, we'll use an event to signal to the
	# main thread that there's something in there. Then the main thread will
	# process the queue and reset the event. If the knob is turned very quickly,
	# this event loop will fall behind, but that's OK because it consumes the
	# queue completely each time through the loop, so it's guaranteed to catch up.
	EVENT = threading.Event()

	def debug(str):
	if not DEBUG:
	return
	print(str)

	class RotaryEncoder:
	"""
	A class to decode mechanical rotary encoder pulses.

	Ported to RPi.GPIO from the pigpio sample here:
	http://abyz.co.uk/rpi/pigpio/examples.html
	"""

	def __init__(self, gpioA, gpioB, callback=None, buttonPin=None, buttonCallback=None):
	"""
	Instantiate the class. Takes three arguments: the two pin numbers to
	which the rotary encoder is connected, plus a callback to run when the
	switch is turned.

	The callback receives one argument: a `delta` that will be either 1 or -1.
	One of them means that the dial is being turned to the right; the other
	means that the dial is being turned to the left. I'll be damned if I know
	yet which one is which.
	"""

	self.lastGpio = None
	self.gpioA = gpioA
	self.gpioB = gpioB
	self.callback = callback

	self.gpioButton = buttonPin
	self.buttonCallback = buttonCallback

	self.levA = 0
	self.levB = 0

	GPIO.setmode(GPIO.BCM)
	GPIO.setup(self.gpioA, GPIO.IN, pull_up_down=GPIO.PUD_UP)
	GPIO.setup(self.gpioB, GPIO.IN, pull_up_down=GPIO.PUD_UP)

	GPIO.add_event_detect(self.gpioA, GPIO.BOTH, self._callback)
	GPIO.add_event_detect(self.gpioB, GPIO.BOTH, self._callback)

	if self.gpioButton:
	GPIO.setup(self.gpioButton, GPIO.IN, pull_up_down=GPIO.PUD_UP)
	GPIO.add_event_detect(self.gpioButton, GPIO.FALLING, self._buttonCallback, bouncetime=500)


	def destroy(self):
	GPIO.remove_event_detect(self.gpioA)
	GPIO.remove_event_detect(self.gpioB)
	GPIO.cleanup()

	def _buttonCallback(self, channel):
	self.buttonCallback(GPIO.input(channel))

	def _callback(self, channel):
	level = GPIO.input(channel)
	if channel == self.gpioA:
	self.levA = level
	else:
	self.levB = level

	# Debounce.
	if channel == self.lastGpio:
	return

	# When both inputs are at 1, we'll fire a callback. If A was the most
	# recent pin set high, it'll be forward, and if B was the most recent pin
	# set high, it'll be reverse.
	self.lastGpio = channel
	if channel == self.gpioA and level == 1:
	if self.levB == 1:
	self.callback(1)
	elif channel == self.gpioB and level == 1:
	if self.levA == 1:
	self.callback(-1)

	class VolumeError(Exception):
	pass

	class Volume:
	"""
	A wrapper API for interacting with the volume settings on the RPi.
	"""
	MIN = VOLUME_MIN
	MAX = VOLUME_MAX
	INCREMENT = VOLUME_INCREMENT

	def __init__(self):
	# Set an initial value for last_volume in case we're muted when we start.
	self.last_volume = self.MIN
	self._sync()

	def up(self):
	"""
	Increases the volume by one increment.
	"""
	return self.change(self.INCREMENT)

	def down(self):
	"""
	Decreases the volume by one increment.
	"""
	return self.change(-self.INCREMENT)

	def change(self, delta):
	v = self.volume + delta
	v = self._constrain(v)
	return self.set_volume(v)

	def set_volume(self, v):
	"""
	Sets volume to a specific value.
	"""
	self.volume = self._constrain(v)
	output = self.amixer("set 'PCM' unmute {}%".format(v))
	self._sync(output)
	return self.volume

	def toggle(self):
	"""
	Toggles muting between on and off.
	"""
	if self.is_muted:
	output = self.amixer("set 'PCM' unmute")
	else:
	# We're about to mute ourselves, so we should remember the last volume
	# value we had because we'll want to restore it later.
	self.last_volume = self.volume
	output = self.amixer("set 'PCM' mute")

	self._sync(output)
	if not self.is_muted:
	# If we just unmuted ourselves, we should restore whatever volume we
	# had previously.
	self.set_volume(self.last_volume)
	return self.is_muted

	def status(self):
	if self.is_muted:
	return "{}% (muted)".format(self.volume)
	return "{}%".format(self.volume)

	# Read the output of `amixer` to get the system volume and mute state.
	#
	# This is designed not to do much work because it'll get called with every
	# click of the knob in either direction, which is why we're doing simple
	# string scanning and not regular expressions.
	def _sync(self, output=None):
	if output is None:
	output = self.amixer("get 'PCM'")

	lines = output.readlines()
	if DEBUG:
	strings = [line.decode('utf8') for line in lines]
	debug("OUTPUT:")
	debug("".join(strings))
	last = lines[-1].decode('utf-8')

	# The last line of output will have two values in square brackets. The
	# first will be the volume (e.g., "[95%]") and the second will be the
	# mute state ("[off]" or "[on]").
	i1 = last.rindex('[') + 1
	i2 = last.rindex(']')

	self.is_muted = last[i1:i2] == 'off'

	i1 = last.index('[') + 1
	i2 = last.index('%')
	# In between these two will be the percentage value.
	pct = last[i1:i2]

	self.volume = int(pct)

	# Ensures the volume value is between our minimum and maximum.
	def _constrain(self, v):
	if v < self.MIN:
	return self.MIN
	if v > self.MAX:
	return self.MAX
	return v

	def amixer(self, cmd):
	p = subprocess.Popen("amixer {}".format(cmd), shell=True, stdout=subprocess.PIPE)
	code = p.wait()
	if code != 0:
	raise VolumeError("Unknown error")
	sys.exit(0)

	return p.stdout


	if __name__ == "__main__":

	gpioA = GPIO_A
	gpioB = GPIO_B
	gpioButton = GPIO_BUTTON

	v = Volume()

	def on_press(value):
	v.toggle()
	print("Toggled mute to: {}".format(v.is_muted))
	EVENT.set()

	# This callback runs in the background thread. All it does is put turn
	# events into a queue and flag the main thread to process them. The
	# queueing ensures that we won't miss anything if the knob is turned
	# extremely quickly.
	def on_turn(delta):
	QUEUE.put(delta)
	EVENT.set()

	def consume_queue():
	while not QUEUE.empty():
	delta = QUEUE.get()
	handle_delta(delta)

	def handle_delta(delta):
	if v.is_muted:
	debug("Unmuting")
	v.toggle()
	if delta == 1:
	vol = v.up()
	else:
	vol = v.down()
	print("Set volume to: {}".format(vol))

	def on_exit(a, b):
	print("Exiting...")
	encoder.destroy()
	sys.exit(0)

	debug("Volume knob using pins {} and {}".format(gpioA, gpioB))

	if gpioButton != None:
	debug("Volume button using pin {}".format(gpioButton))

	debug("Initial volume: {}".format(v.volume))

	encoder = RotaryEncoder(GPIO_A, GPIO_B, callback=on_turn, buttonPin=GPIO_BUTTON, buttonCallback=on_press)
	signal.signal(signal.SIGINT, on_exit)

	while True:
	# This is the best way I could come up with to ensure that this script
	# runs indefinitely without wasting CPU by polling. The main thread will
	# block quietly while waiting for the event to get flagged. When the knob
	# is turned we're able to respond immediately, but when it's not being
	# turned we're not looping at all.
	#
	# The 1200-second (20 minute) timeout is a hack; for some reason, if I
	# don't specify a timeout, I'm unable to get the SIGINT handler above to
	# work properly. But if there is a timeout set, even if it's a very long
	# timeout, then Ctrl-C works as intended. No idea why.
	EVENT.wait(1200)
	consume_queue()
	EVENT.clear()
	[Unit]
	Description=Volume knob monitor

	[Service]
	User=pi
	Group=pi
	ExecStart=/home/pi/bin/monitor-volume

	[Install]
	WantedBy=multi-user.target