def on_profile_changed(self, c, profile):
""" Called when controller profile is changed from daemon """
if not self.set_profile(profile, True):
if self._first_time:
def later():
# Cannot be executed right away, as profile-changed is
# emitted before DaemonManager finishes initiaalisation
self.emit('unknown-profile', profile)
GLib.idle_add(later)
self._first_time = False
python类idle_add()的实例源码
def setup_statusicon(self):
menu = self.builder.get_object("mnuTray")
self.statusicon = get_status_icon(self.imagepath, menu)
self.statusicon.connect('clicked', self.on_statusicon_clicked)
if not self.statusicon.is_clickable():
self.builder.get_object("mnuShowWindowTray").set_visible(True)
GLib.idle_add(self.statusicon.set, "scc-%s" % (self.status,), _("SC Controller"))
def set_daemon_status(self, status, daemon_runs):
""" Updates image that shows daemon status and menu shown when image is clicked """
log.debug("daemon status: %s", status)
icon = os.path.join(self.imagepath, "scc-%s.svg" % (status,))
imgDaemonStatus = self.builder.get_object("imgDaemonStatus")
btDaemon = self.builder.get_object("btDaemon")
mnuEmulationEnabled = self.builder.get_object("mnuEmulationEnabled")
mnuEmulationEnabledTray = self.builder.get_object("mnuEmulationEnabledTray")
imgDaemonStatus.set_from_file(icon)
mnuEmulationEnabled.set_sensitive(True)
mnuEmulationEnabledTray.set_sensitive(True)
self.window.set_icon_from_file(icon)
self.status = status
if self.statusicon:
GLib.idle_add(self.statusicon.set, "scc-%s" % (self.status,), _("SC Controller"))
self.recursing = True
if status == "alive":
btDaemon.set_tooltip_text(_("Emulation is active"))
elif status == "error":
btDaemon.set_tooltip_text(_("Error enabling emulation"))
elif status == "dead":
btDaemon.set_tooltip_text(_("Emulation is inactive"))
else:
btDaemon.set_tooltip_text(_("Checking emulation status..."))
mnuEmulationEnabled.set_active(daemon_runs)
mnuEmulationEnabledTray.set_active(daemon_runs)
self.recursing = False
def on_sbArea_focus_out_event(self, button, *a):
GLib.idle_add(self.on_sbArea_output, button)
def on_scc_import_confirmed(self, *a):
files = self.builder.get_object("lstImportPackage")
new_profile_names = {}
new_menu_names = {}
for enabled, name, importas, trash, obj in files:
if enabled != 0:
if isinstance(obj.obj, Profile):
new_profile_names[name] = importas
elif isinstance(obj.obj, MenuData):
new_menu_names["%s.menu" % (name,)] = "%s.menu" % (importas,)
def apply_replacements(obj):
for a in obj.get_all_actions():
if isinstance(a, ChangeProfileAction):
if a.profile in new_profile_names:
a.profile = new_profile_names[a.profile]
a.parameters = tuple([ a.profile ] + list(a.parameters[1:]))
elif isinstance(a, MenuAction):
if a.menu_id in new_menu_names:
a.menu_id = new_menu_names[a.menu_id]
a.parameters = tuple([ a.menu_id ] + list(a.parameters[1:]))
for enabled, trash, importas, trash, obj in files:
if enabled != 0:
# TODO: update references
if isinstance(obj.obj, Profile):
apply_replacements(obj.obj)
obj.obj.save(os.path.join(get_profiles_path(), "%s.sccprofile" % (importas,)))
elif isinstance(obj.obj, MenuData):
apply_replacements(obj.obj)
jstr = Encoder(sort_keys=True, indent=4).encode(obj.obj)
filename = os.path.join(get_menus_path(), "%s.menu" % (importas,))
open(filename, "w").write(jstr)
trash, trash, importas, trash, obj = files[0] # 1st is always profile that's being imported
self.app.new_profile(obj.obj, importas)
GLib.idle_add(self.window.destroy)
def _parse_profile_list(self, i, filename):
"""
Parses sharedconfig.vdf and loads game and profile IDs. That is later
decoded into name of game and profile name.
Called from _load_profiles, in thread. Exceptions are catched and logged
from there.
Calls GLib.idle_add to send loaded data into UI.
"""
data = parse_vdf(open(filename, "r"))
# Sanity check
if "userroamingconfigstore" not in data: return
if "controller_config" not in data["userroamingconfigstore"]: return
# Grab config
cc = data["userroamingconfigstore"]["controller_config"]
# Go through all games
listitems = []
for gameid in cc:
if "selected" in cc[gameid] and cc[gameid]["selected"].startswith("workshop"):
profile_id = cc[gameid]["selected"].split("/")[-1]
listitems.append(( i, gameid, profile_id, None ))
i += 1
if len(listitems) > 10:
GLib.idle_add(self.fill_list, listitems)
listitems = []
GLib.idle_add(self.fill_list, listitems)
return i
def _load_game_names(self):
"""
Loads names for game ids in q_games.
This is done in thread (not in same thread as _load_profiles),
because it involves searching for apropriate file and parsing it
entirely.
Calls GLib.idle_add to send loaded data into UI.
"""
sa_path = self._find_steamapps()
while True:
self._s_games.acquire(True) # Wait until something is added to the queue
try:
index, gameid = self._q_games.popleft()
except IndexError:
break
if gameid.isdigit():
name = _("Unknown App ID %s") % (gameid)
filename = os.path.join(sa_path, "appmanifest_%s.acf" % (gameid,))
self._lock.acquire()
if os.path.exists(filename):
try:
data = parse_vdf(open(filename, "r"))
name = data['appstate']['name']
except Exception, e:
log.error("Failed to load app manifest for '%s'", gameid)
log.exception(e)
else:
log.warning("Skiping non-existing app manifest '%s'", filename)
self._lock.release()
else:
name = gameid
GLib.idle_add(self._set_game_name, index, name)
def _load_profile_names(self):
"""
Loads names for profiles ids in q_profiles.
This is same as _load_game_names, but for profiles.
"""
content_path = os.path.join(self._find_steamapps(), "workshop/content")
if not os.path.exists(content_path):
log.warning("Cannot find '%s'; Cannot import anything without it", content_path)
return
while True:
self._s_profiles.acquire(True) # Wait until something is added to the queue
try:
index, gameid, profile_id = self._q_profiles.popleft()
except IndexError:
break
self._lock.acquire()
for user in os.listdir(content_path):
filename = os.path.join(content_path, user, profile_id, "controller_configuration.vdf")
if not os.path.exists(filename):
# If there is no 'controller_configuration.vdf', try finding *_legacy.bin
filename = self._find_legacy_bin(os.path.join(content_path, user, profile_id))
if not filename or not os.path.exists(filename):
# If not even *_legacy.bin is found, skip to next user
continue
log.info("Reading '%s'", filename)
try:
data = parse_vdf(open(filename, "r"))
name = data['controller_mappings']['title']
GLib.idle_add(self._set_profile_name, index, name, filename)
break
except Exception, e:
log.error("Failed to read profile name from '%s'", filename)
log.exception(e)
else:
log.warning("Profile %s for game %s not found.", profile_id, gameid)
name = _("(not found)")
GLib.idle_add(self._set_profile_name, index, name, None)
self._lock.release()
def on_daemon_reconfigured(self, *a):
# config is reloaded in main window 'reconfigured' handler.
# Using GLib.idle_add here ensures that main window hanlder will run
# *before* self.load_conditions
GLib.idle_add(self.load_settings)
def SignalHandler(app):
def signal_action(signal):
if signal is 1: print("Caught signal SIGHUP(1)")
elif signal is 2: print("Caught signal SIGINT(2)")
elif signal is 15: print("Caught signal SIGTERM(15)")
app.exit_gracefully()
def handler(*args):
# Activate GLib signal handler
signal_action(args[0])
def idle_handler(*args):
# Activate python signal handler
GLib.idle_add(signal_action, priority=GLib.PRIORITY_HIGH)
def install_glib_handler(sig):
unix_signal_add = None
if hasattr(GLib, "unix_signal_add"):
unix_signal_add = GLib.unix_signal_add
elif hasattr(GLib, "unix_signal_add_full"):
unix_signal_add = GLib.unix_signal_add_full
if unix_signal_add:
# Register GLib signal handler
unix_signal_add(GLib.PRIORITY_HIGH, sig, handler, sig)
else:
print("WARNING: Can't install GLib signal handler, too old gi.")
SIGS = [getattr(signal, s, None) for s in "SIGINT SIGTERM SIGHUP".split()]
for sig in filter(None, SIGS):
# Register Python signal handler
signal.signal(sig, idle_handler)
GLib.idle_add(install_glib_handler, sig, priority=GLib.PRIORITY_HIGH)
def emit(self, *args):
GLib.idle_add(GObject.GObject.emit, self, *args)
def emit(self, *args):
GLib.idle_add(GObject.GObject.emit, self, *args)
def read_stdout(self):
while True:
try:
e = ("<-" + self.side + "(" +
self.get_running_engine().strip() + "):")
line= ""
# python2 line = unicode(self.p.stdout.readline(), errors ='ignore')
# or: 'your iso 8859-15 text'.decode('iso8859-15')
# python3 (doesn't work) lineb = self.p.stdout.readline().encode("utf-8", "ignore")
#print(lineb)
#line = str(lineb)
#print(line, "line")
line = self.p.stdout.readline()
if line == "":
if gv.verbose:
print(e + "eof reached")
if gv.verbose:
print(e + "stderr:", self.p.stderr.read())
break
#line = line[2:-3]
#print(line)
line = line.strip()
if gv.verbose or gv.verbose_uci:
print(e + line)
GObject.idle_add(self.engine_debug.add_to_log, e+line)
if line.startswith("info"):
GObject.idle_add(
self.engine_output.add_to_log, self.side,
self.get_running_engine().strip(), line)
self.op.append(line)
except Exception as e:
# line = e + "error"
print("subprocess error in uci.py read_stdout:", e, "at:", line)
def build_board(self):
GObject.idle_add(self.update)
# convert jcchess co-ordinates for square into
# standard notation (e.g. (1, 0) -> b1)
def set_square_as_unoccupied(self, x, y):
self.dnd = (x, y)
# redraw square
GLib.idle_add(gv.gui.get_event_box(x, y).queue_draw)
# called from gui.py when editing the board position to set the piece
# on a square.
def set_piece_at_square(self, x, y, piece, colour):
self.chessboard.set_piece_at(chess.square(x, y), chess.Piece(piece, colour))
GLib.idle_add(gv.gui.get_event_box(x, y).queue_draw)
# called from gui.py to remove piece during promotion dialog
def remove_piece_at_square(self, x, y):
piece=self.chessboard.remove_piece_at(chess.square(x, y))
GLib.idle_add(gv.gui.get_event_box(x, y).queue_draw)
return piece
# called when user does a "clear board" in board edit
def set_toolbar_time_control(self, txt, side_to_move):
if side_to_move is None:
return
GLib.idle_add(self.tc_lbl[side_to_move].set_text, txt)
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
def update_color(self, bg_color: str, is_dark: bool):
"""Updates the widget's background color and theme.
:bg_color: string representing the color, or 'None'.
:is_dark: whether the background is dark, see `:h background`.
"""
self.get_settings().props.gtk_application_prefer_dark_theme = is_dark
if bg_color != 'None':
rgba = Gdk.RGBA()
rgba.parse(bg_color)
self.set_color_background(rgba)
else:
GLib.idle_add(self.reset_color)
