python类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)