PKlN aioroku/__init__.py"""An asynchronous library to control all the rokus in your life. Easy, breezy, beautiful.""" from aioroku.core import AioRoku, Application, AioRokuException __version__ = '0.0.2' PKJmNAD!!aioroku/core.pyfrom urllib.parse import quote_plus from aiohttp import client_exceptions import xml.etree.ElementTree as ET COMMANDS = { # Standard Keys 'home': 'Home', 'reverse': 'Rev', 'forward': 'Fwd', 'play': 'Play', 'select': 'Select', 'left': 'Left', 'right': 'Right', 'down': 'Down', 'up': 'Up', 'back': 'Back', 'replay': 'InstantReplay', 'info': 'Info', 'backspace': 'Backspace', 'search': 'Search', 'enter': 'Enter', 'literal': 'Lit', # For devices that support "Find Remote" 'find_remote': 'FindRemote', # For Roku TV 'volume_down': 'VolumeDown', 'volume_up': 'VolumeUp', 'volume_mute': 'VolumeMute', # For Roku TV while on TV tuner channel 'channel_up': 'ChannelUp', 'channel_down': 'ChannelDown', # For Roku TV current input 'input_tuner': 'InputTuner', 'input_hdmi1': 'InputHDMI1', 'input_hdmi2': 'InputHDMI2', 'input_hdmi3': 'InputHDMI3', 'input_hdmi4': 'InputHDMI4', 'input_av1': 'InputAV1', # For devices that support being turned on/off 'power': 'Power', 'poweroff': 'PowerOff', } SENSORS = ('acceleration', 'magnetic', 'orientation', 'rotation') TOUCH_OPS = ('up', 'down', 'press', 'move', 'cancel') class AioRokuException(Exception): pass class Application: '''Application class with the icon property.''' def __init__(self, roku_id, version, name, roku=None, is_screensaver=False): self.id = str(roku_id) self.version = version self.name = name self.is_screensaver = is_screensaver self.roku = roku def __eq__(self, other): return isinstance(other, Application) and \ (self.id, self.version) == (other.id, other.version) def __repr__(self): return ('' % (self.id, self.name, self.version)) @property def icon(self): if self.roku: return self.roku.icon(self) def launch(self): if self.roku: self.roku.launch(self) def store(self): if self.roku: self.roku.store(self) class DeviceInfo: def __init__(self, model_name, model_num, software_version, software_build, serial_num, user_device_name, roku_type): self.model_name = model_name self.model_num = model_num self.software_version = software_version self.software_build = software_build self.serial_num = serial_num self.user_device_name = user_device_name self.roku_type = roku_type def __repr__(self): return ('' % (self.model_name, self.model_num, self.software_version, self.serial_num, self.roku_type)) class AioRoku: def __init__(self, host, session, port=8060): self.host = host self.port = port self.session = session self.sw_info = self.roku_type = self.dinfo = 'unknown' # self.log = Logger.with_default_handlers(name=__name__) def __repr__(self): return "" % (self.host, self.port) def __getattr__(self, name): if name not in COMMANDS and name not in SENSORS: raise AttributeError('%s is not a valid method' % name) async def command(*args): if name in SENSORS: keys = ['%s.%s' % (name, axis) for axis in ('x', 'y', 'z')] params = dict(zip(keys, args)) self.input(params) elif name == 'literal': for char in args[0]: path = '/keypress/%s_%s' % (COMMANDS[name], quote_plus(char)) await self.request('POST', path) elif name == 'search': keys = ['title', 'season', 'launch', 'provider', 'type'] params = dict(zip(keys, args)) path = '/search/browse' await self.request('POST', path) else: path = '/keypress/%s' % COMMANDS[name] await self.request('POST', path) return command def __getitem__(self, key): key = str(key) app = self._app_for_name(key) if not app: app = self._app_for_id(key) return app def _app_for_name(self, name): for app in self.apps: if app.name == name: return app def _app_for_id(self, app_id): for app in self.apps: if app.id == app_id: return app async def request(self, method, path, params=None): url = 'http://%s:%s%s' % (self.host, self.port, path) # self.log.debug('%s', url) try: async with self.session.request(method, url, params=params) as res: data = await res.read() return data except client_exceptions.ClientError as err: raise ValueError(err) @property async def apps(self): resp = await self.request('get', '/query/apps') applications = deserialize_apps(resp, self) for a in applications: a.roku = self return applications @property async def active_app(self): resp = await self.request('get', '/query/active-app') active_app = deserialize_apps(resp, self) if type(active_app) is Application: return active_app elif type(active_app) is list: return active_app[0] else: return None @property async def device_info(self): resp = await self.request('get', '/query/device-info') root = ET.fromstring(resp) self.roku_type = "Box" if root.find('is-tv') is not None and root.find( 'is-tv').text == "true": self.roku_type = "TV" elif root.find('is-stick') is not None and \ root.find('is-stick').text == "true": self.roku_type = "Stick" self.dinfo = DeviceInfo( model_name=root.find('model-name').text, model_num=root.find('model-number').text, software_version=root.find('software-version').text, software_build=root.find('software-build').text, serial_num=root.find('serial-number').text, user_device_name=root.find('user-device-name').text, roku_type=self.roku_type ) return self.dinfo # @property # async def sw_info(self): # response = await self.request('get' 'query/device-info') @property def commands(self): return sorted(COMMANDS.keys()) @property async def power_state(self): resp = await self.request('get', '/query/device-info') root = ET.fromstring(resp) if root.find('power-mode').text: return root.find('power-mode').text def icon(self, app): return self.request('get', '/query/icon/%s' % app.id) async def launch(self, app): if app.roku and app.roku != self: raise AioRokuException('this app belongs to another Roku') launch_url = '/launch/%s' % app.id return await self.request('post', launch_url, {'contentID': app.id}) async def store(self, app): return await self.request('post', '/launch/11', params={'contentID': app.id}) async def input(self, params): return await self.request('post', '/input', params=params) async def touch(self, x, y, op='down'): if op not in TOUCH_OPS: raise AioRokuException('%s is not a valid touch operation' % op) params = { 'touch.0.x': x, 'touch.0.y': y, 'touch.0.op': op, } await self.input(params) def deserialize_apps(doc, roku_inst): applications = [] root = ET.fromstring(doc) if root.find('screensaver') is not None: app_node = root.find('screensaver') return Application(roku_id=app_node.attrib['id'], version=app_node.attrib['version'], name=app_node.text, is_screensaver=True, roku=roku_inst) app_node = root.find('app') for elem in root: app = Application(roku_id=elem.get('id'), version=elem.get('version'), name=elem.text, is_screensaver=False, roku=roku_inst) applications.append(app) return applications if __name__ == '__main__': pass PK6N!!aioroku/discovery.py""" Code adapted from Dan Krause. https://gist.github.com/dankrause/6000248 http://github.com/dankrause """ import socket import six from six.moves import http_client ST_DIAL = 'urn:dial-multiscreen-org:service:dial:1' ST_ECP = 'roku:ecp' class _FakeSocket(six.BytesIO): def makefile(self, *args, **kw): return self class SSDPResponse(object): def __init__(self, response): self.location = response.getheader('location') self.usn = response.getheader('usn') self.st = response.getheader('st') self.cache = response.getheader('cache-control').split('=')[1] def __repr__(self): return 'Lt9Q}} ) ϛ ,jkÍ;a)z_@EwxnXM,oR.ص:zvXBX+3IL(mEU}Կ׻UKC6XIm;RA7»ĉ ڑ q[[i&O?(~hC\\hCX+ODUta-qߠP)֖}^6nz})eQiyH#]>/ ɗzmN8uM&=<ys;VGaebhs\}j1RpJB#떕:^( #a.bV6cYߋ).ZUEwNWwg%_PK!Hpaioroku-0.0.2.dist-info/RECORD}˖k@y? i%TSD4)L,רE.<$'>/п1dznbU‹q{\Uᆢ1)(繏n5xElXe ޒT֞)XRb^ *4oFr]F^v(N FldH1*dzr}$MĊ1ik,䀵2X/TD.mG(8dg'#S/~a7HXKɩhxkԕٛ_Y!'r3n/,e^hE?~}pF7u;!hɥjJ/W]oW1%;]FMDW k8;E˘ȇვjT'XЧzOpi[u9$OB,~=:0oPKlN aioroku/__init__.pyPKJmNAD!!aioroku/core.pyPK6N!!"aioroku/discovery.pyPKmmN9KK*aioroku/example.pyPK-lN~,aioroku/util.pyPK6N.aioroku-0.0.2.dist-info/LICENSEPK!HMuSa4aioroku-0.0.2.dist-info/WHEELPK!H #Ь+ _5aioroku-0.0.2.dist-info/METADATAPK!Hp7aioroku-0.0.2.dist-info/RECORDPK o9