Skip to content

Commit

Permalink
Updated to simplify code and usage
Browse files Browse the repository at this point in the history
  • Loading branch information
@RogerSelwyn
RogerSelwyn committed Apr 27, 2020
1 parent 7b798c5 commit 3a909a8
Show file tree
Hide file tree
Showing 2 changed files with 205 additions and 55 deletions.
145 changes: 145 additions & 0 deletions pyskyqremote/const.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
"""Constants for pyskyqremote."""
# SOAP/UPnP Constants
SKY_PLAY_URN = "urn:nds-com:serviceId:SkyPlay"
SKYCONTROL = "SkyControl"
SOAP_ACTION = '"urn:schemas-nds-com:service:SkyPlay:2#{0}"'
SOAP_CONTROL_BASE_URL = "http://{0}:49153{1}"
SOAP_DESCRIPTION_BASE_URL = "http://{0}:49153/description{1}.xml"
SOAP_PAYLOAD = """<s:Envelope xmlns:s='http://schemas.xmlsoap.org/soap/envelope/' s:encodingStyle='http://schemas.xmlsoap.org/soap/encoding/'>
<s:Body>
<u:{0} xmlns:u="urn:schemas-nds-com:service:SkyPlay:2">
<InstanceID>0</InstanceID>
</u:{0}>
</s:Body>
</s:Envelope>"""
SOAP_RESPONSE = "u:{0}Response"
SOAP_USER_AGENT = "SKYPLUS_skyplus"
UPNP_GET_MEDIA_INFO = "GetMediaInfo"
UPNP_GET_TRANSPORT_INFO = "GetTransportInfo"

# WebSocket Constants
WS_BASE_URL = "ws://{0}:9006/as/{1}"
WS_CURRENT_APPS = "apps/status"

# REST Constants
REST_BASE_URL = "http://{0}:{1}/as/{2}"
REST_CHANNEL_LIST = "services"
REST_RECORDING_DETAILS = "pvr/details/{0}"
REST_PATH_INFO = "system/information"
REST_PATH_DEVICEINFO = "system/deviceinformation"

# Generic Constants
DEFAULT_ENCODING = "utf-8"

# Sky specific constants
CURRENT_URI = "CurrentURI"
CURRENT_TRANSPORT_STATE = "CurrentTransportState"
APP_STATUS_VISIBLE = "VISIBLE"

PVR = "pvr"
XSI = "xsi"
PAST_END_OF_EPG = "past end of epg"

CONNECTTIMEOUT = 1000
TIMEOUT = 2

SKY_STATE_PLAYING = "PLAYING"
SKY_STATE_PAUSED = "PAUSED_PLAYBACK"
SKY_STATE_STANDBY = "STANDBY"
SKY_STATE_ON = "ON"
SKY_STATE_OFF = "POWERED OFF"

# Application Constants
APP_EPG = "com.bskyb.epgui"


COMMANDS = {
"power": 0,
"select": 1,
"backup": 2,
"dismiss": 2,
"channelup": 6,
"channeldown": 7,
"interactive": 8,
"sidebar": 8,
"help": 9,
"services": 10,
"search": 10,
"tvguide": 11,
"home": 11,
"i": 14,
"text": 15,
"up": 16,
"down": 17,
"left": 18,
"right": 19,
"red": 32,
"green": 33,
"yellow": 34,
"blue": 35,
"0": 48,
"1": 49,
"2": 50,
"3": 51,
"4": 52,
"5": 53,
"6": 54,
"7": 55,
"8": 56,
"9": 57,
"play": 64,
"pause": 65,
"stop": 66,
"record": 67,
"fastforward": 69,
"rewind": 71,
"boxoffice": 240,
"sky": 241,
}

RESPONSE_OK = 200

TEST_CHANNEL_LIST = {
"documentId": "2416",
"services": [
{
"c": "317",
"dvbtriplet": "64511.8800.11684",
"schedule": True,
"servicetype": "DSAT",
"servicetypes": ["DSAT"],
"sf": "hd",
"sg": 12,
"sid": "684",
"sk": 684,
"t": "Premium Comedy HD",
"xsg": 3,
},
{
"c": "400",
"dvbtriplet": "64511.6400.11075",
"schedule": True,
"servicetype": "DSAT",
"servicetypes": ["DSAT"],
"sf": "hd",
"sg": 13,
"sid": "75",
"sk": 75,
"t": "Sky Arte HD",
"xsg": 4,
},
{
"c": "120",
"dvbtriplet": "64511.6400.11074",
"schedule": True,
"servicetype": "DSAT",
"servicetypes": ["DSAT"],
"sf": "hd",
"sg": 21,
"sid": "74",
"sk": 74,
"t": "Sky Arte HD",
"xsg": 1,
},
],
}
115 changes: 60 additions & 55 deletions pyskyqremote/skyq_remote.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,12 @@
CONNECTTIMEOUT,
TIMEOUT,
COMMANDS,
CONST_SKY_STATE_PLAYING,
CONST_SKY_STATE_PAUSED,
CONST_SKY_STATE_STANDBY,
CONST_SKY_STATE_ON,
CONST_SKY_STATE_OFF,
CONST_APP_EPG,
SKY_STATE_PLAYING,
SKY_STATE_PAUSED,
SKY_STATE_STANDBY,
SKY_STATE_ON,
SKY_STATE_OFF,
APP_EPG,
)
from .const import TEST_CHANNEL_LIST

Expand All @@ -58,15 +58,6 @@ class SkyQRemote:

commands = COMMANDS

SKY_STATE_PLAYING = CONST_SKY_STATE_PLAYING
SKY_STATE_PAUSED = CONST_SKY_STATE_PAUSED
SKY_STATE_STANDBY = CONST_SKY_STATE_STANDBY
SKY_STATE_ON = CONST_SKY_STATE_ON
SKY_STATE_OFF = CONST_SKY_STATE_OFF

# Application Constants
APP_EPG = CONST_APP_EPG

def __init__(
self, host, overrideCountry=None, test_channel=None, port=49160, jsonport=9006,
):
Expand All @@ -91,43 +82,43 @@ def powerStatus(self) -> str:
self._setupDevice()

if self._soapControlURL is None:
return self.SKY_STATE_OFF
return SKY_STATE_OFF
try:
output = self._http_json(REST_PATH_INFO)
if "activeStandby" in output and output["activeStandby"] is False:
return self.SKY_STATE_ON
return self.SKY_STATE_STANDBY
return SKY_STATE_ON
return SKY_STATE_STANDBY
except (
requests.exceptions.ConnectTimeout,
requests.exceptions.ReadTimeout,
):
return self.SKY_STATE_STANDBY
return SKY_STATE_STANDBY
except (requests.exceptions.ConnectionError):
_LOGGER.info(
f"I0010 - Device has control URL but connection request failed: {self._host}"
)
return self.SKY_STATE_OFF
return SKY_STATE_OFF
except Exception as err:
_LOGGER.exception(f"X0060 - Error occurred: {self._host} : {err}")
return self.SKY_STATE_STANDBY
return SKY_STATE_STANDBY

def getCurrentState(self):
"""Get current state of the SkyQ box."""
if self.powerStatus() == self.SKY_STATE_STANDBY:
return self.SKY_STATE_STANDBY
if self.powerStatus() == SKY_STATE_STANDBY:
return SKY_STATE_STANDBY
response = self._callSkySOAPService(UPNP_GET_TRANSPORT_INFO)
if response is not None:
state = response[CURRENT_TRANSPORT_STATE]
if state == self.SKY_STATE_PLAYING:
return self.SKY_STATE_PLAYING
if state == self.SKY_STATE_PAUSED:
return self.SKY_STATE_PAUSED
return self.SKY_STATE_STANDBY
if state == SKY_STATE_PLAYING:
return SKY_STATE_PLAYING
if state == SKY_STATE_PAUSED:
return SKY_STATE_PAUSED
return SKY_STATE_STANDBY

def getActiveApplication(self):
"""Get the active application on Sky Q box."""
try:
result = self.APP_EPG
result = APP_EPG
apps = self._callSkyWebSocket(WS_CURRENT_APPS)
if apps is None:
return result
Expand All @@ -146,10 +137,8 @@ def getCurrentMedia(self):
result = {
"channel": None,
"imageUrl": None,
"title": None,
"season": None,
"episode": None,
"sid": None,
"pvrId": None,
"live": False,
}
response = self._callSkySOAPService(UPNP_GET_MEDIA_INFO)
Expand All @@ -171,27 +160,9 @@ def getCurrentMedia(self):
elif PVR in currentURI:
# Recorded content
pvrId = "P" + currentURI[11:]
recording = self._http_json(REST_RECORDING_DETAILS.format(pvrId))
result.update({"channel": recording["details"]["cn"]})
result.update({"title": recording["details"]["t"]})
if (
"seasonnumber" in recording["details"]
and "episodenumber" in recording["details"]
):
result.update({"season": recording["details"]["seasonnumber"]})
result.update(
{"episode": recording["details"]["episodenumber"]}
)
if "programmeuuid" in recording["details"]:
programmeuuid = recording["details"]["programmeuuid"]
imageUrl = self._remoteCountry.pvr_image_url.format(
str(programmeuuid)
)
if "osid" in recording["details"]:
osid = recording["details"]["osid"]
imageUrl += "?sid=" + str(osid)
result.update({"imageUrl": imageUrl})
return result
result.update({"pvrId": pvrId, "live": False})

return _objectview(result)

def getEpgData(self, sid, epgDate):
"""Get EPG data for the specified channel."""
Expand Down Expand Up @@ -231,10 +202,39 @@ def getCurrentLiveTVProgramme(self, sid):
result.update({"episode": programme["episode"]})
result.update({"season": programme["season"]})
result.update({"imageUrl": programme["imageUrl"]})
return result
return _objectview(result)
except Exception as err:
_LOGGER.exception(f"X0030 - Error occurred: {self._host} : {sid} : {err}")
return result
return _objectview(result)

def getRecording(self, pvrId):
"""Get the recording details."""
result = {
"channel": None,
"imageUrl": None,
"title": None,
"season": None,
"episode": None,
}

recording = self._http_json(REST_RECORDING_DETAILS.format(pvrId))
result.update({"channel": recording["details"]["cn"]})
result.update({"title": recording["details"]["t"]})
if (
"seasonnumber" in recording["details"]
and "episodenumber" in recording["details"]
):
result.update({"season": recording["details"]["seasonnumber"]})
result.update({"episode": recording["details"]["episodenumber"]})
if "programmeuuid" in recording["details"]:
programmeuuid = recording["details"]["programmeuuid"]
imageUrl = self._remoteCountry.pvr_image_url.format(str(programmeuuid))
if "osid" in recording["details"]:
osid = recording["details"]["osid"]
imageUrl += "?sid=" + str(osid)
result.update({"imageUrl": imageUrl})

return _objectview(result)

def press(self, sequence):
"""Issue the specified sequence of commands to SkyQ box."""
Expand Down Expand Up @@ -467,3 +467,8 @@ def __init__(self, url):

def received_message(self, message):
self.data = message.data


class _objectview(object):
def __init__(self, d):
self.__dict__ = d

0 comments on commit 3a909a8

Please sign in to comment.