def run_background(self, interval=1):
"""
Convert minute interval to seconds
and updated services with new timeout.
"""
timeout = interval * 60 * 1000
self.services['timeout'] = GObject.timeout_add(timeout, self.update_reddit_data)
python类timeout_add()的实例源码
def _update_hr_msrmt_simulation(self):
print('Update HR Measurement Simulation')
if not self.notifying:
return
GObject.timeout_add(1000, self.hr_msrmt_cb)
def __init__(self, bus, index, service):
Characteristic.__init__(
self, bus, index,
self.BATTERY_LVL_UUID,
['read', 'notify'],
service)
self.notifying = False
self.battery_lvl = 100
GObject.timeout_add(5000, self.drain_battery)
def scan_devices( duration = 10 ):
"""
This causes the bluetooth system to scan for any broadcasting devices. Once found, the devices get added to a
dbus backed database specific for bluetooth for retrieval later.
Args:
duration (int): the amount of time that the scan should run for in seconds.
"""
adapter = DeviceManager().find_adapter()
adapter.StartDiscovery()
mainloop = GObject.MainLoop()
GObject.timeout_add(duration * 1000, quit, mainloop)
mainloop.run()
def main():
signal.signal(signal.SIGINT, signal.SIG_DFL)
indicator = appindicator.Indicator.new(APPINDICATOR_ID, \
os.path.abspath('/usr/share/icons/gnome/24x24/emotes/face-smile-big.png'), \
appindicator.IndicatorCategory.SYSTEM_SERVICES)
indicator.set_status(appindicator.IndicatorStatus.ACTIVE)
menu = gtk.Menu()
indicator.set_menu(build_menu(menu))
# GObject.timeout_add(1000, timespent, indicator, menu)
gtk.main()
def set_play(self, pos, time):
if self._timer is not None:
GObject.source_remove(self._timer)
self._timer = None
tracks = self._current_album.get_tracks()
for index in range(0, pos):
time = time + tracks[index].get_length()
self._songs_scale.set_value(time+1)
self._timer = GObject.timeout_add(1000, self._playing)
def maybe_auto_new_question(self):
if self.get_bool('new_question_automatically'):
if self.m_timeout_handle is None:
def remove_timeout(self=self):
self.m_timeout_handle = None
self.g_view.new_question()
self.m_timeout_handle = GObject.timeout_add(int(self.get_float('seconds_before_new_question')*1000), remove_timeout)
def on_start_practise(self):
soundcard.solfege_c_midi.dsp_open_record()
#self.__idle_tag = GObject.idle_add(self.update_view)
self.__idle_tag = GObject.timeout_add(300, self.update_view)
def flash(self, s):
self.clear()
l = Gtk.Label(label=s)
l.set_name("Feedback")
l.set_alignment(0.0, 0.5)
l.show()
self.g_box.pack_start(l, True, True, 0)
self.g_box.set_size_request(
max(l.size_request().width + gu.PAD * 2, self.g_box.size_request().width),
max(l.size_request().height + gu.PAD * 2, self.g_box.size_request().height))
self.__timeout = GObject.timeout_add(2000, self.unflash)
def flash(self, s):
self.clear()
l = Gtk.Label(label=s)
l.set_name("Feedback")
l.set_alignment(0.0, 0.5)
l.show()
self.g_box.pack_start(l, True, True, 0)
self.g_box.set_size_request(
max(l.size_request().width + gu.PAD * 2, self.g_box.size_request().width),
max(l.size_request().height + gu.PAD * 2, self.g_box.size_request().height))
self.__timeout = GObject.timeout_add(2000, self.unflash)
def delayed_flash(self, milliseconds, msg):
GObject.timeout_add(milliseconds, lambda: self.flash(msg))
def on_up_press(self, eb, ev):
if self.m_timeout:
GObject.source_remove(self.m_timeout)
self.m_timeout = None
if ev.type == Gdk.EventType.BUTTON_PRESS:
if self.m_value < 127:
self.up()
if self.m_value < 127:
self.m_timeout = GObject.timeout_add(DELAY1, self.on_up_timeout)
def on_up_timeout(self, *v):
if self.m_value < 127:
self.up()
self.m_timeout = GObject.timeout_add(DELAY2, self.on_up_timeout)
def on_down_press(self, eb, ev):
if self.m_timeout:
GObject.source_remove(self.m_timeout)
self.m_timeout = None
if ev.type == Gdk.EventType.BUTTON_PRESS:
if self.m_value > 0:
self.down()
if self.m_value > 0:
self.m_timeout = GObject.timeout_add(DELAY1, self.on_down_timeout)
def on_down_timeout(self, *v):
if self.m_value > 0:
self.down()
self.m_timeout = GObject.timeout_add(DELAY2, self.on_down_timeout)
def do_real_clean_task(self):
if len(self.clean_tasks) != 0:
plugin, cruft_dict = self.clean_tasks.pop(0)
plugin.set_property('clean_finished', False)
for row in self.janitor_model:
for child_row in row.iterchildren():
if child_row[self.JANITOR_PLUGIN] == plugin:
plugin_iter = child_row.iter
log.debug("Call %s to clean cruft" % plugin)
self._object_clean_handler = plugin.connect('object_cleaned',
self.on_plugin_object_cleaned,
(plugin_iter, cruft_dict))
self._all_clean_handler = plugin.connect('all_cleaned', self.on_plugin_cleaned, plugin_iter)
self._error_handler = plugin.connect('clean_error', self.on_clean_error, plugin_iter)
self.janitor_view.scroll_to_cell(self.janitor_model.get_path(plugin_iter))
t = threading.Thread(target=plugin.clean_cruft,
kwargs={'cruft_list': cruft_dict.keys(),
'parent': self.get_toplevel()})
for row in self.result_model:
if row[self.RESULT_PLUGIN] == plugin:
self.result_view.get_selection().select_path(row.path)
self.result_view.scroll_to_cell(row.path)
row[self.RESULT_DISPLAY] = '<b>%s</b>' % _('Cleaning cruft for "%s"...') % plugin.get_title()
self.result_view.expand_row(self.result_model.get_path(row.iter), True)
self.janitor_model[plugin_iter][self.JANITOR_SPINNER_ACTIVE] = True
self.janitor_model[plugin_iter][self.JANITOR_SPINNER_PULSE] = 0
GObject.timeout_add(50, self._on_clean_spinner_timeout, plugin_iter, t)
t.start()
else:
self.on_scan_button_clicked()
self.unset_busy()
def askInstallPackage(self, package, summary, description, homepage):
# populate the dialog
dia = self.dia
dia_xml = self.dia_xml
header = _("Install additional software?")
body = _("Do you want to install package '%s'?") % package
dia.set_title('')
header_label = dia_xml.get_object('header_label')
header_label.set_markup("<b><big>%s</big></b>" % header)
body_label = dia_xml.get_object('body_label')
body_label.set_label(body)
description_text_view = dia_xml.get_object('description_text_view')
tbuf = Gtk.TextBuffer()
desc = "%s\n\n%s" % (summary, Helpers.format_description(description))
tbuf.set_text(desc)
description_text_view.set_buffer(tbuf)
dia.set_icon(Gtk.IconTheme.get_default().load_icon('package-x-generic', 16, False))
# check if another package manager is already running
# FIXME: just checking for the existance of the file is
# not sufficient, it need to be tested if it can
# be locked via apt_pkg.get_lock()
# - but that needs to run as root
# - a dbus helper might be the best answer here
#args = (update_button_status, dia_xml.get_object("yes_button"),
# dia_xml.get_object("infolabel"))
#args[0](*args[1:])
#timer_id = GObject.timeout_add(750, *args )
# show the dialog
res = dia.run()
#GObject.source_remove(timer_id)
if res != Gtk.ResponseType.YES:
dia.hide()
return False
return True
# progress etc
def _update_temp_value(self):
if not self.props[constants.GATT_CHRC_IFACE]['Notifying']:
return
print('Starting timer event')
GObject.timeout_add(500, self.temperature_cb)
def _update(self, widget, data=None):
# This loop makes sure that we only call _find every 500 ms .
if self.timer_id:
# We destroy the last event source and create another one.
GObject.source_remove(self.timer_id)
self.timer_id = GObject.timeout_add(500, self._refresh, widget, data)
def _validate_cb(self, widget):
# We create a timeout object to avoid fast lookups on disk when typing.
self.timer_id = GObject.timeout_add(500, self._validate, widget.get_child())
