Python gi.repository.GLib 模块，threads_init() 实例源码

def __init__(self, f_bottom_bar_changer: callable, f_bottom_bar_adder: callable):
        """ Constructor.

        :param f_bottom_bar_changer: change the bottom bar main text.
        :param f_bottom_bar_adder: add to the actions list in bottom bar.
        """
        # will be set from UI class
        self.window = None
        self.document = None
        self.view = None
        self.tab = None
        # initialize the settings class
        DictonatorSettings()
        # Using like a global function
        self.bottom_bar_text_set = f_bottom_bar_changer
        self.bottom_bar_add = f_bottom_bar_adder
        GLib.threads_init()
        self.recogniser = SpeechRecogniser(self.action_handler)

def _oldGiInit():
    """
    Make sure pygtk and gi aren't loaded at the same time, and import Glib if
    possible.
    """
    # We can't immediately prevent imports, because that confuses some buggy
    # code in gi:
    _glibbase.ensureNotImported(
        _PYGTK_MODULES,
        "Introspected and static glib/gtk bindings must not be mixed; can't "
        "import gireactor since pygtk2 module is already imported.")

    global GLib
    from gi.repository import GLib
    if getattr(GLib, "threads_init", None) is not None:
        GLib.threads_init()

    _glibbase.ensureNotImported([], "",
                                preventImports=_PYGTK_MODULES)

def __init__(self):
        # initially set the standard logger
        self.set_logger(logging.getLogger(__name__))
        # initially set an empty configuration
        self.set_config(configparser.ConfigParser())
        # set up the quit signals
        self.setup_signals(
            signals = [signal.SIGINT, signal.SIGTERM, signal.SIGHUP],
            handler = self.please_stop_now
        )

        # use glib as default mailoop for dbus
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        dbus.mainloop.glib.threads_init()
        GLib.threads_init()

        self.loop = GLib.MainLoop() # create mainloop

        self.systembus = dbus.SystemBus() # the system bus
        self.systembus.request_name(CO2MONITOR_BUSNAME) # request the bus name
        bus_name = dbus.service.BusName(CO2MONITOR_BUSNAME, self.systembus) # create bus name
        # register the object on the bus name
        dbus.service.Object.__init__(self, bus_name, CO2MONITOR_OBJECTPATH)

def __init__(self, devicefile):
        # by default, log!
        self.do_data_logging = True
        # initially set the standard logger
        self.set_logger(logging.getLogger(__name__))
        # initially set an empty configuration
        self.set_config(configparser.ConfigParser())

        # use glib as default mailoop for dbus
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        dbus.mainloop.glib.threads_init()
        GLib.threads_init()

        self.systembus = dbus.SystemBus() # the system bus
        self.systembus.request_name(CO2MONITOR_BUSNAME) # request the bus name
        bus_name = dbus.service.BusName(CO2MONITOR_BUSNAME, self.systembus) # create bus name

        self.devicefile = devicefile

        # set up the device
        self.device = device.co2device(self.devicefile)

        # register the object on the bus name
        objectpath = "/".join([CO2MONITOR_OBJECTPATH,
            utils.devicefile2objectname(self.devicefile)])
        dbus.service.Object.__init__(self, bus_name, objectpath)
        self.update_status(_("idle"))
        threading.Thread.__init__(self)


    # set the config

def run(self):
        logging.info('D-Bus process started')
        GLib.threads_init()  # allow threads in GLib
        GLib.idle_add(self._idleQueueSync)

        DBusGMainLoop(set_as_default=True)
        dbusService = SessionDBus(self.taskQueue, self.resultQueue)

        try:
            GLib.MainLoop().run()
        except KeyboardInterrupt:
            logging.debug("\nThe MainLoop will close...")
            GLib.MainLoop().quit()
        return