我们从Python开源项目中，提取了以下24个代码示例，用于说明如何使用gi.repository.GLib.PRIORITY_HIGH。
def copy_files(self, file_name, action): if action == 'make_backup': command = ['cp', '-R', self.winetricks_cache + '/' + file_name, self.winetricks_cache_backup] elif action == 'restore_backup': command = ['cp', '-R', self.winetricks_cache_backup + '/' + file_name, self.winetricks_cache] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'copy_files', priority=GLib.PRIORITY_HIGH)
def check_for_new_games(self): self.goglib_new_games_list = [] self.goglib_updated_games_list = [] command = ['lgogdownloader', '--exclude', '1,2,4,8,16,32','--list'] pid, stdin, stdout, stderr = GLib.spawn_async( command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True ) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch( GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'check_for_new_games', priority=GLib.PRIORITY_HIGH )
def run_goglib_setup_script(self, game_name): self.progressbar_goglib.set_text(_("Executing script...")) self.set_environ(game_name, self.goglib_download_dir, self.goglib_install_dir) command = [data_dir + '/scripts/goglib/' + game_name + '/setup'] goglib_name_to_pid_install_dict[game_name], stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'run_goglib_setup_script', priority=GLib.PRIORITY_HIGH)
def InitSignal(gui): def signal_action(signal): if signal.value is signal.SIGINT.value: print("\r", end="") logging.debug("Caught signal SIGINT(2)") gui.stop() def handler(*args): signal_action(args[0]) def idle_handler(*args): GLib.idle_add(signal_action, priority=GLib.PRIORITY_HIGH) def install_glib_handler(sig): GLib.unix_signal_add(GLib.PRIORITY_HIGH, sig, handler, sig) SIGS = [getattr(signal, s, None) for s in "SIGINT".split()] for sig in filter(None, SIGS): signal.signal(sig, idle_handler) GLib.idle_add(install_glib_handler, sig, priority=GLib.PRIORITY_HIGH)
def setup_signals(self, signals, handler): """ This is a workaround to signal.signal(signal, handler) which does not work with a GLib.MainLoop() for some reason. Thanks to: http://stackoverflow.com/a/26457317/5433146 args: signals (list of signal.SIG... signals): the signals to connect to handler (function): function to be executed on these signals """ def install_glib_handler(sig): # add a unix signal handler GLib.unix_signal_add( GLib.PRIORITY_HIGH, sig, # for the given signal handler, # on this signal, run this function sig # with this argument ) for sig in signals: # loop over all signals GLib.idle_add( # 'execute' install_glib_handler, sig, # add a handler for this signal priority = GLib.PRIORITY_HIGH ) # set the config
def setup_signals(self, signals, handler): """ This is a workaround to signal.signal(signal, handler) which does not work with a GLib.MainLoop() for some reason. Thanks to: http://stackoverflow.com/a/26457317/5433146 args: signals (list of signal.SIG... signals): the signals to connect to handler (function): function to be executed on these signals """ def install_glib_handler(sig): # add a unix signal handler GLib.unix_signal_add( GLib.PRIORITY_HIGH, sig, # for the given signal handler, # on this signal, run this function sig # with this argument ) for sig in signals: # loop over all signals GLib.idle_add( # 'execute' install_glib_handler, sig, # add a handler for this signal priority = GLib.PRIORITY_HIGH ) # build the gui
def attach_readfd(iochannel, context): # This does not work currently :-( # Also, things are mixed up a bit with iochannel vs. FDs apparently # as I am getting an int back in the handler # if isinstance(iochannel, int): # iochannel = GLib.IOChannel(iochannel) #source = GLib.io_create_watch(iochannel, GLib.IOCondition(GLib.IO_IN | GLib.IO_PRI)) #source.set_callback(_glib_signal_cb, context) # source.set_priority(GLib.PRIORITY_HIGH) # source.attach(context) GLib.io_add_watch(iochannel, GLib.IO_IN | GLib.IO_PRI, _glib_signal_cb, context) # Now, this is our magic exception hook. It is *magic* # it detects whether the exception happened in our signal handler. # The fun part is that the handler does not run properly. The mainloop # will actually try to run it again as soon as it can (if it manages). # We use that to clear the exception option again.
def setup_signals(self, signals, handler): """ This is a workaround to signal.signal(signal, handler) which does not work with a GLib.MainLoop() for some reason. Thanks to: http://stackoverflow.com/a/26457317/5433146 args: signals (list of signal.SIG... signals): the signals to connect to handler (function): function to be executed on these signals """ def install_glib_handler(sig): # add a unix signal handler GLib.unix_signal_add( GLib.PRIORITY_HIGH, sig, # for the given signal handler, # on this signal, run this function sig # with this argument ) for sig in signals: # loop over all signals GLib.idle_add( # 'execute' install_glib_handler, sig, # add a handler for this signal priority = GLib.PRIORITY_HIGH ) # get an object from the builder
def crop_file(self, file_name, crop): if crop == 1: region = '640:360:0:60' elif crop == 2: region = '640:400:0:40' command = ['ffmpeg', '-i', video_dir_bak + '/' + file_name, '-filter:v', 'crop=' + region, '-q', '1', video_dir + '/' + file_name] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'cropping_video', priority=GLib.PRIORITY_HIGH)
def crop_file_480(self, file_name, crop): if crop == 1: region = '640:360:0:60' elif crop == 2: region = '640:400:0:40' command = ['ffmpeg', '-i', video_dir_bak + '/' + file_name, '-filter:v', 'crop=' + region, '-vcodec', 'wmv2', '-q', '3', video_dir + '/' + file_name] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'cropping_video_480', priority=GLib.PRIORITY_HIGH)
def crop_file_600(self, file_name, crop): if crop == 1: region = '800:450:0:75' elif crop == 2: region = '800:500:0:50' command = ['ffmpeg', '-i', video_dir_bak + '/' + file_name, '-filter:v', 'crop=' + region, '-vcodec', 'wmv2', '-q', '5', video_dir + '/' + file_name] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'cropping_video_600', priority=GLib.PRIORITY_HIGH)
def goglib_download_game(self, game_name): if self.goglib_offline_mode: self.goglib_unpack_game(goglib_installation_queue[0]) else: if not self.queue_tab_exists(): self.append_queue_tab() self.progressbar_goglib.set_text(_("Downloading...")) if not os.path.exists(self.goglib_download_dir + '/' + game_name): os.makedirs(self.goglib_download_dir + '/' + game_name) self.preferred_language = self.lang_index[self.goglib_lang.lower()] if self.goglib_download_extras == False: command = ['lgogdownloader', '--download', '--ignore-dlc-count', '--platform', '4,1', \ '--language', self.preferred_language + ',1,', '--game', game_name + '$', \ '--directory=' + self.goglib_download_dir + '/', '--exclude', '2,4,16'] elif self.goglib_download_extras == True: command = ['lgogdownloader', '--download', '--ignore-dlc-count', '--platform', '4,1', \ '--language', self.preferred_language + ',1,', '--game', game_name + '$', \ '--directory=' + self.goglib_download_dir + '/', '--exclude', '4,16'] goglib_name_to_pid_download_dict[game_name], stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'goglib_download_game', priority=GLib.PRIORITY_HIGH)
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 idle_handler(*args): GLib.idle_add(quitApplication, 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: unix_signal_add(GLib.PRIORITY_HIGH, sig, handler, sig) else: print("Can't user Glib signal handler")
def main(): SIGS = [getattr(signal, s, None) for s in "SIGINT SIGTERM SIGHUP".split()] for sig in filter(None, SIGS): signal.signal(sig, idle_handler) GLib.idle_add(install_glib_handler, sig, priority=GLib.PRIORITY_HIGH) IEE = ImunesExperimentExporter() IEE.main()
def execute_command(self, command, process_name): pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, process_name, priority=GLib.PRIORITY_HIGH)
def update_goglib(self): self.window_update_message = Gtk.Window( title = _("Changes in library"), type = Gtk.WindowType.POPUP, window_position = Gtk.WindowPosition.CENTER_ALWAYS, resizable = False, icon = app_icon, ) self.box_update_message = Gtk.Box( orientation = Gtk.Orientation.HORIZONTAL ) self.label_update_message = Gtk.Label( label = _("Updating GOG library..."), margin_right = 10, margin_top = 20, margin_bottom = 20, ) self.spinner_update_message = Gtk.Spinner( active = True, visible = True, margin_left = 10, width_request = 48, height_request = 48 ) self.box_update_message.pack_start(self.spinner_update_message, True, True, 0) self.box_update_message.pack_start(self.label_update_message, True, True, 0) self.window_update_message.add(self.box_update_message) self.main_window.hide() if len(self.additional_windows_list) != 0: for window in self.additional_windows_list: window.hide() while Gtk.events_pending(): Gtk.main_iteration() self.window_update_message.show_all() command = ['lgogdownloader', '--exclude', '1,2,4,8,16,32','--list-details'] pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'update_goglib', priority=GLib.PRIORITY_HIGH)
def mylib_install_game(self, game_name): if not self.queue_tab_exists(): self.append_queue_tab() if not os.path.exists(self.mylib_install_dir): os.makedirs(self.mylib_install_dir) game_dir = self.mylib_install_dir + '/' + game_name if not os.path.exists(game_dir): os.makedirs(game_dir) self.mylib_tab_progressbar.set_text(_("Installing...")) self.set_environ(game_name, self.mylib_download_dir, self.mylib_install_dir) os.environ['KEEP_INSTALLERS'] = str(self.mylib_keep_installers) autosetup.autosetup('mylib', self.mylib_install_dir, game_name) settings_py_path = os.getenv('HOME') + '/.games_nebula/scripts/mylib/' + game_name + '/settings.py' if os.path.exists(settings_py_path): os.system('cp ' + settings_py_path + ' ' + game_dir) os.system('echo "Writing settings.sh"') settings_lines = ['#!/bin/bash\n', 'python2 "$INSTALL_DIR/' + game_name + '/settings.py"'] settings_file = open(game_dir + '/settings.sh', 'w') for line in settings_lines: settings_file.write(line) settings_file.close() os.system('chmod +x ' + game_dir + '/settings.sh') if self.own_prefix: config_file = open(game_dir + '/config.ini', 'w') config_file.write('[Settings]\n') config_file.write('own_prefix = True') config_file.close() command = [data_dir + '/scripts/mylib/' + game_name + '/setup'] mylib_name_to_pid_install_dict[game_name], stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'mylib_install_game', priority=GLib.PRIORITY_HIGH)
def cb_button_patch(self, button): if os.path.exists(game_dir + '/' + 'br2fsaaConfig.exe'): self.launch_fsaa_settings() else: patch_path = download_dir + '/_distr/bloodrayne_2/BR2_FSAA_Patch_1.666.rar' if not os.path.exists(patch_path): message_dialog = Gtk.MessageDialog( self.main_window, 0, Gtk.MessageType.ERROR, Gtk.ButtonsType.OK, _("Patch not found in download directory.") ) content_area = message_dialog.get_content_area() content_area.set_property('margin-left', 10) content_area.set_property('margin-right', 10) content_area.set_property('margin-top', 10) content_area.set_property('margin-bottom', 10) self.main_window.hide() message_dialog.run() message_dialog.destroy() self.main_window.show() else: self.button_patch.set_sensitive(False) while Gtk.events_pending(): Gtk.main_iteration_do(False) command = ['7z', 'e', '-aoa', '-o' + game_dir, patch_path] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'extracting', priority=GLib.PRIORITY_HIGH)
def cb_button_install_std(self, button): modpacks_path = download_dir + '/_distr/the_temple_of_elemental_evil/' + exe_co8_modpack_std if not os.path.exists(modpacks_path): message_dialog = Gtk.MessageDialog( self.co8_std_window, 0, Gtk.MessageType.ERROR, Gtk.ButtonsType.OK, _("Modpack not found in download directory") ) content_area = message_dialog.get_content_area() content_area.set_property('margin-left', 10) content_area.set_property('margin-right', 10) content_area.set_property('margin-top', 10) content_area.set_property('margin-bottom', 10) self.co8_std_window.hide() message_dialog.run() message_dialog.destroy() self.co8_std_window.show() else: self.box_std.set_visible(False) self.progressbar_std.set_visible(True) command = ['innoextract', modpacks_path, '-d', game_dir + '/tmp'] self.pid_std, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'extracting std', priority=GLib.PRIORITY_HIGH)
def cb_button_install_nc(self, button): modpacks_path = download_dir + '/_distr/the_temple_of_elemental_evil/' + exe_co8_modpack_nc if not os.path.exists(modpacks_path): message_dialog = Gtk.MessageDialog( self.co8_nc_window, 0, Gtk.MessageType.ERROR, Gtk.ButtonsType.OK, _("Modpack not found in download directory") ) content_area = message_dialog.get_content_area() content_area.set_property('margin-left', 10) content_area.set_property('margin-right', 10) content_area.set_property('margin-top', 10) content_area.set_property('margin-bottom', 10) self.co8_nc_window.hide() message_dialog.run() message_dialog.destroy() self.co8_nc_window.show() else: self.box_nc.set_visible(False) self.progressbar_nc.set_visible(True) command = ['innoextract', modpacks_path, '-d', game_dir + '/tmp'] self.pid_nc, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'extracting nc', priority=GLib.PRIORITY_HIGH)
def cb_button_patch(self, button): patch_path = download_dir + '/_distr/pathologic/Pathologic_Widescreen_Addon.exe' if not os.path.exists(patch_path): message_dialog = Gtk.MessageDialog( self.main_window, 0, Gtk.MessageType.ERROR, Gtk.ButtonsType.OK, _("Addon not found in download directory.") ) content_area = message_dialog.get_content_area() content_area.set_property('margin-left', 10) content_area.set_property('margin-right', 10) content_area.set_property('margin-top', 10) content_area.set_property('margin-bottom', 10) self.main_window.hide() message_dialog.run() message_dialog.destroy() self.main_window.show() else: self.button_config.set_sensitive(False) self.button_patch.set_visible(False) self.progressbar.set_visible(True) while Gtk.events_pending(): Gtk.main_iteration_do(False) command = ['innoextract', patch_path, '-d', game_dir + '/tmp'] self.pid, stdin, stdout, stderr = GLib.spawn_async(command, flags=GLib.SpawnFlags.SEARCH_PATH|GLib.SpawnFlags.DO_NOT_REAP_CHILD, standard_output=True, standard_error=True) io = GLib.IOChannel(stdout) self.source_id_out = io.add_watch(GLib.IO_IN|GLib.IO_HUP, self.watch_process, 'extracting', priority=GLib.PRIORITY_HIGH)
