我们从Python开源项目中，提取了以下28个代码示例，用于说明如何使用gi.repository.GLib.IO_IN。
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 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 color_update(self, bytedata): """Launch new calculation process with given image bytedata""" if bytedata is None: if self.default is None: if not self.config["image"]["default"].endswith(".svg"): # fix this file_ = self.config["image"]["default"] self.handler_unblock(self.catcher) else: return else: self.emit("ac-update", self.default) return else: file_ = io.BytesIO(bytedata) if self.process is None or not self.process.is_alive(): if self.watcher is None: self.watcher = GLib.io_add_watch(self.pc, GLib.IO_IN | GLib.IO_HUP, self.color_setup) self.process = multiprocessing.Process( target=self.calculate, args=(file_, self.config["autocolor"], self.cc) ) self.process.start() else: logger.error("Autocolor threading error: previus process still running, refusing to start new one")
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 inputhook_gtk3(): GLib.io_add_watch(sys.stdin, GLib.IO_IN, _main_quit) Gtk.main() return 0
def inputhook(context): GLib.io_add_watch(context.fileno(), GLib.IO_IN, _main_quit) Gtk.main()
def connect(self, connectionInfo, elementListener, onConnected): super(GLibUnixTransport, self).connect( connectionInfo, elementListener, onConnected) fd = self._socket.fileno() io_channel = GLib.IOChannel.unix_new(fd) self._watch_id = GLib.io_add_watch( io_channel, GLib.PRIORITY_DEFAULT, GLib.IO_IN, self._socket_ready)
def color_setup(self, conn, flag): """Read data from resent calculation and transform it to rgba color""" if flag == GLib.IO_IN: color_values = conn.recv() rgba = Gdk.RGBA(*color_values, self.config["color"]["autofg"].alpha) self.emit("ac-update", rgba) return True else: logger.error("Autocolor multiprocessing error: connection was unexpectedy terminated") self.watcher = None
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 start_process(self): """ Start the ffmpeg subprocess """ if self.process and self.process.poll() is None: raise RuntimeError('Refusing to start process when already running') cmd = self.get_process_command() self.process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, ) # Make the pipes non-blocking flags = fcntl.fcntl(self.process.stdout, fcntl.F_GETFL) fcntl.fcntl(self.process.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK) flags = fcntl.fcntl(self.process.stderr, fcntl.F_GETFL) fcntl.fcntl(self.process.stderr, fcntl.F_SETFL, flags | os.O_NONBLOCK) # Clear any output from previous incarnations self.clear_process_stdout() self.clear_process_stderr() # Update the UI when there's output from the process def output_callback(fd, condition, pipe, func): """ Read from pipe, and pass to the given func """ output = pipe.read() if not output: return False func(output.decode('utf-8')) return condition != GLib.IO_HUP stdout_read_cb = GLib.io_add_watch( self.process.stdout, GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP, output_callback, self.process.stdout, self.append_process_stdout, ) GLib.io_add_watch( self.process.stderr, GLib.PRIORITY_DEFAULT, GLib.IO_IN | GLib.IO_HUP, output_callback, self.process.stderr, self.append_process_stderr, ) # Update the UI self.show_process_state()
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)
