diff --git a/pyskyqremote/const.py b/pyskyqremote/const.py new file mode 100644 index 0000000..e7e1ed3 --- /dev/null +++ b/pyskyqremote/const.py @@ -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 = """ + + + 0 + + +""" +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, + }, + ], +} diff --git a/pyskyqremote/skyq_remote.py b/pyskyqremote/skyq_remote.py index 639648b..6f91b6a 100644 --- a/pyskyqremote/skyq_remote.py +++ b/pyskyqremote/skyq_remote.py @@ -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 @@ -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, ): @@ -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 @@ -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) @@ -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.""" @@ -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.""" @@ -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