From e25724a85aa20d24c6e1afc9da55e771878b051a Mon Sep 17 00:00:00 2001 From: Koushik Dutta Date: Thu, 4 Nov 2021 14:41:03 -0700 Subject: [PATCH] server/sdk: python rpc wip --- rpc/.vscode/launch.json | 2 +- rpc/rpc.py | 15 +- sdk/bin/scrypted-webpack.js | 20 +- server/python/__pycache__/rpc.cpython-39.pyc | Bin 0 -> 10928 bytes server/python/plugin-remote.py | 111 ++++++ server/python/rpc.py | 382 +++++++++++++++++++ server/src/plugin/plugin-api.ts | 2 +- server/src/plugin/plugin-host.ts | 141 ++++--- server/src/plugin/plugin-remote.ts | 17 +- server/src/scrypted-main.ts | 43 ++- server/src/state.ts | 4 +- 11 files changed, 650 insertions(+), 87 deletions(-) create mode 100644 server/python/__pycache__/rpc.cpython-39.pyc create mode 100644 server/python/plugin-remote.py create mode 100644 server/python/rpc.py diff --git a/rpc/.vscode/launch.json b/rpc/.vscode/launch.json index 41d62525d..64885145d 100644 --- a/rpc/.vscode/launch.json +++ b/rpc/.vscode/launch.json @@ -9,7 +9,7 @@ "type": "python", "request": "launch", "program": "${workspaceFolder}/rpc.py", - "console": "integratedTerminal" + "console": "integratedTerminal", }, { "type": "pwa-node", diff --git a/rpc/rpc.py b/rpc/rpc.py index 5b300d37f..3fb0a1a9c 100644 --- a/rpc/rpc.py +++ b/rpc/rpc.py @@ -1,3 +1,4 @@ +import debugpy from asyncio.events import AbstractEventLoop from asyncio.futures import Future from typing import Callable @@ -337,7 +338,7 @@ async def readLoop(loop, peer, reader): pass -async def main(loop: AbstractEventLoop): +async def async_main(loop: AbstractEventLoop): reader, writer = await asyncio.open_connection( '127.0.0.1', 3033) @@ -368,6 +369,12 @@ async def main(loop: AbstractEventLoop): # # pokemon = json.loads(contents) # # print(pokemon['name']) -loop = asyncio.get_event_loop() -loop.run_until_complete(main(loop)) -loop.close() + +def main(): + loop = asyncio.get_event_loop() + loop.run_until_complete(async_main(loop)) + loop.close() + + +if __name__ == "__main__": + main() diff --git a/sdk/bin/scrypted-webpack.js b/sdk/bin/scrypted-webpack.js index de6e111ba..61e8e3434 100755 --- a/sdk/bin/scrypted-webpack.js +++ b/sdk/bin/scrypted-webpack.js @@ -17,8 +17,26 @@ const rimraf = require('rimraf'); const webpack = require('webpack'); const esbuild = require('esbuild'); +if (fs.existsSync(path.resolve(cwd, 'src/main.py'))) { -if (false) { + let out; + if (process.env.NODE_ENV == 'production') + out = path.resolve(cwd, 'dist'); + else + out = path.resolve(cwd, 'out'); + + const resolved = path.resolve(cwd, 'src/main.py'); + + const zip = new AdmZip(); + + zip.addLocalFile(resolved); + + const zipfs = path.join(cwd, 'fs'); + if (fs.existsSync(zipfs)) + zip.addLocalFolder(zipfs, 'fs'); + zip.writeZip(path.join(out, 'plugin.zip')); +} +else if (false) { let out; if (process.env.NODE_ENV == 'production') diff --git a/server/python/__pycache__/rpc.cpython-39.pyc b/server/python/__pycache__/rpc.cpython-39.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c66cb4c7e3d6691d274c971a20e07048846f318b GIT binary patch literal 10928 zcmbta+m9UCS+9Fn_w@8!J)W_>>#;Yx*`1to+)@Dgbska#12`Tf4C>dTBJ zAatuvpE~E%xqRn4zwh^b)vejtih<8xzHzVh>I;VP&%BxZIJkKkS9HxZ3}q;@YwVc~ z(`4T2T6=cGmb~3{_S}Xmd8g|&Jk&egQllile#4jFa-%H2l}2Tv-0jZnRU6g4*~V=C ze63Ln=7vUNKA1mR2o{eqp5&HBx#dyr%qX`q%AI9Sd8%~JXsmw4P`)bPGgLWP8`_O? zxUZ-g+|LZH#(CUT)hzC2ad!cCH8qF3Iov&gyLq*My9L}msTS1|`nwpIYWabw&Zw1p zPGeo2RjVky6kJwox2%s${h8p>(7@=<4WoYUXOv~#Oc$C8kS zX5`*!b@zjZhOT0Se}C@kyS?uIUJzYw zt4xT~Nf+Phx8Dk){cil)2koHGnQyi!jpD;RJA4IKG>;^KY??*_xir!74MUeu2q}B* z*8a}TI4xmSQEMlFfSf4kZficI7npGASu%@U>NlI6uoE|%>$r<}FbvZ(2j@>u=}OVE z?(wOxwHGv-X{FiR>#6-N^VMea_I|6I*Esx5-6(FgZ|NoUJ2A9Og5&EsBovVd)HfZ| zv83#OD^O)YkhZdwgI`Cv5U7{VV)i!z-D!0@gFv?_y5hq;l}$v+<&MNB(|c&?MHK28 zCMQHyhGt}4K7-mwdNl`Ur^dQ6YV&^?RiAsza5E|xbylAkaBX_PiB_9+S1+@tGfb$a z`aBcLcVaPP;jf|MX7sJf#Vf_CMccWD6O$fk`V>Ujc!Ub`4L&M|(AX7_+O-n% z*eVX8t)E1LwA^e0@VW=hroM>#hxxRemfnd{@77)ZPV7l>7TbW#X7h6#VnV;B=a#j3 z%x23W<9<$bJl^alQ1MxI%tc_c9nF$co9&|TWn9s#NDANzYyIt`QMD*$NRby#+qAxQc6^RSt_;MFnx1MKH1Tsq0AsdKZQo8 zFp7F5eGoz^98wXSa%0zqR$0d&L7BmIG!K-%#N;zft|G}1NPnIM?EmBx+59WGiW5%r z4~75W02`QN#%8^80?AVw$0W5|;X&$jJIKpnCs1fP@iK?8#z;=kj$8gBNAr>R=4#1z zar9Z*#x?aB9sTdXLKa8=9qa)NtkrNjU@U9ftgENhGkEfnDnp+?otC)q zIQ>6^xQ#*eosf)4t+Ocg`Yqksi_&Vh2c@NYbOM#mj_wY?r_)+L2o*?mMjoSdK?i%i zICw8;-O8Sb(k1r!>X=DrJV!nbS{SO-(*ekOc#5m=6v_!8)?{RJgq{ygeI2C*3^0FE z!>OC^;VLwIK40hvq3VtxUOEd*9D}OG})*VxCDd)bSpp|&Wx)0aF zG-7*b9F~&O5v&C7lM=wtjlH-8Z8Jp6ya(2gj31fXmMSU#UiHw&(|+O~lf#!cUr${* zyRa=ze9#Y4tD`bhe+`qBmA##MnF#>N{3;45{av^&r=(Bl`Z2C(6-i=XuBOZtIWi5p zlY)oRi{?J*-$3%vYWTX}zE0YavDh&RaF;?{Il~7mUL5a94mwvhM7yGwKw?8xFQ*@%gGR-zzoYVlVzkwtU9X?f(kq)TlbwBq2+N#(4Uvnh$8BelrzQH1jppSQ8? zskbf8qbOLWR9|OP0kTWTHE@Z0XuwnO&1Ks%^Y70+zlxFytYQT|#JZi~K915DRi})N ziLR#9T!~MWi}vlh+3;?4y4^t4%?V+SB$_&`jW{3nVwi(!Ay8J-9quk&x{QL-^-8UH zYf7i7-;^kKFhZQdNF41aP6$6SF_pDJon1{V+{5lDAZudlXA>LuF7APxGTGARzz=)t znV&_C8I*9#W3S)lT?Kc!J2>_L;VFop!xrNi>UVf6YaB0hA=}N@wC?E)Xzrj@ZrSI} zd2>LIgrIfhRLd#TN}q~HSe8o@lbCurv7*%&xuF@GcML2S#|bP1_xe-H3|C_=-2zK_ zAu)HI(9hcLx`!ZnyWUX=wZ^e^=s}3d@1qtPm@w&+_Lmfm7jPWGi#4A%15PQ%wzKq2OiOo<;0iqD;OU!AdFv5%11Le zoaSLAspx-A5Hn%qBB>0`-6~hcrvONgogZdge&FUaoW+>6p@|jYS5?je0Gc(@*j+CH z3=oWW=MpNz|G3)v%x1%XJ&%PnmI{74L24VbMWK+^#!7K}f~z{|0vs7JE}1I^x}2_! ziY9s#CoEkaRiWPx4)_cx@e?OXrw3^ul#sOs6-Ou`f0D*EV&hJLO@J8;%GcuL>WQ5G==h2{XUPwEJEQX(0Sw_e< z*pChVnwD}Bo32hYDmo8nXZ6cyEa;im#vRMaByOfvS>h|b{V;|L@MRR$XR>3-qmouN zB4r&NyeYo#rEw4WMobJVQqN4P4U^tj95)_`bopc%9goJsNMOPYfEo?pdl5RB$g2~$ zAnW=r<{7<@grP9UteMWEis?VHEaWUAbH%Luhvzv~)x_6w9yzvSd1%3u?JMTFqMd2| z+$%X&1ugNEAiF$W8YB5=V!&it&{LL^pf2!2lQd#1GV_A3@VOI9N&W3kD6yhHMP-5K zBhpe->QAF~3c!hGbQTEW5&fMQlmbTD0?}}WIMFc(M&be$yu`b2?%MZ_4FhFP;^tHW z*mlpSNhdnh?aekinfkD6(kb!!z&OAG$T%|p!_+&dlGT7yh-np6AA`5Q(QWN*sn$z> zg@*rz%NSfP&>HTjsMgr-7mabe=pdw}yI_(!*iNBT+xpi~Bm>q<6Sn+>)4I;*th+iB z@l8HL>z(?=Xn&5oh#^xtXHW=ic*R^X2P1J84W~AjE25-jbN>;U#EgLSK=j;p+_E#< zkwD@qvk9;w$b$v~pJ7Fihg}d`!90nc1?Vqhm3V(^mEvBNS%rfvPVc+SPXXQRO$M1K?W%|R)#SB#I1xZlQNGo0f&8-v|otrG&_I%-{@ab3>&I&Pniz_kSK z9Mx&V6X#>HcO-xUGZJCV(CGro>eVsoG}gv2aRTrgtK+I^IMxArX`CBZj}B^%!g**# zfOV=@%(@>no*q9rRXc4U^exVRoyj)Gt4@SIbA1~rzQjEL8m*$&kr==lVo-)d$V2Uj zNRqROL%^$C9UV(>4A@1K5}e|%S+NF7#mSi%7)V5JH>Ir9ZaPjMjvW)vd<$b7y5Dm{yt3l& zHRBpo@~D#7^nO0F!Y^W;CA|8vv_CylnL$Q+N&~9;jLf>6&sqplWW|oEkk+gjPw<)=g%!5yI*?lRVS2YkpcWA>W z{6*!+J(}Yxlamkw0fFaFy*t z_$ADAS!VhVvcqRF`k8#DB&e%LXOmg>Y8qip=Dd<{R{(C%#}`DZ_QB;`s^=ipvv_JP zlk4NPPkfox%?!C*iGGOxB8DEun~0V&%6u*&CdhfWP&y7q4%FRVhvPa&Rez zYZ=4<#&y*TBB{RFx)Z=F;wuW2zuqWqw>n+Cw$xwa6ASy{t+01DT+duMw8%UsdRLjZ zBzBk3JUBmU+iw?s);c2qcxte;AKq+*3SQ*8Mm1PoPuI`sUu8dn%Y=%|zh=-gE%SA1 zu!T2fX=S^o_gZl?_@J%dWveYF;%{GPj&xDK!GsVl^i_X{xfhs_jAST3nsYi!za8CW zZy5C^nxuBW*Vh}&`J{uJ;D7a6W(^2ynbgHe6iR9anTyESX6j{f&0?Itg_4E`lhnHv z=(kwyij6o)wPrJUypg@1{3kSweu{($jdv2@g5H05u4OI?B3B<(!6f-saPiSOyJl6g zdZ)_Y5sUO5dBE^Ra~UY@0o5(DCbVMNTC|qopE53x_W%xBK}!#9s%Ybxt9gzVfUyHa z5FYtxll4B2aeUm*qjqqnK>w)~AR2A8x2E8ql#Q?~nz=sHh z6;#pE5$Z%GV6y1JKQv{ZvJH|5O9!PdfkNwDWCout)|+|tQ_-BqP)0{|Gc9*i;m}YH zqKHLAe82(`mQALo8Po zDTu`a3DFg$zlnlIRq!o81fcHHYMQ!B4D8n_6jBKTnuzazdhX!KVqbu1Z!$`Dn^mTRMzsE$DJWfRQegzfWlIS@kh#ca(EZiw`e`kzcQzR3rp)^?ypR-!L7Aovr7!|wDo+%S~c zQi#Lg%~Kd;rjQ6SA<#^R*o=Pz+N2KAv>AY31_4$CUiFVq0@%5SpD7HZx>0^|cquzd z3Vs(zZiapDTZiU{CW7yzDb79r&;q@)hUSL|+}}5L@fLc+5bWf?qY#_J^9bHEbdE@( zoglbh^u$oV0*%ar=;h5p`T5U%;mS39pYQw~HgM|vb7DMxa{jmTiSf@!IQ{rr5`vQ! z;hX_VvZ}!%N}b~zu(_$>HQ7`|I}gDq2MBNKuOozKa{@UJvcG2w3&$^6xbPr*Gc_xz zvmbAN5$}nAfPqt2?e8Im=HAtv7=NqTI*5Y^{~*$>PADe>uT;chQ<>9>1iE6pCEXFM zhn4c=!Q4+0)xVCG?~iw?0<|v;?_gCFrn!gSr{P8D1rP1$gC4}(iHx4b4Gj?9!xUT` z1`eA^=i5293!^5hbT`uygIGNT;v6g|Zn%ADPt+*&1Eqp~1lq@go7&yp4y@Kk$m549 znW>&=-Avxa+vwmXmy>}P;?fr%w}^?>siWXwBAP+HYs~fVvIA!$3;{L(D>>pBn)WW%K5OUNJ-mOBy@;-~w%S@u z3+Ew`0K5?oV*PElv|CE0?l!P4PHp}(;6*m-vC7%%^}3nC5Uwo( z_EuyfYs@U&*kDL5L%NOZPj6&PMnsb)0Zxc#{GA7#C5st-ZQq+O*Zf+!cBb}y None: + self.send = send + + def __apply__(self, proxyId: str, oneWayMethods: list[str], method: str, argArray: list): + args = [] + for arg in argArray: + args.append(self.serialize(arg, False)) + + rpcApply = { + 'type': 'apply', + 'id': None, + 'proxyId': proxyId, + 'argArray': args, + 'method': method, + } + + if not oneWayMethods or method not in oneWayMethods: + rpcApply['oneway'] = True + self.send(rpcApply) + future = Future() + future.set_result(None) + return future + + async def send(id: str, reject: Callable[[Exception], None]): + rpcApply['id'] = id + self.send(rpcApply, reject) + return self.createPendingResult(send) + + def kill(self): + self.killed = True + + def createErrorResult(self, result: any, name: str, message: str, tb: str): + result['stack'] = tb if tb else 'no stack' + result['result'] = name if name else 'no name' + result['message'] = message if message else 'no message' + + def serialize(self, value, requireProxy): + if (not value or (not requireProxy and type(value) in jsonSerializable)): + return value + __remote_constructor_name = 'Function' if callable(value) else value.__proxy_constructor if hasattr( + value, '__proxy_constructor') else type(value).__name__ + proxyId = self.localProxied.get(value, None) + if proxyId: + ret = { + '__remote_proxy_id': proxyId, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + } + return ret + + __proxy_id = getattr(value, '__proxy_id', None) + __proxy_peer = getattr(value, '__proxy_peer', None) + if __proxy_id and __proxy_peer == self: + ret = { + '__local_proxy_id': __proxy_id, + } + return ret + + serializerMapName = self.constructorSerializerMap.get( + type(value).__name__) + if serializerMapName: + __remote_constructor_name = serializerMapName + serializer = self.nameDeserializerMap.get(serializerMapName, None) + serialized = serializer.serialize(value) + if not serialized or (not requireProxy and type(serialized).__name in jsonSerializable): + ret = { + '__remote_proxy_id': None, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + '__serialized_value': value, + } + return ret + + proxyId = str(self.proxyCounter) + self.proxyCounter = self.proxyCounter + 1 + self.localProxied[value] = proxyId + self.localProxyMap[proxyId] = value + + ret = { + '__remote_proxy_id': proxyId, + '__remote_constructor_name': __remote_constructor_name, + '__remote_proxy_props': getattr(value, '__proxy_props', None), + '__remote_proxy_oneway_methods': getattr(value, '__proxy_oneway_methods', None), + } + + return ret + + def finalize(self, id: str): + pass + + def newProxy(self, proxyId: str, proxyConstructorName: str, proxyProps: any, proxyOneWayMethods: list[str]): + proxy = RpcProxy(self, proxyId, proxyConstructorName, + proxyProps, proxyOneWayMethods) + wr = weakref.ref(proxy) + self.remoteWeakProxies[proxyId] = wr + weakref.finalize(proxy, lambda: self.finalize(proxyId)) + return proxy + + def deserialize(self, value): + if not value: + return value + + if type(value) != dict: + return value + + __remote_proxy_id = value.get('__remote_proxy_id', None) + __local_proxy_id = value.get('__local_proxy_id', None) + __remote_constructor_name = value.get( + '__remote_constructor_name', None) + __serialized_value = value.get('__serialized_value', None) + __remote_proxy_props = value.get('__remote_proxy_props', None) + __remote_proxy_oneway_methods = value.get( + '__remote_proxy_oneway_methods', None) + + if __remote_proxy_id: + weakref = self.remoteWeakProxies.get('__remote_proxy_id', None) + proxy = weakref() if weakref else None + if not proxy: + proxy = self.newProxy(__remote_proxy_id, __remote_constructor_name, + __remote_proxy_props, __remote_proxy_oneway_methods) + return proxy + + if __local_proxy_id: + ret = self.localProxyMap.get(__local_proxy_id, None) + if not ret: + raise RpcResultException( + None, 'invalid local proxy id %s' % __local_proxy_id) + return ret + + deserializer = self.nameDeserializerMap.get( + __remote_constructor_name, None) + if deserializer: + return deserializer.deserialize(__serialized_value) + + return value + + async def handleMessage(self, message: any): + try: + type = message['type'] + if type == 'param': + result = { + 'type': 'result', + 'id': message['id'], + } + + try: + value = self.params.get(message['param'], None) + value = await maybe_await(value) + result['result'] = self.serialize( + value, message.get('requireProxy', None)) + except Exception as e: + tb = traceback.format_exc() + self.createErrorResult( + result, type(e).__name, str(e), tb) + + self.send(result) + + elif type == 'apply': + result = { + 'type': 'result', + 'id': message['id'], + } + method = message.get('method', None) + + try: + target = self.localProxyMap.get( + message['proxyId'], None) + if not target: + raise Exception('proxy id %s not found' % + message['proxyId']) + + args = [] + for arg in (message['argArray'] or []): + args.append(self.deserialize(arg)) + + value = None + if method: + if not hasattr(target, method): + raise Exception( + 'target %s does not have method %s' % (type(target), method)) + invoke = getattr(target, method) + value = await maybe_await(invoke(*args)) + else: + value = await maybe_await(target(*args)) + + result['result'] = self.serialize(value, False) + except Exception as e: + print('failure', method, e) + tb = traceback.format_exc() + self.createErrorResult( + result, type(e).__name, str(e), tb) + + if not message.get('oneway', False): + self.send(result) + + elif type == 'result': + future = self.pendingResults.get(message['id'], None) + if not future: + raise RpcResultException( + None, 'unknown result %s' % message['id']) + del message['id'] + if hasattr(message, 'message') or hasattr(message, 'stack'): + e = RpcResultException( + None, message.get('message', None)) + e.stack = message.get('stack', None) + e.name = message.get('name', None) + future.set_exception(e) + return + future.set_result(self.deserialize( + message.get('result', None))) + elif type == 'finalize': + local = self.localProxyMap.pop( + message['__local_proxy_id'], None) + self.localProxied.pop(local, None) + else: + raise RpcResultException( + None, 'unknown rpc message type %s' % type) + except Exception as e: + print("unhandled rpc error", self.peerName, e) + pass + + async def createPendingResult(self, cb: Callable[[str, Callable[[Exception], None]], None]): + # if (Object.isFrozen(this.pendingResults)) + # return Promise.reject(new RPCResultError('RpcPeer has been killed')); + + id = str(self.idCounter) + self.idCounter = self.idCounter + 1 + future = Future() + self.pendingResults[id] = future + await cb(id, lambda e: future.set_exception(RpcResultException(e, None))) + return await future + + async def getParam(self, param): + async def send(id: str, reject: Callable[[Exception], None]): + paramMessage = { + 'id': id, + 'type': 'param', + 'param': param, + } + self.send(paramMessage, reject) + return await self.createPendingResult(send) + +# c = RpcPeer() + + +async def readLoop(loop, peer, reader): + async for line in reader: + try: + message = json.loads(line) + asyncio.run_coroutine_threadsafe(peer.handleMessage(message), loop) + except Exception as e: + print('read loop error', e) + pass + + +async def async_main(loop: AbstractEventLoop): + reader, writer = await asyncio.open_connection( + '127.0.0.1', 3033) + + async def send(message, reject): + jsonString = json.dumps(message) + writer.write(bytes(jsonString + '\n', 'utf8')) + try: + await writer.drain() + except Exception as e: + if reject: + reject(e) + + peer = RpcPeer(send) + peer.params['print'] = print + + async def consoleTest(): + console = await peer.getParam('console') + await console.log('test', 'poops', 'peddeps') + + await asyncio.gather(readLoop(loop, peer, reader), consoleTest()) + print('done') + + # print("line %s" % line) + + # async with aiofiles.open(0, mode='r') as f: + # async for line in f: + # print("line %s" % line) + # # pokemon = json.loads(contents) + # # print(pokemon['name']) + + +def main(): + loop = asyncio.get_event_loop() + loop.run_until_complete(async_main(loop)) + loop.close() + + +if __name__ == "__main__": + main() diff --git a/server/src/plugin/plugin-api.ts b/server/src/plugin/plugin-api.ts index 4a0bceebd..e5638bdcf 100644 --- a/server/src/plugin/plugin-api.ts +++ b/server/src/plugin/plugin-api.ts @@ -140,7 +140,7 @@ export interface PluginRemote { loadZip(packageJson: any, zipData: Buffer, options?: PluginRemoteLoadZipOptions): Promise; setSystemState(state: {[id: string]: {[property: string]: SystemDeviceState}}): Promise; setNativeId(nativeId: ScryptedNativeId, id: string, storage: {[key: string]: any}): Promise; - updateDescriptor(id: string, state: {[property: string]: SystemDeviceState}): Promise; + updateDeviceState(id: string, state: {[property: string]: SystemDeviceState}): Promise; notify(id: string, eventTime: number, eventInterface: string, property: string|undefined, value: SystemDeviceState|any, changed?: boolean): Promise; ioEvent(id: string, event: string, message?: any): Promise; diff --git a/server/src/plugin/plugin-host.ts b/server/src/plugin/plugin-host.ts index 25701baff..7d1556e6b 100644 --- a/server/src/plugin/plugin-host.ts +++ b/server/src/plugin/plugin-host.ts @@ -24,6 +24,8 @@ import { install as installSourceMapSupport } from 'source-map-support'; import net from 'net' import child_process from 'child_process'; import { PluginDebug } from './plugin-debug'; +import readline from 'readline'; +import { Readable, Writable } from 'stream'; export class PluginHost { worker: child_process.ChildProcess; @@ -88,44 +90,16 @@ export class PluginHost { this.packageJson = plugin.packageJson; const logger = scrypted.getDeviceLogger(scrypted.findPluginDevice(plugin._id)); - if (true) { - const cwd = path.join(process.cwd(), 'volume', 'plugins', this.pluginId); - try { - mkdirp.sync(cwd); - } - catch (e) { - } - - this.startPluginClusterHost(logger, { - SCRYPTED_PLUGIN_VOLUME: cwd, - }); + const cwd = path.join(process.cwd(), 'volume', 'plugins', this.pluginId); + try { + mkdirp.sync(cwd); } - else { - const remote = new RpcPeer((message, reject) => { - try { - this.peer.handleMessage(message); - } - catch (e) { - if (reject && reject) - reject(e); - } - }); - - this.peer = new RpcPeer((message, reject) => { - try { - remote.handleMessage(message); - } - catch (e) { - if (reject) - reject(e); - } - }); - - attachPluginRemote(remote, { - createMediaManager: async (systemManager) => new MediaManagerImpl(systemManager, console), - }); + catch (e) { } + this.startPluginClusterHost(logger, { + SCRYPTED_PLUGIN_VOLUME: cwd, + }, plugin.packageJson.scrypted.runtime); this.io.on('connection', async (socket) => { try { @@ -219,23 +193,82 @@ export class PluginHost { }); } - startPluginClusterHost(logger: Logger, env?: any) { - const execArgv: string[] = process.execArgv.slice(); - if (this.pluginDebug) { - execArgv.push(`--inspect=0.0.0.0:${this.pluginDebug.inspectPort}`); + startPluginClusterHost(logger: Logger, env?: any, runtime?: string) { + let connected = true; + + if (runtime === 'python') { + const args: string[] = []; + if (this.pluginDebug) { + args.push( + '-m', + 'debugpy', + '--listen', + `0.0.0.0:${this.pluginDebug.inspectPort}`, + '--wait-for-client', + path.join(__dirname, '../../python', 'plugin-remote.py'), + ) + } + + this.worker = child_process.spawn('python', args, { + // stdin, stdout, stderr, peer in, peer out + stdio: ['pipe', 'pipe', 'pipe', 'pipe', 'pipe'], + }); + + const peerin = this.worker.stdio[3] as Writable; + const peerout = this.worker.stdio[4] as Readable; + peerout.on('data', data => { + console.log(data.toString()); + }) + + this.peer = new RpcPeer((message, reject) => { + if (connected) { + peerin.write(JSON.stringify(message) + '\n', e => e && reject?.(e)); + } + else if (reject) { + reject(new Error('peer disconnected')); + } + }); + + const readInterface = readline.createInterface({ + input: peerout, + terminal: false, + }); + readInterface.on('line', line => { + this.peer.handleMessage(JSON.parse(line)); + }); + } + else { + const execArgv: string[] = process.execArgv.slice(); + if (this.pluginDebug) { + execArgv.push(`--inspect=0.0.0.0:${this.pluginDebug.inspectPort}`); + } + + this.worker = child_process.fork(require.main.filename, ['child', JSON.stringify(env)], { + stdio: 'pipe', + serialization: 'advanced', + execArgv, + }); + + this.peer = new RpcPeer((message, reject) => { + if (connected) { + this.worker.send(message, undefined, e => { + if (e && reject) + reject(e); + }); + } + else if (reject) { + reject(new Error('peer disconnected')); + } + }); + + this.worker.on('message', message => this.peer.handleMessage(message as any)); } - this.worker = child_process.fork(require.main.filename, ['child', JSON.stringify(env)], { - stdio: 'pipe', - serialization: 'advanced', - execArgv, - }); this.worker.stdout.on('data', data => { process.stdout.write(data); }); this.worker.stderr.on('data', data => process.stderr.write(data)); - let connected = true; this.worker.on('disconnect', () => { connected = false; logger.log('e', `${this.pluginName} disconnected`); @@ -248,19 +281,7 @@ export class PluginHost { connected = false; logger.log('e', `${this.pluginName} error ${e}`); }); - this.worker.on('message', message => this.peer.handleMessage(message as any)); - this.peer = new RpcPeer((message, reject) => { - if (connected) { - this.worker.send(message, undefined, e => { - if (e && reject) - reject(e); - }); - } - else if (reject) { - reject(new Error('peer disconnected')); - } - }); this.peer.peerName = this.pluginId; this.peer.onOob = (oob: any) => { @@ -519,7 +540,7 @@ export function startPluginClusterWorker() { // deleted? return; } - const {pluginId, nativeId: mixinNativeId} = await plugins.getDeviceInfo(mixinId); + const { pluginId, nativeId: mixinNativeId } = await plugins.getDeviceInfo(mixinId); const port = await plugins.getRemoteServicePort(pluginId, 'console-writer'); const socket = net.connect(port); socket.write(mixinNativeId + '\n'); @@ -626,10 +647,10 @@ class LazyRemote implements PluginRemote { await this.remoteReadyPromise; return this.remote.setNativeId(nativeId, id, storage); } - async updateDescriptor(id: string, state: { [property: string]: SystemDeviceState; }): Promise { + async updateDeviceState(id: string, state: { [property: string]: SystemDeviceState; }): Promise { if (!this.remote) await this.remoteReadyPromise; - return this.remote.updateDescriptor(id, state); + return this.remote.updateDeviceState(id, state); } async notify(id: string, eventTime: number, eventInterface: string, property: string, propertyState: SystemDeviceState, changed?: boolean): Promise { if (!this.remote) diff --git a/server/src/plugin/plugin-remote.ts b/server/src/plugin/plugin-remote.ts index 975d1a1e8..cedd10d68 100644 --- a/server/src/plugin/plugin-remote.ts +++ b/server/src/plugin/plugin-remote.ts @@ -4,7 +4,7 @@ import path from 'path'; import { ScryptedNativeId, DeviceManager, Logger, Device, DeviceManifest, DeviceState, EndpointManager, SystemDeviceState, ScryptedStatic, SystemManager, MediaManager, ScryptedMimeTypes, ScryptedInterface, ScryptedInterfaceProperty, HttpRequest } from '@scrypted/sdk/types' import { PluginAPI, PluginLogger, PluginRemote, PluginRemoteLoadZipOptions } from './plugin-api'; import { SystemManagerImpl } from './system'; -import { RpcPeer } from '../rpc'; +import { RpcPeer, RPCResultError } from '../rpc'; import { BufferSerializer } from './buffer-serializer'; import { EventEmitter } from 'events'; import { createWebSocketClass } from './plugin-remote-websocket'; @@ -271,9 +271,14 @@ interface WebSocketCallbacks { export async function setupPluginRemote(peer: RpcPeer, api: PluginAPI, pluginId: string): Promise { - peer.addSerializer(Buffer, 'Buffer', new BufferSerializer()); - const getRemote = await peer.getParam('getRemote'); - return getRemote(api, pluginId); + try { + peer.addSerializer(Buffer, 'Buffer', new BufferSerializer()); + const getRemote = await peer.getParam('getRemote'); + return await getRemote(api, pluginId); + } + catch (e) { + throw new RPCResultError('error while retrieving PluginRemote', e); + } } export interface PluginRemoteAttachOptions { @@ -325,7 +330,7 @@ export function attachPluginRemote(peer: RpcPeer, options?: PluginRemoteAttachOp __proxy_required: true, __proxy_oneway_methods: [ 'notify', - 'updateDescriptor', + 'updateDeviceState', 'setSystemState', 'ioEvent', 'setNativeId', @@ -367,7 +372,7 @@ export function attachPluginRemote(peer: RpcPeer, options?: PluginRemoteAttachOp } }, - async updateDescriptor(id: string, state: { [property: string]: SystemDeviceState }) { + async updateDeviceState(id: string, state: { [property: string]: SystemDeviceState }) { if (!state) { delete systemManager.state[id]; systemManager.events.notify(id, Date.now(), ScryptedInterface.ScryptedDevice, ScryptedInterfaceProperty.id, id, true); diff --git a/server/src/scrypted-main.ts b/server/src/scrypted-main.ts index f452eb0c8..9e74505ec 100644 --- a/server/src/scrypted-main.ts +++ b/server/src/scrypted-main.ts @@ -25,6 +25,7 @@ import httpAuth from 'http-auth'; import semver from 'semver'; import { Info } from './services/info'; import { getAddresses } from './addresses'; +import { sleep } from './sleep'; if (!semver.gte(process.version, '16.0.0')) { throw new Error('"node" version out of date. Please update node to v16 or higher.') @@ -64,23 +65,41 @@ else { let workerInspectPort: number = undefined; - const debugServer = net.createServer(socket => { + async function doconnect(): Promise { + return new Promise((resolve, reject) => { + const target = net.connect(workerInspectPort); + target.once('error', reject) + target.once('connect', () => resolve(target)) + }) + } + + const debugServer = net.createServer(async (socket) => { if (!workerInspectPort) { socket.destroy(); return; } - const target = net.connect(workerInspectPort); - socket.pipe(target).pipe(socket); - socket.on('error', () => { - socket.destroy(); - target.destroy(); - }); - target.on('error', e => { - console.error('debugger target error', e); - socket.destroy(); - target.destroy(); - }); + for (let i = 0; i < 10; i++) { + try { + const target = await doconnect(); + socket.pipe(target).pipe(socket); + socket.on('error', () => { + socket.destroy(); + target.destroy(); + }); + target.on('error', e => { + console.error('debugger target error', e); + socket.destroy(); + target.destroy(); + }); + return; + } + catch (e) { + await sleep(500); + } + } + console.warn('debugger connect timed out'); + socket.destroy(); }) listenServerPort('SCRYPTED_DEBUG_PORT', SCRYPTED_DEBUG_PORT, debugServer); diff --git a/server/src/state.ts b/server/src/state.ts index 3245034f8..0486ebe22 100644 --- a/server/src/state.ts +++ b/server/src/state.ts @@ -73,13 +73,13 @@ export class ScryptedStateManager extends EventRegistry { updateDescriptor(device: PluginDevice) { for (const plugin of Object.values(this.scrypted.plugins)) { - plugin.remote?.updateDescriptor(device._id, device.state); + plugin.remote?.updateDeviceState(device._id, device.state); } } removeDevice(id: string) { for (const plugin of Object.values(this.scrypted.plugins)) { - plugin.remote?.updateDescriptor(id, undefined); + plugin.remote?.updateDeviceState(id, undefined); } this.notify(undefined, undefined, ScryptedInterface.ScryptedDevice, ScryptedInterfaceProperty.id, id, true);