tilman/piplayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mopidy REST tests

Browse Source
This commit is contained in:
Tilman
2017-10-09 21:21:11 +02:00
parent a3a61293a3
commit 6a7607a4da
5 changed files with 312 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/piplayer3.py
+210
View File
@@ -0,0 +1,210 @@
# piplayer3 - AudioSaucer
# http://www.tilman.de/piplayer3
import RPi.GPIO as GPIO
import logging
import time
import subprocess
import select # for polling zbarcam, see http://stackoverflow.com/a/10759061/3761783
import os
# Configuration
MUSIC_BASE_DIRECTORY = "/home/pi/music/"
SOUND_SCANNING = "/home/pi/piplayer2/sounds/scanning.mp3"
SOUND_OK = "/home/pi/piplayer2/sounds/ok-05.mp3"
SOUND_SCAN_FAIL = "/home/pi/piplayer2/sounds/fail-05.mp3"
SOUND_PLAYBACK_ERROR = "/home/pi/piplayer2/sounds/error.mp3"
QR_SCANNER_TIMEOUT = 3
MUSIC_EXTENSIONS = ['.aac', '.ac3', '.aiff', '.amr', '.au', '.flac', '.m4a', '.mid', '.mka', '.mp3', '.ogg', '.ra', '.voc', '.wav', '.wma']
# photo sensor on PIN 17
PIN_SENSOR = 17
# IR LED on PIN 22
PIN_IR_LED = 22
# LED on PIN 27
PIN_LED = 27
# Buttons on PINs 9, 10 and 11
PIN_PREV = 9
PIN_PLAY = 10
PIN_NEXT = 11
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s - %(message)s')
logging.info('Initializing')
GPIO.setmode(GPIO.BCM)
GPIO.setup(PIN_SENSOR, GPIO.IN)
GPIO.setup(PIN_IR_LED, GPIO.OUT)
GPIO.output(PIN_IR_LED, GPIO.LOW)
GPIO.setup(PIN_LED, GPIO.OUT)
GPIO.output(PIN_LED, GPIO.LOW)
GPIO.setup(PIN_PREV, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(PIN_PLAY, GPIO.IN, pull_up_down=GPIO.PUD_UP)
GPIO.setup(PIN_NEXT, GPIO.IN, pull_up_down=GPIO.PUD_UP)
position = 0
def btn1_callback(channel):
logging.debug('PREV_CALLBACK')
out, err = subprocess.Popen(['mocp','-Q', '%state'], stdout=subprocess.PIPE).communicate()
if (out.decode("utf-8").startswith('PLAY')):
out, err = subprocess.Popen(['mocp','-Q', '%cs'], stdout=subprocess.PIPE).communicate()
position = int(out.decode("utf-8"))
if (position < 4):
logging.debug('mocp --previous')
subprocess.call(["mocp", "--previous"])
else:
logging.debug('mocp --jump 0s')
subprocess.call(["mocp", "--jump", "0s"])
def btn2_callback(channel):
logging.debug('mocp --toggle-pause')
subprocess.call(["mocp", "--toggle-pause"])
def btn3_callback(channel):
logging.debug('mocp --next')
subprocess.call(["mocp", "--next"])
def clear_playlist():
return True
def add_file(filename):
return True
def start_playing():
return True
def play_sound(filename):
clear_playlist()
add_file(filename)
start_playing()
GPIO.add_event_detect(PIN_PREV, GPIO.FALLING, callback=btn1_callback, bouncetime=400)
GPIO.add_event_detect(PIN_PLAY, GPIO.FALLING, callback=btn2_callback, bouncetime=400)
GPIO.add_event_detect(PIN_NEXT, GPIO.FALLING, callback=btn3_callback, bouncetime=400)
try:
logging.info('Start moc server')
subprocess.call(["mocp", "--server"])
subprocess.call(["mocp", "--clear"])
subprocess.call(["mocp", "-l", SOUND_OK])
while True:
logging.debug('Wait for photo sensor')
GPIO.wait_for_edge(PIN_SENSOR, GPIO.RISING)
logging.debug('Photo sensor active, activating light and camera')
subprocess.call(["mocp", "-l", SOUND_SCANNING])
# turn LED on
GPIO.output(PIN_LED, GPIO.HIGH)
# scan QR code
zbarcam = subprocess.Popen(['zbarcam', '--quiet', '--nodisplay', '--raw', '-Sdisable', '-Sqrcode.enable', '--prescale=320x240', '/dev/video0'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
poll_obj = select.poll()
poll_obj.register(zbarcam.stdout, select.POLLIN)
# wait for scan result (or timeout)
start_time = time.time()
poll_result = False
while ((time.time() - start_time) < QR_SCANNER_TIMEOUT and (not poll_result)):
poll_result = poll_obj.poll(100)
if (poll_result):
try:
qr_code = zbarcam.stdout.readline().rstrip()
qr_code = qr_code.decode("utf-8") # python3
logging.info("QR Code: {}".format(qr_code))
if (not qr_code.startswith("http://")):
# create full path
full_path = MUSIC_BASE_DIRECTORY + qr_code
logging.debug("Full Path {}".format(full_path))
if (not os.path.isfile(full_path)):
logging.debug("not a file, add as directory")
directory = full_path
logging.debug("Directory {}".format(directory))
else:
logging.debug("it's a file")
directory = os.path.dirname(os.path.realpath(full_path))
filename = os.path.basename(full_path)
logging.debug("Directory {}".format(directory))
logging.debug("Filename {}".format(filename))
os.chdir(directory)
else:
logging.debug("URL, open as stream")
stream_url = qr_code
subprocess.call(["mocp", "--clear"])
subprocess.call(["mocp", "--stop"])
logging.debug("Stopped")
# play confirmation sound
subprocess.call(["mocp", "-l", SOUND_OK])
if ('stream_url' in locals()):
logging.debug("Add stream")
subprocess.check_call(["mocp", "-a", stream_url])
del stream_url
elif ('filename' in locals()):
logging.debug("Add file {}".format(filename))
subprocess.check_call(["mocp", "-a", filename])
del filename
else:
logging.debug("Add directory {}".format(directory))
subprocess.check_call(["mocp", "-a", "."])
logging.debug("Start playback")
subprocess.check_call(["mocp", "-p"])
except subprocess.CalledProcessError as e:
logging.error("Error starting playback, mocp returned {}".format(e.returncode))
logging.error(e.output)
subprocess.call(["mocp", "-l", SOUND_PLAYBACK_ERROR])
except FileNotFoundError as e:
logging.error("Could not open directory {}".format(directory))
subprocess.call(["mocp", "-l", SOUND_PLAYBACK_ERROR])
else:
logging.warning('Timeout on zbarcam')
subprocess.call(["mocp", "-l", SOUND_SCAN_FAIL])
zbarcam.terminate()
# LED off
GPIO.output(PIN_LED, GPIO.LOW)
# wait until sensor is not blocked anymore
if (GPIO.input(PIN_SENSOR) == GPIO.HIGH):
GPIO.wait_for_edge(PIN_SENSOR, GPIO.FALLING)
time.sleep(1)
# Exit when Ctrl-C is pressed
except KeyboardInterrupt:
logging.info('Shutdown')
finally:
logging.info('Close moc server')
subprocess.call(["mocp", "--exit"])
logging.info('Reset GPIO configuration and close')
GPIO.cleanup()
code/test/call_mopidy.py
+63
View File
@@ -0,0 +1,63 @@
import os
import requests
import logging
# Configuration
MUSIC_BASE_DIRECTORY = "/home/pi/music/"
SOUND_SCANNING = "/home/pi/piplayer2/sounds/scanning.mp3"
SOUND_OK = "/home/pi/piplayer2/sounds/ok-05.mp3"
SOUND_SCAN_FAIL = "/home/pi/piplayer2/sounds/fail-05.mp3"
SOUND_PLAYBACK_ERROR = "/home/pi/piplayer2/sounds/error.mp3"
MOPIDY_URL = 'http://127.0.0.1:6680/mopidy/rpc'
MUSIC_EXTENSIONS = ['.aac', '.ac3', '.aiff', '.amr', '.au', '.flac', '.m4a', '.mid', '.mka', '.mp3', '.ogg', '.ra', '.voc', '.wav', '.wma']
HEADERS = { "Content-Type": "application/json" }
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)s - %(message)s')
def call(method, params=None):
if (params is None):
json = '{"jsonrpc": "2.0", "id": 1, "method": "' + method + '"}'
else:
json = '{"jsonrpc": "2.0", "id": 1, "method": "' + method + '" , "params": ' + params + '}'
response = requests.post(MOPIDY_URL, data=json, headers=HEADERS)
logging.debug(response.text)
def clear_playlist():
call('core.tracklist.clear')
def add_uri(filename):
logging.debug('Adding ' + filename)
call('core.tracklist.add', '{"uri": "' + filename + '"}')
def start_playing():
return True
def play_sound(filename):
clear_playlist()
add_file(filename)
start_playing()
def list_files(dir):
r = []
for path, dirs, files in os.walk(dir):
dirs.sort()
files.sort()
for name in files:
r.append(os.path.join(path, name))
return r
clear_playlist();
filenames = list_files('/home/tilman/Musik')
for filename in filenames:
file_uri = 'file://' + filename
add_uri(file_uri)
code/test/listfiles.py
+10
View File
@@ -0,0 +1,10 @@
import os
def list_files(dir):
r = []
for root, dirs, files in os.walk(dir):
for name in files:
r.append(os.path.join(root, name))
return r
print(list_files('/home/tilman/Musik'))
code/test/rest.py
+18
View File
@@ -0,0 +1,18 @@
# '{"jsonrpc": "2.0", "id": 1, "method": "core.tracklist.add" , "params": {"uri": "spotify:track:5tf1VVWniHgryyumXyJM7w"}}'
# http://127.0.0.1:6680/mopidy/rpc
import requests
#import json
#specify url
url = 'http://127.0.0.1:6680/mopidy/rpc'
data = '{"jsonrpc": "2.0", "id": 1, "method": "core.tracklist.add" , "params": {"uri": "spotify:track:5tf1VVWniHgryyumXyJM7w"}}'
headers = { "Content-Type": "application/json" }
#Call REST API
response = requests.post(url, data=data, headers=headers)
#Print Response
print(response.text)