server.py 文件源码

def _run(self):
        dm = UChromaDeviceManager()

        atexit.register(UChromaServer.exit, self._loop)

        dbus = DeviceManagerAPI(dm, self._logger)
        power = PowerMonitor()

        for sig in (signal.SIGINT, signal.SIGTERM):
            self._loop.add_signal_handler(sig, self._shutdown_callback)

        try:
            dbus.run()
            power.start()

            ensure_future(dm.monitor_start(), loop=self._loop)

            self._loop.run_forever()

        except KeyboardInterrupt:
            pass

        finally:
            for sig in (signal.SIGTERM, signal.SIGINT):
                self._loop.remove_signal_handler(sig)

            power.stop()

            self._loop.run_until_complete(asyncio.wait( \
                    [dm.close_devices(), dm.monitor_stop()],
                    return_when=futures.ALL_COMPLETED))