cat / goq
clone

turns a linux box into a private steam streaming appliance: plug in, tap your key, play from another computer

fix/architecture-review-findings
goq / decky / main.py
main.py 2.1K
1"""Decky RPC facade for the local Sigil appliance controller."""23import decky45from goq_sigil import Controller, ControllerError6from goq_sigil.runner import Runner789class Plugin:10 async def _main(self):11 self.controller = Controller(Runner(decky.DECKY_USER_HOME))12 decky.logger.info("Goq Sigil controller loaded")1314 async def _unload(self):15 # systemd owns Sigil. A Decky reload must never stop the daemon.16 decky.logger.info("Goq Sigil controller unloaded")1718 async def _call(self, operation, *arguments):19 try:20 value = await operation(*arguments)21 return {"ok": True, "value": value}22 except ControllerError as error:23 if error.detail:24 decky.logger.warning(25 "Sigil controller operation failed: %s (%s)",26 error.code,27 error.detail,28 )29 else:30 decky.logger.warning(31 "Sigil controller operation failed: %s", error.code32 )33 return {"ok": False, "error": error.code}34 except Exception:35 decky.logger.exception("Unexpected Sigil controller failure")36 return {"ok": False, "error": "internal_error"}3738 async def get_snapshot(self):39 return await self._call(self.controller.get_snapshot)4041 async def get_config(self):42 return await self._call(self.controller.get_config)4344 async def validate_config(self, request):45 return await self._call(self.controller.validate_config, request)4647 async def apply_config(self, request):48 return await self._call(self.controller.apply_config, request)4950 async def restart_service(self):51 return await self._call(self.controller.restart_service)5253 async def rollback_pending(self, transaction):54 return await self._call(self.controller.rollback_pending, transaction)5556 async def reset_enrollment(self, expected_host_fingerprint):57 return await self._call(58 self.controller.reset_enrollment, expected_host_fingerprint59 )