secstate/Iris
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Snapcast thread

Browse Source
This commit is contained in:
James Barnsley
2018-11-03 23:19:03 +13:00
parent d207b39792
commit 4327771b2e
7 changed files with 237 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
mopidy_iris/core.py
View File

@ -1,9 +1,8 @@
from __future__ import unicode_literals
import random, string, socket, logging, json, pykka, pylast, urllib, urllib2, os, sys, mopidy_iris, subprocess
import random, string, logging, json, pykka, pylast, urllib, urllib2, os, sys, mopidy_iris, subprocess
import tornado.web
import tornado.websocket
import tornado.ioloop
import tornado.httpclient
import requests
@ -15,6 +14,8 @@ from pkg_resources import parse_version
from tornado.escape import json_encode, json_decode
from .system import IrisSystemThread
from .snapcast import IrisSnapcast
from .snapcast_thread import IrisSnapcastThread
if sys.platform == 'win32':
import ctypes
@ -36,7 +37,7 @@ class IrisCore(pykka.ThreadingActor):
"seed_tracks": [],
"results": []
}
snapcast_listener = False
snapcast_daemon = False
##
@ -48,6 +49,12 @@ class IrisCore(pykka.ThreadingActor):
# Load our commands from file
self.commands = self.load_from_file('commands')
# Start our TCP watcher with no request, so it becomes our
# long-running socket connection
if self.config['iris'].get('snapcast_enabled'):
self.snapcast_daemon = IrisSnapcastThread(self.config, self.broadcast)
self.snapcast_daemon.start()
##
# Mopidy is shutting down
@ -55,9 +62,29 @@ class IrisCore(pykka.ThreadingActor):
def stop(self):
logger.info('Stopping Iris')
self.snapcast_disconnect_listener()
if self.snapcast_daemon:
self.snapcast_daemon.stop()
##
# Make a request to snapcast
##
def snapcast(self, *args, **kwargs):
callback = kwargs.get('callback', None)
request_id = kwargs.get('request_id', None)
data = kwargs.get('data', {})
# Start a new thread, just for this request
socket = IrisSnapcast(self.config)
socket.connect()
response = socket.request(data)
if (callback):
callback(response)
else:
return response
##
# Save dict object to disk
@ -101,150 +128,6 @@ class IrisCore(pykka.ThreadingActor):
return {}
##
# Create a new snapcast TCP connection
#
# @return socket
##
def new_snapcast_socket(self):
if not self.config['iris'].get('snapcast_enabled'):
logger.error("Iris Snapcast not enabled")
raise Exception("Snapcast not enabled")
try:
snapcast = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
snapcast.settimeout(10)
snapcast.connect((self.config['iris']['snapcast_host'], self.config['iris']['snapcast_port']))
except socket.gaierror, e:
logger.error("Iris could not connect to Snapcast: %s" % e)
raise Exception(e);
except socket.error, e:
logger.error("Iris could not connect to Snapcast: %s" % e)
raise Exception(e);
return snapcast
##
# Create our ongoing notification listener
#
# TODO: Make non-blocking
##
def create_snapcast_listener(self):
try:
listener = self.new_snapcast_socket()
self.snapcast_listener = listener
logger.info("Iris connected to Snapcast")
except Exception, e:
logger.error("Iris could not connect to Snapcast: %s" % e)
##
# Disconnect our Snapcast listener
##
def snapcast_disconnect_listener(self):
if self.snapcast_listener:
self.snapcast_listener.close()
self.snapcast_listener = None
##
# Handle a message from the Snapcast Telnet API
# We create a connection for this request, and then drop it once completed. This is because
# we have a separate socket for monitoring notifications
#
# @param data = string
##
def snapcast_handle_message(self, data):
logger.debug("Iris received Snapcast message: "+data)
try:
data = json.loads(data)
except:
logger.error("Iris could not digest Snapcast message: "+data)
print "broadcasting..."
print data
self.broadcast(data)
##
# Send a request to Snapcast
#
# We create a connection for this request, and then drop it once completed. This is because
# we have a separate socket for monitoring notifications
##
def snapcast_instruct(self, *args, **kwargs):
callback = kwargs.get('callback', None)
request_id = kwargs.get('request_id', None)
data = kwargs.get('data', {})
request = {
'id': self.generateGuid(),
'jsonrpc': '2.0',
'method': data['method'],
'params': data['params'] if 'params' in data else {}
}
# Convert to string. For some really nuts reason we need an extra trailing curly brace...
request = json.dumps(request)+'}'
# Create our connection
try:
snapcast_socket = self.new_snapcast_socket()
except Exception, e:
callback(response=None, error={
'message': "Could not connect to Snapcast",
'data': str(e)
})
return
# Attempt to send the request
try:
snapcast_socket.send(request.encode('ascii')+b"\n")
except socket.error, e:
logger.error("Iris could not send request to Snapcast: %s" % e)
callback(response=None, error={
'message': "Failed to send request to Snapcast",
'data': str(e)
})
return
# Wait for response
while True:
try:
response = snapcast_socket.recv(8192)
except socket.error, e:
logger.error("Iris failed to receive Snapcast response: %s" % e)
callback(response=None, error={
'message': "Failed to receive Snapcast response",
'data': str(e)
})
snapcast_socket.close()
if not len(response):
break
try:
response = json.loads(response)
if 'result' in response:
callback(response=response['result'])
else:
callback(error=response['error'])
except:
logger.error("Iris received malformed Snapcast response: "+response)
callback(response=None, error={
'message': "Malformed Snapcast response",
'data': response
})
return
return
##
# Generate a random string
#
@ -293,6 +176,8 @@ class IrisCore(pykka.ThreadingActor):
}
def send_message(self, *args, **kwargs):
callback = kwargs.get('callback', None)
data = kwargs.get('data', None)

175
mopidy_iris/snapcast.py Executable file
View File

@ -0,0 +1,175 @@
from threading import Thread
import os, logging, subprocess, socket, json, random, string
# import logger
logger = logging.getLogger(__name__)
class IrisSnapcast(object):
socket = None
listen = False
def __init__(self, config):
self.config = config
self.path = os.path.dirname(__file__)
##
# This thread has been started
##
def connect(self):
host = str(self.config['iris']['snapcast_host'])
port = int(self.config['iris']['snapcast_port'])
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(10)
self.socket.connect((host, port))
logger.debug("Snapcast connection established on "+host+":"+str(port))
except socket.gaierror, e:
raise Exception(e);
except socket.error, e:
raise Exception(e);
##
# Listen for socket activity
# Broadcast events to the piped broadcaster
#
# @param broadcast = Method
##
def listen(self, broadcast):
logger.info("Established Snapcast listener")
broadcast(data={'method':'snapcast_connected'})
self.listen = True
data = ""
while self.listen:
# TODO: This works great, but when no data over X seconds we hit the timeout
# Possibly stop timeout? seems to lock things up
data = data + self.socket.recv(1024)
print data
if data.endswith(u"\r\n"):
message = data
data = ""
logger.debug("Incoming Snapcast message: "+message)
try:
message = json.loads(message)
broadcast(data=message)
except:
logger.error("Malformed Snapcast message: "+message)
##
# Stop our listener
#
# @param broadcast = Method
##
def stop_listening(self):
self.listen = False
##
# Send a single request to Snapcast
#
# @param data = Dict
##
def request(self, data):
# No socket to make request on
if self.socket == None:
raise Exception("Socket not established");
# Construct our request, based on the provided data
request = {
'id': self.generateGuid(),
'jsonrpc': '2.0',
'method': data['method'],
'params': data['params'] if 'params' in data else {}
}
logger.debug("Outgoing Snapcast message")
logger.debug(request)
# Convert to string. For some really nuts reason we need an extra trailing curly brace...
request = json.dumps(request)+'}'
# Attempt to send the request
try:
self.socket.send(request.encode('ascii')+b"\n")
except socket.error, e:
logger.error("Iris could not send request to Snapcast: %s" % e)
socket_response = {
'error': {
'message': "Failed to send request to Snapcast",
'data': str(e)
}
}
# Wait for response
while True:
try:
response = self.socket.recv(8192)
except socket.error, e:
logger.error("Iris failed to receive Snapcast response: %s" % e)
socket_response = {
'error': {
'message': "Failed to receive Snapcast response",
'data': str(e)
}
}
if not len(response):
break
try:
response = json.loads(response)
logger.debug("Incoming Snapcast message")
logger.debug(response)
if 'result' in response:
socket_response = {
'response': response['result']
}
else:
socket_response = {
'error': response['error']
}
except:
logger.error("Iris received malformed Snapcast response: "+response)
socket_response = {
'error': {
'message': "Malformed Snapcast response",
'data': response
}
}
return socket_response
##
# Generate a random string
#
# Used for connection_ids where none is provided by client
# @return string
##
def generateGuid(self):
length = 12
return ''.join(random.choice(string.lowercase) for i in range(length))

25
mopidy_iris/snapcast_thread.py Executable file
View File

@ -0,0 +1,25 @@
from threading import Thread
import os, logging, subprocess, socket, json, random, string
from .snapcast import IrisSnapcast
# import logger
logger = logging.getLogger(__name__)
class IrisSnapcastThread(Thread):
def __init__(self, config, broadcast):
Thread.__init__(self)
self.config = config
self.broadcast = broadcast
##
# This thread has been started
##
def run(self):
socket = IrisSnapcast(self.config)
socket.connect()
socket.listen(self.broadcast)

2
mopidy_iris/static/app.js
View File

@ -60725,7 +60725,7 @@ var SnapcastMiddleware = function () {
var response_callback = arguments.length > 2 && arguments[2] !== undefined ? arguments[2] : null;
var error_callback = arguments.length > 3 && arguments[3] !== undefined ? arguments[3] : null;
store.dispatch(pusherActions.request('snapcast_instruct', params, response_callback, error_callback));
store.dispatch(pusherActions.request('snapcast', params, response_callback, error_callback));
return;
};

2
mopidy_iris/static/app.js.map
View File

File diff suppressed because one or more lines are too long

2
mopidy_iris/static/index.html
View File

@ -98,7 +98,7 @@
// Release details
// These are automatically injected to built HTML
var build = "1541140401";
var build = "1541209549";
var version = "3.29.2";
// Construct the script tag