Compare commits
6 Commits
master
...
rotencoder
Author | SHA1 | Date |
---|---|---|
Martin Felis | 651da1c896 | |
Martin Felis | 5c97976f15 | |
Martin Felis | b3778d8cd2 | |
Martin Felis | a65b65ab13 | |
Martin Felis | 97592cb790 | |
Martin Felis | 3f0a4a33a5 |
|
@ -41,9 +41,10 @@ Mopidy-Raspberry-GPIO to your Mopidy configuration file::
|
|||
[raspberry-gpio]
|
||||
enabled = true
|
||||
bcm5 = play_pause,active_low,250
|
||||
bcm6 = volume_down,active_low,250
|
||||
bcm6 = prev,active_low,250
|
||||
bcm16 = next,active_low,250
|
||||
bcm20 = volume_up,active_low,250
|
||||
bcm21 = volume_down,active_low,10,rotenc_id=vol,step=1
|
||||
bcm20 = volume_up,active_low,10,rotenc_id=vol,step=1
|
||||
|
||||
Each bcmN entry corresponds to the BCM pin of that number.
|
||||
|
||||
|
|
|
@ -3,6 +3,8 @@ import logging
|
|||
import pykka
|
||||
from mopidy import core
|
||||
|
||||
from .rotencoder import RotEncoder
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
@ -14,6 +16,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)
|
||||
|
@ -33,6 +36,17 @@ class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
|
|||
pull = GPIO.PUD_DOWN
|
||||
edge = GPIO.RISING
|
||||
|
||||
if "rotenc_id" in settings.options:
|
||||
edge = GPIO.BOTH
|
||||
rotenc_id = settings.options["rotenc_id"]
|
||||
encoder = None
|
||||
if rotenc_id in self.rot_encoders.keys():
|
||||
encoder = self.rot_encoders[rotenc_id]
|
||||
else:
|
||||
encoder = RotEncoder(rotenc_id)
|
||||
self.rot_encoders[rotenc_id] = encoder
|
||||
encoder.add_pin(pin, settings.event)
|
||||
|
||||
GPIO.setup(pin, GPIO.IN, pull_up_down=pull)
|
||||
|
||||
GPIO.add_event_detect(
|
||||
|
@ -44,17 +58,30 @@ class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
|
|||
|
||||
self.pin_settings[pin] = settings
|
||||
|
||||
# TODO validate all self.rot_encoders have two pins
|
||||
|
||||
def find_pin_rotenc(self, pin):
|
||||
for encoder in self.rot_encoders.values():
|
||||
if pin in encoder.pins:
|
||||
return encoder
|
||||
|
||||
def gpio_event(self, pin):
|
||||
settings = self.pin_settings[pin]
|
||||
self.dispatch_input(settings)
|
||||
event = settings.event
|
||||
encoder = self.find_pin_rotenc(pin)
|
||||
if encoder:
|
||||
event = 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):
|
||||
|
|
|
@ -0,0 +1,51 @@
|
|||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RotEncoder:
|
||||
def __init__(self, rot_id):
|
||||
self.id = rot_id
|
||||
self.pins = []
|
||||
self.events = []
|
||||
self.state = (None, None)
|
||||
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
|
|
@ -18,6 +18,8 @@ dummy_config = {
|
|||
"bcm1": deserialize("play_pause,active_low,30"),
|
||||
"bcm2": deserialize("volume_up,active_high,30"),
|
||||
"bcm3": deserialize("volume_down,active_high,30"),
|
||||
"bcm4": deserialize("volume_down,active_high,250,rotenc_id=vol"),
|
||||
"bcm5": deserialize("volume_up,active_high,250,rotenc_id=vol"),
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -61,7 +63,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 +80,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 +97,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 +114,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 +131,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 +148,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 +165,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 +182,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()
|
||||
|
||||
|
@ -196,3 +198,31 @@ def test_frontend_gpio_event():
|
|||
frontend.gpio_event(3)
|
||||
|
||||
stop_mopidy_core()
|
||||
|
||||
|
||||
@mock.patch("RPi.GPIO.input")
|
||||
def test_frontend_rot_encoder_event(patched_input):
|
||||
patched_input.return_value = False
|
||||
|
||||
frontend = frontend_lib.RaspberryGPIOFrontend(
|
||||
dummy_config, dummy_mopidy_core()
|
||||
)
|
||||
|
||||
# Check that transition (False, True) -> (False, False) triggers volume_up
|
||||
encoder = frontend.rot_encoders["vol"]
|
||||
encoder.state = (False, True)
|
||||
|
||||
dispatch_input = mock.Mock()
|
||||
frontend.dispatch_input = dispatch_input
|
||||
|
||||
frontend.gpio_event(4)
|
||||
assert dispatch_input.call_args[0][0] == "volume_up"
|
||||
assert encoder.state == (False, False)
|
||||
|
||||
# Check that we do not submit an event for the invalid transition
|
||||
# (False, False) -> (False, False)
|
||||
dispatch_input.reset_mock()
|
||||
frontend.gpio_event(4)
|
||||
dispatch_input.assert_not_called()
|
||||
|
||||
stop_mopidy_core()
|
||||
|
|
|
@ -0,0 +1,54 @@
|
|||
import unittest
|
||||
|
||||
from unittest.mock import patch
|
||||
from mopidy_raspberry_gpio import RotEncoder
|
||||
|
||||
|
||||
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):
|
||||
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)
|
Loading…
Reference in New Issue