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

dos2unix crlf -> lf

remotes/origin/python3
Phil Howard 2019-11-21 19:06:13 +00:00
parent eadc6248df
commit 68abdf508b
3 changed files with 181 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
mopidy_raspberry_gpio/frontend.py
View File

@ -1,82 +1,82 @@
import logging import logging
import pykka import pykka
from mopidy import core from mopidy import core
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener): class RaspberryGPIOFrontend(pykka.ThreadingActor, core.CoreListener):
def __init__(self, config, core): def __init__(self, config, core):
super().__init__() super().__init__()
import RPi.GPIO as GPIO import RPi.GPIO as GPIO
self.core = core self.core = core
self.config = config["raspberry-gpio"] self.config = config["raspberry-gpio"]
self.pin_settings = {} self.pin_settings = {}
GPIO.setwarnings(False) GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM) GPIO.setmode(GPIO.BCM)
# Iterate through any bcmN pins in the config # Iterate through any bcmN pins in the config
# and set them up as inputs with edge detection # and set them up as inputs with edge detection
for key in self.config: for key in self.config:
if key.startswith("bcm"): if key.startswith("bcm"):
pin = int(key.replace("bcm", "")) pin = int(key.replace("bcm", ""))
settings = self.config[key] settings = self.config[key]
if settings is None: if settings is None:
continue continue
pull = GPIO.PUD_UP pull = GPIO.PUD_UP
edge = GPIO.FALLING edge = GPIO.FALLING
if settings.active == "active_high": if settings.active == "active_high":
pull = GPIO.PUD_DOWN pull = GPIO.PUD_DOWN
edge = GPIO.RISING edge = GPIO.RISING
GPIO.setup(pin, GPIO.IN, pull_up_down=pull) GPIO.setup(pin, GPIO.IN, pull_up_down=pull)
GPIO.add_event_detect( GPIO.add_event_detect(
pin, pin,
edge, edge,
callback=self.gpio_event, callback=self.gpio_event,
bouncetime=settings.bouncetime, bouncetime=settings.bouncetime,
) )
self.pin_settings[pin] = settings self.pin_settings[pin] = settings
def gpio_event(self, pin): def gpio_event(self, pin):
settings = self.pin_settings[pin] settings = self.pin_settings[pin]
self.dispatch_input(settings.event) self.dispatch_input(settings.event)
def dispatch_input(self, event): def dispatch_input(self, event):
handler_name = f"handle_{event}" handler_name = f"handle_{event}"
try: try:
getattr(self, handler_name)() getattr(self, handler_name)()
except AttributeError: except AttributeError:
raise RuntimeError( raise RuntimeError(
f"Could not find input handler for event: {event}" f"Could not find input handler for event: {event}"
) )
def handle_play_pause(self): def handle_play_pause(self):
if self.core.playback.state.get() == core.PlaybackState.PLAYING: if self.core.playback.state.get() == core.PlaybackState.PLAYING:
self.core.playback.pause() self.core.playback.pause()
else: else:
self.core.playback.play() self.core.playback.play()
def handle_next(self): def handle_next(self):
self.core.playback.next() self.core.playback.next()
def handle_prev(self): def handle_prev(self):
self.core.playback.previous() self.core.playback.previous()
def handle_volume_up(self): def handle_volume_up(self):
volume = self.core.playback.volume.get() volume = self.core.playback.volume.get()
volume += 5 volume += 5
volume = min(volume, 100) volume = min(volume, 100)
self.core.playback.volume = volume self.core.playback.volume = volume
def handle_volume_down(self): def handle_volume_down(self):
volume = self.core.playback.volume.get() volume = self.core.playback.volume.get()
volume -= 5 volume -= 5
volume = max(volume, 0) volume = max(volume, 0)
self.core.playback.volume = volume self.core.playback.volume = volume

118
mopidy_raspberry_gpio/pinconfig.py
View File

@ -1,59 +1,59 @@
from collections import namedtuple from collections import namedtuple
from mopidy import config from mopidy import config
from mopidy.config import types from mopidy.config import types
class ValidList(list): class ValidList(list):
def __format__(self, format_string=None): def __format__(self, format_string=None):
if format_string is None: if format_string is None:
format_string = ", " format_string = ", "
return format_string.join(self) return format_string.join(self)
class PinConfig(config.ConfigValue): class PinConfig(config.ConfigValue):
tuple_pinconfig = namedtuple("PinConfig", ("event", "active", "bouncetime")) tuple_pinconfig = namedtuple("PinConfig", ("event", "active", "bouncetime"))
valid_events = ValidList(["play_pause", "prev", "next", "volume_up", "volume_down"]) valid_events = ValidList(["play_pause", "prev", "next", "volume_up", "volume_down"])
valid_modes = ValidList(["active_low", "active_high"]) valid_modes = ValidList(["active_low", "active_high"])
def __init__(self): def __init__(self):
pass pass
def deserialize(self, value): def deserialize(self, value):
if value is None: if value is None:
return None return None
value = types.decode(value).strip() value = types.decode(value).strip()
try: try:
event, active, bouncetime = value.split(",") event, active, bouncetime = value.split(",")
except ValueError: except ValueError:
return None return None
if event not in self.valid_events: if event not in self.valid_events:
raise ValueError( raise ValueError(
f"invalid event for pin config {event} (Must be {valid_events})" f"invalid event for pin config {event} (Must be {valid_events})"
) )
if active not in self.valid_modes: if active not in self.valid_modes:
raise ValueError( raise ValueError(
f"invalid event for pin config {active} (Must be one of {valid_modes})" f"invalid event for pin config {active} (Must be one of {valid_modes})"
) )
try: try:
bouncetime = int(bouncetime) bouncetime = int(bouncetime)
except ValueError: except ValueError:
raise ValueError( raise ValueError(
f"invalid bouncetime value for pin config {bouncetime}" f"invalid bouncetime value for pin config {bouncetime}"
) )
return self.tuple_pinconfig(event, active, bouncetime) return self.tuple_pinconfig(event, active, bouncetime)
def serialize(self, value, display=False): def serialize(self, value, display=False):
if value is None: if value is None:
return "" return ""
value = f"{value.event},{value.active},{value.bouncetime}" value = f"{value.event},{value.active},{value.bouncetime}"
return types.encode(value) return types.encode(value)

80
tests/test_frontend.py
View File

@ -1,40 +1,40 @@
import sys import sys
from unittest import mock from unittest import mock
import pytest import pytest
from mopidy_raspberry_gpio import Extension from mopidy_raspberry_gpio import Extension
from mopidy_raspberry_gpio import frontend as frontend_lib from mopidy_raspberry_gpio import frontend as frontend_lib
from mopidy_raspberry_gpio import pinconfig from mopidy_raspberry_gpio import pinconfig
deserialize = pinconfig.PinConfig().deserialize deserialize = pinconfig.PinConfig().deserialize
dummy_config = { dummy_config = {
"raspberry-gpio": { "raspberry-gpio": {
# Plugins expect settings to be deserialized # Plugins expect settings to be deserialized
"bcm1": deserialize("play_pause,active_low,30") "bcm1": deserialize("play_pause,active_low,30")
} }
} }
def test_get_frontend_classes(): def test_get_frontend_classes():
sys.modules["RPi"] = mock.Mock() sys.modules["RPi"] = mock.Mock()
sys.modules["RPi.GPIO"] = mock.Mock() sys.modules["RPi.GPIO"] = mock.Mock()
ext = Extension() ext = Extension()
registry = mock.Mock() registry = mock.Mock()
ext.setup(registry) ext.setup(registry)
registry.add.assert_called_once_with( registry.add.assert_called_once_with(
"frontend", frontend_lib.RaspberryGPIOFrontend "frontend", frontend_lib.RaspberryGPIOFrontend
) )
def test_frontend_handler_dispatch(): def test_frontend_handler_dispatch():
sys.modules["RPi"] = mock.Mock() sys.modules["RPi"] = mock.Mock()
sys.modules["RPi.GPIO"] = mock.Mock() sys.modules["RPi.GPIO"] = mock.Mock()
frontend = frontend_lib.RaspberryGPIOFrontend(dummy_config, mock.Mock()) frontend = frontend_lib.RaspberryGPIOFrontend(dummy_config, mock.Mock())
with pytest.raises(RuntimeError): with pytest.raises(RuntimeError):
frontend.dispatch_input("tomato") frontend.dispatch_input("tomato")