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

def copy_code(self, *args):
        """
            Copy code shows the code box for a while (10s by default)
        """
        self.timer = 0
        code = self.account.get_code()
        try:
            clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
            self.window.notification.set_message(
                _('Code "{0}" copied to clipboard'.format(str(code))))
            self.window.notification.show()
            clipboard.clear()
            clipboard.set_text(code, len(code))
            logging.debug("Secret code copied to clipboard")
        except Exception as e:
            logging.error(str(e))
        self.revealer.set_reveal_child(True)
        GLib.timeout_add_seconds(1, self.update_timer)

def remove(self, *args):
        """
            Remove an account
        """
        message = _('Do you really want to remove "%s"?' %
                    self.account.get_name())
        confirmation = ConfirmationMessage(self.window, message)
        confirmation.show()
        if confirmation.get_confirmation():
            self.window.notification.set_message(
                _('"%s" was removed' % self.account.get_name()))
            self.window.notification.set_undo_action(self.undo_remove)
            self.window.notification.show()
            self.remove_timer = self.window.notification.timeout
            GLib.timeout_add_seconds(1, self.update_remove_countdown)
        confirmation.destroy()

def timer(self, name, delay, callback, *data, **kwdata):
        """
        Runs callback after specified number of seconds. Uses
        GLib.timeout_add_seconds with small wrapping to allow named
        timers to be canceled by reset() call
        """
        method = GLib.timeout_add_seconds
        if delay < 1 and delay > 0:
            method = GLib.timeout_add
            delay = delay * 1000.0
        if name is None:
            # No wrapping is needed, call GLib directly
            method(delay, callback, *data, **kwdata)
        else:
            if name in self._timers:
                # Cancel old timer
                GLib.source_remove(self._timers[name])
            # Create new one
            self._timers[name] = method(delay, self._callback, name, callback, *data, **kwdata)

def _on_daemon_died(self, *a):
        """ Called from various places when daemon looks like dead """
        # Log stuff
        if self.alive is True:
            log.debug("Connection to daemon lost")
        if self.alive is True or self.alive is None:
            self.alive = False
            self.emit("dead")
        self.alive = False
        # Close connection, if any
        if self.connection is not None:
            self.connection.close()
            self.connection = None
        # Emit event
        # Try to reconnect
        GLib.timeout_add_seconds(self.RECONNECT_INTERVAL, self._connect)

def save_registration(self):
        self.generate_raw_data()
        buffRawData = self.builder.get_object("buffRawData")
        jsondata = buffRawData.get_text(buffRawData.get_start_iter(),
            buffRawData.get_end_iter(), True)
        try:
            os.makedirs(os.path.join(get_config_path(), "devices"))
        except: pass

        filename = self._evdevice.name.strip()
        if self._tester.driver == "hid":
            filename = "%.4x:%.4x-%s" % (self._evdevice.info.vendor,
                self._evdevice.info.product, filename)

        config_file = os.path.join(get_config_path(), "devices",
                "%s-%s.json" % (self._tester.driver, filename,))

        open(config_file, "w").write(jsondata)
        log.debug("Controller configuration '%s' written", config_file)

        self.kill_tester()
        self.window.destroy()
        GLib.timeout_add_seconds(1, self.app.dm.rescan)

def on_cbAccessMode_changed(self, cb):
        if self._tester:
            btNext = self.builder.get_object("btNext")
            target = cb.get_model().get_value(cb.get_active_iter(), 0)
            if self._tester.driver != target:
                # User changed driver that should be used, a lot of stuff has
                # to be restarted
                log.debug("User-requested driver change")
                self.kill_tester()
                cb.set_sensitive(False)
                btNext.set_sensitive(False)
                if target == "hid":
                    self._tester = Tester("hid", "%.4x:%.4x" % (
                        self._evdevice.info.vendor, self._evdevice.info.product))
                else:
                    self._tester = Tester("evdev", self._evdevice.fn)
                self._tester.__signals = [
                    self._tester.connect('ready', self.on_registration_ready),
                    self._tester.connect('error', self.on_device_open_failed),
                ]
                GLib.timeout_add_seconds(1, self._tester.start)

def do_activate(self):
        # application is already running check
        if len(StickyManager.stickylist) > 0:
            # existing app instance, create a new sticky
            st = StickyManager.create_sticky(self)
            st.connect("delete-event", self.window_close)
            st.btnNew.connect("clicked", self.window_new)
            st.show_all()
        else:
            # new instance, restore saved stickies and show them
            StickyManager.restore_stickies(self)
            for st in StickyManager.stickylist:
                st.connect("delete-event", self.window_close)
                st.btnNew.connect("clicked", self.window_new)
                st.show_all()
            # periodically autosave all stickies
            GLib.timeout_add_seconds(4, self.autosaver)

def on_working_menu_item(self, widget):
        if self.wid == 0:
            self.working_menu_item.set_label(_('Stop'))
            self.indicator.set_icon(comun.STATUS_ICON[self.theme][0])
            if self.show_notifications:
                self.notification.update('Backlight-Indicator',
                                         _('Session starts'),
                                         comun.STATUS_ICON[self.theme][0])
            self.do_the_work()
            self.wid = GLib.timeout_add_seconds(self.sample_time * 60,
                                                self.do_the_work)
        else:
            self.working_menu_item.set_label(_('Start'))
            self.indicator.set_icon(comun.STATUS_ICON[self.theme][1])
            if self.show_notifications:
                self.notification.update('Backlight-Indicator',
                                         _('Session stops'),
                                         comun.STATUS_ICON[self.theme][1])
            GLib.source_remove(self.wid)
            self.wid = 0
        if self.show_notifications:
            self.notification.show()

def test_0(self, tmpdir):
        loop = GLib.MainLoop()
        self.__called = 0

        def cb_changed(M, p, m, f, o, evt, d=None, e=None):
            print('signal', e, p, f, o, evt, d)
            assert e == 'created'
            self.__called += 1

        d = tmpdir.mkdir("ndn")
        m = DirTools.Monitor(str(d))
        [m.connect(s, cb_changed, s) for s in ['created']]
        [d.mkdir(str(i)) for i in range(ITER_COUNT)]

        GLib.timeout_add_seconds(2, lambda: loop.quit())
        loop.run()
        assert self.__called == ITER_COUNT

def main(self):
        self.indicator = appindicator.Indicator.new(self.APPINDICATOR_ID, self.ICON_OFF,
                                                    appindicator.IndicatorCategory.SYSTEM_SERVICES)
        self.indicator.set_status(appindicator.IndicatorStatus.ACTIVE)
        self.indicator.set_menu(self.build_menu())

        # This sets the handler for “INT” signal processing
        #- the one issued by the OS when “Ctrl+C” is typed.
        #The handler we assign to it is the “default” handler, which,
        #in case of the interrupt signal, is to stop execution.
        signal.signal(signal.SIGINT, signal.SIG_DFL)  #listen to quit signal

        notify.init(self.APPINDICATOR_ID)
        self.update()
        glib.timeout_add_seconds(self.UPDATE_FREQUENCY, self.update)
        gtk.main()

def on_timelog_file_changed(self, monitor, file, other_file, event_type):
        # When I edit timelog.txt with vim, I get a series of notifications:
        # - Gio.FileMonitorEvent.DELETED
        # - Gio.FileMonitorEvent.CREATED
        # - Gio.FileMonitorEvent.CHANGED
        # - Gio.FileMonitorEvent.CHANGED
        # - Gio.FileMonitorEvent.CHANGES_DONE_HINT
        # - Gio.FileMonitorEvent.ATTRIBUTE_CHANGED
        # So, plan: react to CHANGES_DONE_HINT at once, but in case some
        # systems/OSes don't ever send it, react to other events after a
        # short delay, so we wouldn't have to reload the file more than
        # once.
        if event_type == Gio.FileMonitorEvent.CHANGES_DONE_HINT:
            self.check_reload()
        else:
            GLib.timeout_add_seconds(1, self.check_reload)

def start_weather_updater(self):
        if self.weather_updater > 0:
            GLib.source_remove(self.weather_updater)
        self.update_weather()
        self.weather_updater = GLib.timeout_add_seconds(self.refresh * 3600,
                                                        self.update_weather)

def start_looking_for_internet(self):
        if self.internet_updater > 0:
            GLib.source_remove(self.internet_updater)
        if self.looking_for_internet():
            self.internet_updater = GLib.timeout_add_seconds(
                TIME_TO_CHECK, self.looking_for_internet)

def __init__(self, application):
        self.app = application
        self.observable = Observable()
        self.generate_window()
        self.generate_header_bar()
        self.generate_accounts_box()
        self.generate_no_accounts_box()
        self.generate_login_box()
        if settings.get_can_be_locked():
            self.login_box.on_lock()
        settings.connect("changed", self.bind_view)
        GLib.timeout_add_seconds(60, self.refresh_counter)
        self.main_box.show_all()

def __init__(self, message="", undo_action=None, timeout=5):
        Gtk.Revealer.__init__(self)
        self.timeout = timeout
        self.set_transition_type(Gtk.RevealerTransitionType.SLIDE_DOWN)
        self.message = message
        self.undo_action = undo_action
        self.generate_components()
        GLib.timeout_add_seconds(1, self.update_timer)

def restart_timer(self):
        self.cancel_timer()
        self._timer = GLib.timeout_add_seconds(self._timeout, self.on_timeout)

def show(self):
        self.l = Gtk.Label()
        self.l.set_name("osd-label")
        self.l.set_label(self.text)

        self.add(self.l)

        OSDWindow.show(self)
        GLib.timeout_add_seconds(self.timeout, self.quit)

def schedule_save_config(self):
        """
        Schedules config saving in 1s.
        Done to prevent literal madness when user moves slider.
        """
        def cb(*a):
            self._timer = None
            self.app.save_config()

        if self._timer is not None:
            GLib.source_remove(self._timer)
        self._timer = GLib.timeout_add_seconds(1, cb)

def schedule_save_config(self):
        """
        Schedules config saving in 3s.
        Done to prevent literal madness when user moves slider.
        """
        def cb(*a):
            self._timer = None
            self.app.save_config()

        if self._timer is not None:
            GLib.source_remove(self._timer)
        self._timer = GLib.timeout_add_seconds(3, cb)

def on_btRestartEmulation_clicked(self, *a):
        rvRestartWarning = self.builder.get_object("rvRestartWarning")
        self.app.dm.stop()
        rvRestartWarning.set_reveal_child(False)
        GLib.timeout_add_seconds(1, self.app.dm.start)

def start_working_process(self, interval, countdown):
        if self.pw > 0:
            GLib.source_remove(self.pw)
        print('interval', interval)
        print('pomodoros', self.pomodoros)
        self.pw = GLib.timeout_add_seconds(interval, countdown)

def start_ram_updater(self):
        if self.ram_updater > 0:
            GLib.source_remove(self.ram_updater)
        self.ram_update()
        self.ram_updater = GLib.timeout_add_seconds(1, self.ram_update)

def start_uptime_update(self):
        if self.uptime_updater > 0:
            GLib.source_remove(self.uptime_updater)
        self.uptime_update()
        self.uptime_updater = GLib.timeout_add_seconds(60, self.uptime_update)

def start_battery_updater(self):
        if self.battery_updater > 0:
            GLib.source_remove(self.battery_updater)
        if self.exists_battery is True:
            self.get_battery_duration()
            self.battery_updater = GLib.timeout_add_seconds(
                300, self.get_battery_duration)

def read_preferences(self):
        configuration = Configuration()
        self.first_time = configuration.get('first-time')
        self.version = configuration.get('version')
        self.autostart = configuration.get('autostart')
        self.theme = configuration.get('theme')
        self.show_notifications = configuration.get('show-notifications')
        self.show_value = configuration.get('show-value')
        self.minimum_backlight = configuration.get('minimum-backlight')
        self.maximum_backlight = configuration.get('maximum-backlight')
        self.backlight = configuration.get('backlight')
        self.sample_time = configuration.get('sample-time')
        self.autoworking = configuration.get('autoworking')
        self.change_on_ac = configuration.get('change-backlight-on-ac')
        self.value_on_ac = configuration.get('backlight-on-ac')
        self.change_on_low_power = configuration.get(
            'reduce-backlight-on-low-power')
        self.low_battery_value = configuration.get('low-battery-value')
        self.value_on_low_power = configuration.get('backlight-on-low-power')
        backlight = self.backlightManager.get_backlight()
        self.set_backlight(backlight)
        if self.show_value:
            self.indicator.set_label(str(int(self.backlight)), '')
        else:
            self.indicator.set_label('', '')
        if self.wid > 0:
            self.active_icon = comun.STATUS_ICON[self.theme][0]
            GLib.source_remove(self.wid)
            self.do_the_work()
            self.wid = GLib.timeout_add_seconds(self.sample_time * 60,
                                                self.do_the_work)
        else:
            if self.autoworking:
                self.start_automatically()
            else:
                self.active_icon = comun.STATUS_ICON[self.theme][1]

def start_automatically(self):
        if self.wid != 0:
            GLib.source_remove(self.wid)
            self.wid = 0
        self.working_menu_item.set_label(_('Stop'))
        self.indicator.set_icon(comun.STATUS_ICON[self.theme][0])
        if self.show_notifications:
            self.notification.update('Backlight-Indicator',
                                     _('Session starts'),
                                     comun.STATUS_ICON[self.theme][0])
            self.notification.show()
        self.do_the_work()
        self.wid = GLib.timeout_add_seconds(self.sample_time * 60,
                                            self.do_the_work)

def run(self):
        """
        Run the Sniffer main loop.
        """
        if self.adapter is not None:
            self._log.debug("Clearing the BlueZ device registry.")
            for path, _ in get_known_devices():
                self.adapter.RemoveDevice(path)

            self._log.debug("Registering the signals InterfacesAdded and PropertiesChanged.")
            bus = pydbus.SystemBus()
            bus.subscribe(
                sender=SERVICE_NAME,
                iface=OBJECT_MANAGER_INTERFACE,
                signal="InterfacesAdded",
                signal_fired=self._cb_interfaces_added
            )
            bus.subscribe(
                sender=SERVICE_NAME,
                iface=OBJECT_MANAGER_INTERFACE,
                signal="InterfacesRemoved",
                signal_fired=self._cb_interfaces_removed
            )
            bus.subscribe(
                sender=SERVICE_NAME,
                iface=PROPERTIES_INTERFACE,
                signal="PropertiesChanged",
                arg0=DEVICE_INTERFACE,
                signal_fired=self._cb_properties_changed
            )

            self._log.debug("Running the main loop.")
            if self.output_path is not None and self.backup_interval > 0:
                GLib.timeout_add_seconds(self.backup_interval, self._cb_backup_registry)
            if self.attempt_connection:
                GLib.timeout_add_seconds(self.queueing_interval, self._cb_connect_check)
            loop = GLib.MainLoop()
            loop.run()
        else:
            raise ValueError("Sniffer.run can only be called in a context "
                             "(e.g. `with Sniffer(...) as s: s.run()`)")

def menu_item_clicked(self, widget):
        """Callback for help menu items"""
        if widget == self.tool_widgets.get_object('get_help_menu_item'):
            # run yelp
            proc = subprocess.Popen(["yelp", "ghelp:dell-recovery"])
            # collect the exit status (otherwise we leave zombies)
            GLib.timeout_add_seconds(1, lambda proc: proc.poll() == None, proc)
        elif widget == self.tool_widgets.get_object('about_menu_item'):
            tool_selector = self.tool_widgets.get_object('tool_selector')
            if not self.about_box:
                self.about_box = Gtk.AboutDialog()
                self.about_box.set_version(check_version())
                self.about_box.set_name(_("Dell Recovery"))
                self.about_box.set_copyright(_("Copyright 2008-2012 Dell Inc."))
                self.about_box.set_website("http://www.dell.com/ubuntu")
                self.about_box.set_authors(["Mario Limonciello"])
                self.about_box.set_destroy_with_transient_for(True)
                self.about_box.set_modal(True)
                self.about_box.set_transient_for(tool_selector)
            tool_selector.set_sensitive(False)
            self.about_box.run()
            self.about_box.hide()
            tool_selector.set_sensitive(True)

#### GUI Functions ###
# This application is functional via command line by using the above functions #

def _on_request_interest(self, name, skeleton, fd_list, fd_variant, first_segment):
        self.emit('interest', Name(name), Interest(name), None, None, None)
        fd = fd_list.get(fd_variant.get_handle())
        logger.debug('RequestInterest Handler: name=%s, self.name=%s, fd=%d, first_segment=%d',
                     name, self.name, fd, first_segment)
        # do we start on chunk 0 ? full file ? do we start on another chunk
        # ? we need to seek the file, subsequent calls to get the same
        # chunks have to be handled in the consumer part and folded into
        # answering to only one dbus call

        try:
            final_segment = self._get_final_segment()
        except NotImplementedError:
            # we can't handle this, let another producer come in and do it.
            self._dbus.return_error(name, 'TryAgain')
            return False

        key = name.toString()
        try:
            worker = self._workers[key]
            logger.debug('already got a worker for name %s', name)
            # self._dbus.return_error(name, 'ETOOMANY')
            return
        except KeyError:
            pass

        self._workers[key] = worker = ProducerWorker(fd, first_segment, final_segment,
                                                      self._send_chunk)
        self._dbus.return_value(name, final_segment)

        last_emited = first_segment - 1
        # XXX: is this racy ?
        GLib.timeout_add_seconds(5,
            lambda: (worker.data.n > last_emited and
                                 skeleton.emit_progress(key, worker.first_segment,
                                                        worker.data.n)) or worker.working)
        return True

def maybe_time_out():
    """Set up a timeout to exit the daemon if no interesting mounts exist"""
    if interesting_mounts_exist():
        return

    GLib.timeout_add_seconds(IDLE_TIMEOUT, timeout_cb)

def instance_launch(self, image, instancetype):
        launch.launch(config, image, instancetype)
        GLib.timeout_add_seconds(10, self.handler_poll_onetime)

def instance_kill(self, job_number, target_image):
        res = requests.get('%s/shutdown?username=%s&apikey=%s&number=%s' % (api_url, username, apikey, job_number))
        res = json.loads(res.content.decode('utf-8'))
        GLib.timeout_add_seconds(10, self.handler_poll_onetime)

def run_loop(self):
        xml = self.read_nvidia()
        idx = 0
        for gpu in xml.gpu:
            self.update_gpu(idx, gpu)
            idx = idx + 1
        # end with another loop
        GLib.timeout_add_seconds(1, self.run_loop)

def retry(self):
        logger.debug('retrying weather update after an error, retry timeout: {}'.format(self.retry_timeout))

        self.update_weather(setup_retry=False)  # Don't create a bunch of timeouts, could get ugly
        if self.error:
            self.retry_timeout += 1

            if self.retry_timeout >= 15:
                self.retry_timeout = 15

            GLib.timeout_add_seconds(60 * self.retry_timeout, self.retry)

        return False  # Don't let the timeout continue to run as we do an incremental backoff

def run(self):
        self._setup_actions()
        self._setup_menu()

        self.bus.export_action_group(BUS_OBJECT_PATH, self.action_group)
        self.menu_export = self.bus.export_menu_model(BUS_OBJECT_PATH_PHONE, self.menu)

        GLib.timeout_add_seconds(60 * 30, self.update_weather)  # TODO allow this to be configurable
        self.update_weather()

def __init__(self, interval=None, configfile=None, platform=None, datadir=None):
        self.configfile, self.datadir, self.default_config = find_config(configfile, datadir)

        if platform:
            self.platform = getattr(platforms, platform, None)
            if self.platform is None:
                logging.error("No such platform: `{}'".format(platform))
                sys.exit(1)
        else:
            self.platform = platforms.find()
        self.data = self.played = self.notified = {}

        self.icon = Gtk.StatusIcon()
        self.icon.set_name('Battray')

        self.menu = Gtk.Menu()
        refresh = Gtk.MenuItem.new_with_label('Refresh')
        refresh.connect('activate', self.cb_update)
        self.menu.append(refresh)

        about = Gtk.MenuItem.new_with_label('About')
        about.connect('activate', self.cb_about)
        self.menu.append(about)

        quit = Gtk.MenuItem.new_with_label('Quit')
        quit.connect('activate', self.cb_destroy)
        self.menu.append(quit)

        self.icon.connect('activate', self.cb_update)
        self.icon.connect('popup-menu', self.cb_popup_menu)
        self.icon.set_visible(True)

        self.update_status()
        GLib.timeout_add_seconds(interval or 15, self.update_status)

def on_tasks_file_changed(self, monitor, file, other_file, event_type):
        if event_type == Gio.FileMonitorEvent.CHANGES_DONE_HINT:
            self.check_reload_tasks()
        else:
            GLib.timeout_add_seconds(1, self.check_reload_tasks)

def _add_timeout(self):
        if self._id:
            GLib.source_remove(self._id)
        self._id = GLib.timeout_add_seconds(self.timeout, self._run_func)

def on_btNext_clicked(self, *a):
        rvController = self.builder.get_object("rvController")
        tvDevices = self.builder.get_object("tvDevices")
        stDialog = self.builder.get_object("stDialog")
        btBack = self.builder.get_object("btBack")
        btNext = self.builder.get_object("btNext")
        pages = stDialog.get_children()
        index = pages.index(stDialog.get_visible_child())
        if index == 0:
            model, iter = tvDevices.get_selection().get_selected()
            dev = evdev.InputDevice(model[iter][0])
            if (dev.info.vendor, dev.info.product) == (0x054c, 0x09cc):
                # Special case for PS4 controller
                cbDS4 = self.builder.get_object("cbDS4")
                imgDS4 = self.builder.get_object("imgDS4")
                imgDS4.set_from_file(os.path.join(
                        self.app.imagepath, "ds4-small.svg"))
                cbDS4.set_active(Config()['drivers']['ds4drv'])
                stDialog.set_visible_child(pages[3])
                btBack.set_sensitive(True)
                btNext.set_label("_Restart Emulation")
                return
            stDialog.set_visible_child(pages[1])
            self.load_buttons()
            self.refresh_controller_image()
            rvController.set_reveal_child(True)
            self.load_buttons()
            self.refresh_controller_image()
            btBack.set_sensitive(True)
        elif index == 1:
            # Disable Next button and determine which driver should be used
            btNext.set_sensitive(False)
            model, iter = tvDevices.get_selection().get_selected()
            dev = evdev.InputDevice(model[iter][0])
            self.prepare_registration(dev)
        elif index == 2:
            self.save_registration()
        elif index == 3:
            # Next pressed on DS4 info page
            # where it means 'Restart Emulation and close'
            self.app.dm.stop()
            GLib.timeout_add_seconds(1, self.app.dm.start)
            self.kill_tester()
            self.window.destroy()

def update_cpu_speeds(self):
        label = 'failed'
        try:
            jobslist = jobs.get_jobs(config)
            label = ''
            for item in self.instance_items:
                self.menu.remove(item)
            self.instance_items.clear()
            for job in jobslist:
                if label != '':
                     label += ' '
                if job['status'] in ['SUBMITTED']:
                    label += '(' + job['type'] + ')'
                    GLib.timeout_add_seconds(10, self.handler_poll_onetime)  # fast poll whilst wait for it to start
                else:
                    label += job['type']

                item = Gtk.MenuItem()
                item.set_label('ssh to %s' % job['image'])
                item.connect("activate", self.handler_instance_ssh)
                item.target_image = job['image']
                item.job_number = job['number']
                item.show()
                self.menu.insert(item, 0)
                self.instance_items.append(item)

                item = Gtk.MenuItem()
                item.set_label('kill %s' % job['image'])
                item.connect("activate", self.handler_instance_kill)
                item.target_image = job['image']
                item.job_number = job['number']
                item.show()
                self.menu.insert(item, 0)
                self.instance_items.append(item)

        except Exception as e:
            label = 'exception occurred'
            try:
                print(traceback.format_exc())
            except:
                print('exception in exception :-P')
        self.ind.set_label(label, "")