PlexKodiConnect/resources/lib/websocket_client.py

302 lines
10 KiB
Python

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, unicode_literals
from logging import getLogger
import json
from . import websocket
from . import backgroundthread, app, variables as v, utils, companion
log = getLogger('PLEX.websocket')
PMS_PATH = '/:/websockets/notifications'
PMS_INTERESTING_MESSAGE_TYPES = ('playing', 'timeline', 'activity')
SETTINGS_STRING = '_status'
def get_pms_uri():
uri = app.CONN.server
if not uri:
return
# Get the appropriate prefix for the websocket
if uri.startswith('https'):
uri = "wss%s" % uri[5:]
else:
uri = "ws%s" % uri[4:]
uri += PMS_PATH
log.debug('uri to connect pms websocket: %s', uri)
if app.ACCOUNT.pms_token:
uri += '?X-Plex-Token=' + app.ACCOUNT.pms_token
return uri
def get_alexa_uri():
if not app.ACCOUNT.plex_token:
return
return ('wss://pubsub.plex.tv/sub/websockets/%s/%s?X-Plex-Token=%s'
% (app.ACCOUNT.plex_user_id,
v.PKC_MACHINE_IDENTIFIER,
app.ACCOUNT.plex_token))
def pms_on_message(ws, message):
"""
Called when we receive a message from the PMS, e.g. when a new library
item has been added.
"""
try:
message = json.loads(message)
except ValueError as err:
log.error('Error decoding PMS websocket message: %s', err)
log.error('message: %s', message)
return
try:
message = message['NotificationContainer']
typus = message['type']
except KeyError:
log.error('Could not parse PMS message: %s', message)
return
# Triage
if typus not in PMS_INTERESTING_MESSAGE_TYPES:
# Drop everything we're not interested in
return
else:
# Put PMS message on queue and let libsync take care of it
app.APP.websocket_queue.put(message)
def alexa_on_message(ws, message):
"""
Called when we receive a message from Alexa
"""
log.debug('alexa message received: %s', message)
try:
message = utils.etree.fromstring(message)
except Exception as err:
log.error('Error decoding message from Alexa: %s %s', type(err), err)
log.error('message from Alexa: %s', message)
return
try:
if message.attrib['command'] == 'processRemoteControlCommand':
message = message[0]
else:
log.error('Unknown Alexa message received: %s', message)
return
companion.process_command(message.attrib['path'][1:], message.attrib)
except Exception as err:
log.exception('Could not parse Alexa message, error: %s %s',
type(err), err)
log.error('message: %s', message)
def on_error(ws, error):
status = ws.name + SETTINGS_STRING
if isinstance(error, IOError):
# We are probably offline
log.debug('%s: IOError connecting', ws.name)
# Status = IOError - not connected
utils.settings(status, value=utils.lang(39092))
ws.sleep_cycle()
elif isinstance(error, websocket.WebSocketTimeoutException):
log.debug('%s: WebSocketTimeoutException', ws.name)
# Status = 'Timeout - not connected'
utils.settings(status, value=utils.lang(39091))
ws.sleep_cycle()
elif isinstance(error, websocket.WebSocketConnectionClosedException):
log.debug('%s: WebSocketConnectionClosedException', ws.name)
# Status = Not connected
utils.settings(ws.name + SETTINGS_STRING, value=utils.lang(15208))
elif isinstance(error, websocket.WebSocketBadStatusException):
# Most likely Alexa not connecting, throwing a 403
log.debug('%s: got a bad HTTP status: %s', ws.name, error)
# Status = <value of exception>
utils.settings(status, value=str(error))
ws.sleep_cycle()
elif isinstance(error, websocket.WebSocketException):
log.error('%s: got another websocket exception %s: %s',
ws.name, type(error), error)
# Status = Error
utils.settings(status, value=utils.lang(257))
ws.sleep_cycle()
elif isinstance(error, SystemExit):
log.debug('%s: SystemExit detected', ws.name)
# Status = Not connected
utils.settings(ws.name + SETTINGS_STRING, value=utils.lang(15208))
else:
log.exception('%s: got an unexpected exception of type %s: %s',
ws.name, type(error), error)
# Status = Error
utils.settings(status, value=utils.lang(257))
raise RuntimeError
def on_close(ws):
"""
This does not seem to get called by our websocket client :-(
"""
log.debug('%s: connection closed', ws.name)
# Status = Not connected
utils.settings(ws.name + SETTINGS_STRING, value=utils.lang(15208))
def on_open(ws):
log.debug('%s: connected', ws.name)
# Status = Connected
utils.settings(ws.name + SETTINGS_STRING, value=utils.lang(13296))
ws.sleeptime = 0
class PlexWebSocketApp(websocket.WebSocketApp,
backgroundthread.KillableThread):
def __init__(self, **kwargs):
self.sleeptime = 0
backgroundthread.KillableThread.__init__(self)
websocket.WebSocketApp.__init__(self, self.get_uri(), **kwargs)
def sleep_cycle(self):
"""
Sleeps for 2^self.sleeptime where sleeping period will be doubled with
each unsuccessful connection attempt.
Will sleep at most 64 seconds
"""
self.sleep(2 ** self.sleeptime)
if self.sleeptime < 6:
self.sleeptime += 1
def close(self, **kwargs):
"""websocket.WebSocketApp is not yet thread-safe. close() might
encounter websockets that have already been closed"""
try:
websocket.WebSocketApp.close(self, **kwargs)
except AttributeError:
pass
def suspend(self, block=False, timeout=None):
"""
Call this method from another thread to suspend this websocket thread
"""
self.close()
backgroundthread.KillableThread.suspend(self, block, timeout)
def cancel(self):
"""
Call this method from another thread to cancel this websocket thread
"""
self.close()
backgroundthread.KillableThread.cancel(self)
def run(self):
"""
Ensure that sockets will be closed no matter what
"""
log.info("----===## Starting %s ##===----", self.name)
app.APP.register_thread(self)
try:
self._run()
except RuntimeError:
pass
except Exception as err:
log.exception('Exception of type %s occured: %s', type(err), err)
finally:
self.close()
# Status = Not connected
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(15208))
app.APP.deregister_thread(self)
log.info("----===## %s stopped ##===----", self.name)
def _run(self):
while not self.should_cancel():
# In the event the server goes offline
while self.should_suspend():
# We will be caught in this loop if either another thread
# called the suspend() method, thus setting _suspended = True
# OR if there any other conditions to not open a websocket
# connection - see methods should_suspend() below
# Status = Suspended - not connected
self.set_suspension_settings_status()
if self.wait_while_suspended():
# Abort was requested while waiting. We should exit
return
if not self._suspended:
# because wait_while_suspended will return instantly if
# this thread did not get suspended from another thread
self.sleep_cycle()
self.url = self.get_uri()
if not self.url:
self.sleep_cycle()
continue
self.run_forever()
class PMSWebsocketApp(PlexWebSocketApp):
name = 'pms_websocket'
def get_uri(self):
return get_pms_uri()
def should_suspend(self):
"""
Returns True if the thread needs to suspend.
"""
return (self._suspended or
utils.settings('enableBackgroundSync') != 'true')
def set_suspension_settings_status(self):
if utils.settings('enableBackgroundSync') != 'true':
# Status = Disabled
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(24023))
else:
# Status = 'Suspended - not connected'
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(39093))
class AlexaWebsocketApp(PlexWebSocketApp):
name = 'alexa_websocket'
def get_uri(self):
return get_alexa_uri()
def should_suspend(self):
"""
Returns True if the thread needs to suspend.
"""
return self._suspended or \
utils.settings('enable_alexa') != 'true' or \
app.ACCOUNT.restricted_user or \
not app.ACCOUNT.plex_token
def set_suspension_settings_status(self):
if utils.settings('enable_alexa') != 'true':
# Status = Disabled
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(24023))
elif app.ACCOUNT.restricted_user:
# Status = Managed Plex User - not connected
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(39094))
elif not app.ACCOUNT.plex_token:
# Status = Not logged in to plex.tv
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(39226))
else:
# Status = 'Suspended - not connected'
utils.settings(self.name + SETTINGS_STRING,
value=utils.lang(39093))
def get_pms_websocketapp():
return PMSWebsocketApp(on_open=on_open,
on_message=pms_on_message,
on_error=on_error,
on_close=on_close)
def get_alexa_websocketapp():
return AlexaWebsocketApp(on_open=on_open,
on_message=alexa_on_message,
on_error=on_error,
on_close=on_close)