martin
/
mopidy-raspberrypi-gpio
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Initial works to add support for rotary encoders.

rotencoder
Martin Felis 2021-03-13 14:54:59 +01:00
parent b4f6137f65
commit 3f0a4a33a5
6 changed files with 149 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
mopidy_raspberry_gpio/__init__.py
View File

@ -4,6 +4,7 @@ import pathlib
from mopidy import config, ext
from .pinconfig import PinConfig
from .rotencoder import RotEncoder
__version__ = "0.0.2"

32
mopidy_raspberry_gpio/frontend.py
View File

@ -5,7 +5,6 @@ from mopidy import core
logger = logging.getLogger(__name__)
class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
def __init__(self, config, core):
super().__init__()
@ -14,6 +13,7 @@ class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
self.core = core
self.config = config["raspberry-gpio"]
self.pin_settings = {}
self.rot_encoders = {}
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
@ -27,12 +27,25 @@ class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
if settings is None:
continue
pull = GPIO.PUD_UP
edge = GPIO.FALLING
if settings.active == "active_high":
pull = GPIO.PUD_DOWN
edge = GPIO.RISING
if 'rotenc_id' in settings.options:
edge = GPIO.BOTH
rotenc_id = self.options['rotenc_id']
encoder = None
if rotenc_id in self.rot_encoders:
encoder = self.rot_encoders[rotenc_id]
else:
encoder = RotEncoder(rotenc_id)
self.rot_encoders[rotenc_id] = encoder
encoder.add_pin(pin, settings.event)
settings.rot_encoder = encoder
GPIO.setup(pin, GPIO.IN, pull_up_down=pull)
GPIO.add_event_detect(
@ -44,17 +57,24 @@ class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
self.pin_settings[pin] = settings
# TODO validate all self.rot_encoders have two pins
def gpio_event(self, pin):
settings = self.pin_settings[pin]
self.dispatch_input(settings)
event = settings.event
if settings.rot_encoder:
event = settings.rot_encoder.get_event()
def dispatch_input(self, settings):
handler_name = f"handle_{settings.event}"
if event:
self.dispatch_input(event, settings.options)
def dispatch_input(self, event, options):
handler_name = f"handle_{event}"
try:
getattr(self, handler_name)(settings.options)
getattr(self, handler_name)(options)
except AttributeError:
raise RuntimeError(
f"Could not find input handler for event: {settings.event}"
f"Could not find input handler for event: {event}"
)
def handle_play_pause(self, config):

4
mopidy_raspberry_gpio/pinconfig.py
View File

@ -13,7 +13,7 @@ class ValidList(list):
class PinConfig(config.ConfigValue):
tuple_pinconfig = namedtuple(
"PinConfig", ("event", "active", "bouncetime", "options")
"PinConfig", ("event", "active", "bouncetime", "options", "rot_encoder")
)
valid_events = ValidList(
@ -62,7 +62,7 @@ class PinConfig(config.ConfigValue):
key, value = option.split("=")
options[key] = value
return self.tuple_pinconfig(event, active, bouncetime, options)
return self.tuple_pinconfig(event, active, bouncetime, options, None)
def serialize(self, value, display=False):
if value is None:

48
mopidy_raspberry_gpio/rotencoder.py Normal file
View File

@ -0,0 +1,48 @@
import logging
logger = logging.getLogger(__name__)
class RotEncoder:
def __init__(self, rot_id):
self.id = rot_id
self.pins = []
self.events = []
self.state_map = {
((False, False), (False,True)): 0,
((False, False), (True,False)): 1,
((False, True), (True,True)): 0,
((False, True), (False,False)): 1,
((True, False), (False,False)): 0,
((True, False), (True,True)): 1,
((True, True), (True, False)): 0,
((True, True), (False,True)): 1
}
def add_pin(self, pin, event):
if len(self.pins) == 2:
raise RuntimeError (f"Too many pins for rotary encoder {self.id}!")
self.pins.append(pin)
self.events.append(event)
def get_state(self):
import RPi.GPIO as GPIO
level0 = GPIO.input(self.pins[0])
level1 = GPIO.input(self.pins[1])
return (level0, level1)
def get_direction (self, current, new):
return self.state_map[(current, new)]
def get_event(self):
next_state = self.get_state()
event = None
try:
direction = self.get_direction(self.state, next_state)
event = self.events[direction]
except KeyError:
pass
self.state = next_state
return event

16
tests/test_frontend.py
View File

@ -61,7 +61,7 @@ def test_frontend_handler_dispatch_play_pause():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("play_pause,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -78,7 +78,7 @@ def test_frontend_handler_dispatch_play_stop():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("play_stop,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -95,7 +95,7 @@ def test_frontend_handler_dispatch_next():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("next,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -112,7 +112,7 @@ def test_frontend_handler_dispatch_prev():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("prev,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -129,7 +129,7 @@ def test_frontend_handler_dispatch_volume_up():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("volume_up,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -146,7 +146,7 @@ def test_frontend_handler_dispatch_volume_down():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("volume_down,active_low,30")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -163,7 +163,7 @@ def test_frontend_handler_dispatch_volume_up_custom_step():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("volume_up,active_low,30,step=1")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()
@ -180,7 +180,7 @@ def test_frontend_handler_dispatch_volume_down_custom_step():
schema = ext.get_config_schema()
settings = schema["bcm1"].deserialize("volume_down,active_low,30,step=1")
frontend.dispatch_input(settings)
frontend.dispatch_input(settings.event, settings.options)
stop_mopidy_core()

64
tests/test_rotencoder.py Normal file
View File

@ -0,0 +1,64 @@
import sys
import pytest
import unittest
from mopidy_raspberry_gpio import RotEncoder
from unittest.mock import patch, MagicMock
MockRPi = MagicMock()
modules = {
"RPi": MockRPi,
"RPi.GPIO": MockRPi.GPIO
}
patcher = patch.dict("sys.modules", modules)
patcher.start()
class RotEncoderTests(unittest.TestCase):
def test_rotenc_init(self):
rot_enc = RotEncoder("vol")
self.assertTrue (rot_enc.id == "vol")
self.assertTrue (((False,False), (False, True)) in rot_enc.state_map)
def test_get_direction(self):
rot_enc = RotEncoder("vol")
rot_enc.add_pin(123, "vol_up")
rot_enc.add_pin(124, "vol_down")
dir_down = rot_enc.get_direction ((False, False), (False, True))
dir_up = rot_enc.get_direction ((False, False), (True, False))
self.assertEqual (dir_up, 1)
self.assertEqual (dir_down, 0)
def test_add_pin_invalid(self):
rot_enc = RotEncoder("vol")
rot_enc.add_pin(123, "vol_up")
rot_enc.add_pin(124, "vol_down")
with self.assertRaises(RuntimeError) as cm:
rot_enc.add_pin(124, "vol_down")
@patch("RPi.GPIO.input")
def test_get_event(self, patched_input):
# Always return False for GPIO.input
patched_input.return_value = False
rot_enc = RotEncoder("vol")
rot_enc.add_pin(123, "vol_down") # dir 0 => vol_down
rot_enc.add_pin(124, "vol_up") # dir 1 => vol_up
# from False,True to False,False => dir 1
rot_enc.state = (False, True)
event = rot_enc.get_event()
self.assertEqual(event, "vol_up")
# from True,False to False,False => dir 0
rot_enc.state = (True, False)
event = rot_enc.get_event()
self.assertEqual(event, "vol_down")
# from True,True to False,False => None
rot_enc.state = (True, True)
event = rot_enc.get_event()
self.assertEqual(event, None)