def _threaded_perform_search(self):
self._perform_search_complete = False
# generate a name and ensure we never have two threads
# with the same name
names = [thread.name for thread in threading.enumerate()]
for i in range(threading.active_count() + 1, 0, -1):
thread_name = 'ThreadedQuery-%s' % i
if not thread_name in names:
break
# create and start it
t = threading.Thread(
target=self._blocking_perform_search, name=thread_name)
t.start()
# don't block the UI while the thread is running
context = GObject.main_context_default()
while not self._perform_search_complete:
time.sleep(0.02) # 50 fps
while context.pending():
context.iteration()
t.join()
# call the query-complete callback
self.emit("query-complete")
python类GObject()的实例源码
def wait_for_apt_cache_ready(f):
""" decorator that ensures that self.cache is ready using a
gtk idle_add - needs a cache as argument
"""
def wrapper(*args, **kwargs):
self = args[0]
# check if the cache is ready and
window = None
if hasattr(self, "app_view"):
window = self.app_view.get_window()
if not self.cache.ready:
if window:
window.set_cursor(self.busy_cursor)
GObject.timeout_add(500, lambda: wrapper(*args, **kwargs))
return False
# cache ready now
if window:
window.set_cursor(None)
f(*args, **kwargs)
return False
return wrapper
def _check_url_reachable_and_then_download_cb(self, f, result,
user_data=None):
self.LOG.debug("_check_url_reachable_and_then_download_cb: %s" % f)
try:
info = f.query_info_finish(result)
etag = info.get_etag()
self.emit('file-url-reachable', True)
self.LOG.debug("file reachable %s %s %s" % (self.url,
info,
etag))
# url is reachable, now download the file
f.load_contents_async(
self._cancellable, self._file_download_complete_cb, None)
except GObject.GError as e:
self.LOG.debug("file *not* reachable %s" % self.url)
self.emit('file-url-reachable', False)
self.emit('error', GObject.GError, e)
del f
def __init__(self):
GObject.GObject.__init__(self)
InstallBackend.__init__(self)
# transaction details for setting as meta
self.new_pkgname, self.new_appname, self.new_iconname = '', '', ''
# this is public exposed
self.pending_transactions = {}
self.client = packagekit.Client()
self.pkginfo = get_pkg_info()
self.pkginfo.open()
self._transactions_watcher = PackagekitTransactionsWatcher()
self._transactions_watcher.connect('lowlevel-transactions-changed',
self._on_lowlevel_transactions_changed)
def __init__(self, oneconfviewpickler):
'''Controller of the installed pane'''
LOG.debug("OneConf Handler init")
super(OneConfHandler, self).__init__()
# OneConf stuff
self.oneconf = DbusConnect()
self.oneconf.hosts_dbus_object.connect_to_signal('hostlist_changed',
self.refresh_hosts)
self.oneconf.hosts_dbus_object.connect_to_signal('packagelist_changed',
self._on_store_packagelist_changed)
self.oneconf.hosts_dbus_object.connect_to_signal('latestsync_changed',
self.on_new_latest_oneconf_sync_timestamp)
self.already_registered_hostids = []
self.is_current_registered = False
self.oneconfviewpickler = oneconfviewpickler
# refresh host list
self._refreshing_hosts = False
GObject.timeout_add_seconds(MIN_TIME_WITHOUT_ACTIVITY,
self.get_latest_oneconf_sync)
GObject.idle_add(self.refresh_hosts)
def emit(self, *args):
GLib.idle_add(GObject.GObject.emit, self, *args)
def __init__(self):
GObject.GObject.__init__(self)
TimerManager.__init__(self)
self._direction = 0
def __init__(self):
GObject.GObject.__init__(self)
self.alive = None
self.connection = None
self.connecting = False
self.buffer = ""
self._connect()
self._requests = []
self._controllers = [] # Ordered as daemon says
self._controller_by_id = {} # Source of memory leak
def __init__(self, daemon_manager, controller_id):
GObject.GObject.__init__(self)
self._dm = daemon_manager
self._controller_id = controller_id
self._config_file = None
self._profile = None
self._type = None
self._connected = False
def _add_gesture(self, gstr, action, select=False):
lstGestures = self.builder.get_object("lstGestures")
o = GObject.GObject()
o.gstr = gstr
o.action = action
iter = lstGestures.append( (
GestureComponent.nice_gstr(gstr),
action.describe(Action.AC_MENU),
o
) )
if select:
tvGestures = self.builder.get_object("tvGestures")
tvGestures.get_selection().select_iter(iter)
self.on_tvGestures_cursor_changed()
self.on_btEditAction_clicked()
def __init__(self, icon_path, popupmenu, force=False):
GObject.GObject.__init__(self)
self.__icon_path = os.path.normpath(os.path.abspath(icon_path))
self.__popupmenu = popupmenu
self.__active = True
self.__visible = False
self.__hidden = False
self.__icon = "si-syncthing-unknown"
self.__text = ""
self.__force = force
def __init__(self, driver, device_id):
GObject.GObject.__init__(self)
self.buffer = b""
self.buttons = []
self.axes = []
self.subprocess = None
self.driver = driver
self.device_id = device_id
self.errorred = False # To prevent sending 'error' signal multiple times
def import_scc_tar(self, filename):
"""
Imports packaged profiles.
Checks for shell() actions everywhere and ask user to
enter main name, check generated ones and optionaly change
them as he wish.
"""
files = self.builder.get_object("lstImportPackage")
try:
# Open tar
tar = tarfile.open(filename, "r:gz")
files.clear()
# Grab 1st profile
name = tar.extractfile(Export.PN_NAME).read()
main_profile = "%s.sccprofile" % name
parser = GuiActionParser()
o = GObject.GObject()
o.obj = Profile(parser).load_fileobj(tar.extractfile(main_profile))
files.append(( 2, name, name, _("(profile)"), o ))
for x in tar:
name = ".".join(x.name.split(".")[0:-1])
if x.name.endswith(".sccprofile") and x.name != main_profile:
o = GObject.GObject()
o.obj = Profile(parser).load_fileobj(tar.extractfile(x))
files.append(( True, name, name, _("(profile)"), o ))
elif x.name.endswith(".menu"):
o = GObject.GObject()
o.obj = MenuData.from_fileobj(tar.extractfile(x), parser)
files.append(( True, name, name, _("(menu)"), o ))
except Exception, e:
# Either entire tar or some profile cannot be parsed.
# Display error message and let user to quit
# Error message reuses same page as above.
log.error(e)
self.error(str(e))
return
self.check_shell_commands()
def on_btAdd_clicked(self, *a):
""" Handler for "Add Item" button """
tvItems = self.builder.get_object("tvItems")
model = tvItems.get_model()
o = GObject.GObject()
o.condition = Condition()
o.action = NoAction()
iter = model.append((o, o.condition.describe(), "None"))
tvItems.get_selection().select_iter(iter)
self.on_tvItems_cursor_changed()
self.btEdit_clicked_cb()
def __init__(self,cua,folder):
threading.Thread.__init__(self)
GObject.GObject.__init__(self)
self.setDaemon(True)
self.cua = cua
self.folder = folder
def __init__(self,cua,folder):
threading.Thread.__init__(self)
GObject.GObject.__init__(self)
self.setDaemon(True)
self.cua = cua
self.folder = folder
def __init__(self):
GObject.GObject.__init__(self)
self.pw = 0
# self.player = Gst.ElementFactory.make("playbin", "player")
# self.player.connect("about-to-finish", self.on_player_finished)
# bus = self.player.get_bus()
# bus.connect("message", self.on_player_message)
self.icon = comun.ICON
self.active_icon = None
self.about_dialog = None
self.active = False
self.animate = False
self.frame = 0
self.pomodoros = 0
self.player = Player()
self.notification = Notify.Notification.new('', '', None)
self.read_preferences()
#
self.indicator = appindicator.Indicator.new('Pomodoro-Indicator',
self.active_icon,
appindicator.
IndicatorCategory.
HARDWARE)
self.indicator.set_status(appindicator.IndicatorStatus.ACTIVE)
self.indicator.connect('scroll-event', self.on_scroll)
menu = self.get_menu()
self.indicator.set_menu(menu)
self.connect('session_end', self.on_session_end)
self.connect('break_end', self.on_break_end)
def emit(self, *args):
GLib.idle_add(GObject.GObject.emit, self, *args)
def main():
if dbus.SessionBus().request_name(
'es.atareao.PomodoroIndicator') !=\
dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER:
print("application already running")
exit(0)
GObject.threads_init()
Gst.init(None)
Gst.init_check(None)
Notify.init('pomodoro-indicator')
Pomodoro_Indicator()
Gtk.main()
def __init__(self):
GObject.GObject.__init__(self)
Gst.init_check(None)
self.status = Status.STOPPED
self.player = None
self.speed = 1.0
self.volume = 1.0
self.removesilence = False
self.equalizer = {'band0': 0, 'band1': 0, 'band2': 0, 'band3': 0,
'band4': 0, 'band5': 0, 'band6': 0, 'band7': 0,
'band8': 0, 'band9': 0}
self.lastpos = 0
