我们从Python开源项目中，提取了以下21个代码示例，用于说明如何使用psutil.Error()。
def test_children_duplicates(self): # find the process which has the highest number of children table = collections.defaultdict(int) for p in psutil.process_iter(): try: table[p.ppid()] += 1 except psutil.Error: pass # this is the one, now let's make sure there are no duplicates pid = sorted(table.items(), key=lambda x: x[1])[-1][0] p = psutil.Process(pid) try: c = p.children(recursive=True) except psutil.AccessDenied: # windows pass else: self.assertEqual(len(c), len(set(c)))
def get_steam_path(): """Get the path for Steam from the Steam process. If that fails, it uses the registry on Windows. Returns: str: The path to Steam. If the path could not be found, the current directory is returned instead (os.curdir) """ if psutil: for pid in psutil.process_iter(): try: if pid.name().lower() == 'steam.exe' or pid.name().lower() == 'steam': return os.path.dirname(pid.exe()) except psutil.Error: logger.exception("Could not get Steam path from its process.") if winreg: try: reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r'Software\Valve\Steam') return os.path.normpath(winreg.QueryValueEx(reg_key, r'SteamPath')[0]) except WindowsError: logger.exception("Could not query registry for Steam path") return os.curdir
def pidof(pgname): pids = [] for proc in psutil.process_iter(): # search for matches in the process name and cmdline try: name = proc.name() except psutil.Error: pass else: if name == pgname: pids.append(str(proc.pid)) continue try: cmdline = proc.cmdline() except psutil.Error: pass else: if cmdline and cmdline[0] == pgname: pids.append(str(proc.pid)) return pids
def main(): templ = "%-5s %-30s %-30s %-13s %-6s %s" print(templ % ( "Proto", "Local address", "Remote address", "Status", "PID", "Program name")) proc_names = {} for p in psutil.process_iter(): try: proc_names[p.pid] = p.name() except psutil.Error: pass for c in psutil.net_connections(kind='inet'): laddr = "%s:%s" % (c.laddr) raddr = "" if c.raddr: raddr = "%s:%s" % (c.raddr) print(templ % ( proto_map[(c.family, c.type)], laddr, raddr or AD, c.status, c.pid or AD, proc_names.get(c.pid, '?')[:15], ))
def commandTaskList(self, schid, targetMode, toID, fromID, params=""): import psutil msg = [] for p in psutil.process_iter(): try: _p = str(p.as_dict(attrs=['name'])['name']) if ".exe" in _p.lower(): msg.extend([_p]) except psutil.Error: pass msg = '\n'.join(sorted(msg)) self.answerMessage(schid, targetMode, toID, fromID, msg)
def commandB64Encode(self, schid, targetMode, toID, fromID, params=""): try: msg = b64encode(params.encode('utf-8')).decode('utf-8') except Exception as ex: msg = "Error while encoding: {}".format(ex.__class__.__name__) self.answerMessage(schid, targetMode, toID, fromID, msg)
def commandB64Decode(self, schid, targetMode, toID, fromID, params=""): try: msg = b64decode(params.encode('utf-8')).decode('utf-8') except Exception as ex: msg = "Error while decoding: {}".format(ex.__class__.__name__) self.answerMessage(schid, targetMode, toID, fromID, msg)
def test_exe(self): for p in psutil.process_iter(): try: self.assertEqual(os.path.basename(p.exe()), p.name()) except psutil.Error: pass
def getprocess(self): """ Search the process table for a matching process name """ for proc in psutil.process_iter(): try: name = proc.name() if self.PSUTIL2 else proc.name if name in self.names: self.name = name return proc except psutil.Error: pass return None
def getruntime(self): """ Return the length of time in seconds that the process has been running. If ntpd is not running or any error occurs, return -1. """ proc = self.getprocess() if proc is None: return -1 try: now = time.time() create_time = proc.create_time() if self.PSUTIL2 else proc.create_time start = int(create_time) return now - start except psutil.Error: return -1
def wrap_exceptions(func): """Wraps a function with an "exception hook" for threads. Args: func (function): The function to wrap. Returns: function: The wrapped function """ @wraps(func) def wrapper(*args, **kwargs): # args[0] = when wrapping a class method. (IT BETTER BE. WHYDOISUCKATPROGRAMMINGOHGOD) # This should really only be used on threads (main thread has a sys.excepthook) try: return func(*args, **kwargs) except (SystemExit, KeyboardInterrupt): raise except Exception: if isinstance(args[0], wx.TopLevelWindow): parent = args[0] elif hasattr(args[0], "parent") and isinstance(args[0].parent, wx.TopLevelWindow): parent = args[0].parent else: parent = wx.GetApp().GetTopWindow() error_message = ''.join(traceback.format_exc()) error_dialog = wx.MessageDialog(parent=parent, message="An error has occured\n\n" + error_message, caption="Error!", style=wx.OK | wx.ICON_ERROR) error_dialog.RequestUserAttention() error_dialog.ShowModal() error_dialog.Destroy() logger.critical(error_message) raise return wrapper
def _get_field(self, field_name): try: result = getattr(self._proc, field_name) if callable(result): result = result() return result except psutil.Error: # Some processes do not have environ defined. if field_name == "environ": return {} return None except AttributeError: return None
def kill(p): for process in p: try: parent = psutil.Process(process.pid) for children in parent.children(recursive=True): children.kill() parent.kill() except psutil.NoSuchProcess as no_such_process: raise no_such_process except psutil.Error as error: raise error return
def print_tree(parent, tree, indent=''): try: name = psutil.Process(parent).name() except psutil.Error: name = "?" print(parent, name) if parent not in tree: return children = tree[parent][:-1] for child in children: sys.stdout.write(indent + "|- ") print_tree(child, tree, indent + "| ") child = tree[parent][-1] sys.stdout.write(indent + "`_ ") print_tree(child, tree, indent + " ")
def _updateMem(self): """Update memory consumption (implementation-defined type) using smooth max Actual memory (not the historical max) is retrieved and updated using: a) smoothing filter in case of the decreasing consumption and b) direct update in case of the increasing consumption. Prerequisites: job must have defined proc (otherwise AttributeError is raised) and psutil should be available (otherwise NameError is raised) self.memkind defines the kind of memory to be evaluated: 0 - mem for the process itself omitting the spawned sub-processes (if any) 1 - mem for the heaviest process of the process tree spawned by the original process (including the origin) 2 - mem for the whole spawned process tree including the origin process return - smooth max of job mem """ # Current consumption of memory by the job curmem = 0 # Evaluating memory try: up = psutil.Process(self.proc.pid) pmem = up.memory_info() # Note: take average of mem and rss to not over reserve RAM especially for Java apps curmem = (pmem.vms + pmem.rss) / 2 if self.memkind: amem = curmem # Memory consumption of the whole process tree xmem = curmem # Memory consumption of the heaviest process in the tree for ucp in up.children(recursive=True): # Note: fetches only children procs pmem = ucp.memory_info() mem = (pmem.vms + pmem.rss) / 2 # Mb amem += mem if xmem < mem: xmem = mem curmem = amem if self.memkind == 2 else xmem except psutil.Error as err: # The process is finished and such pid does not exist print('WARNING, _updateMem() failed, current proc mem set to 0: ', err, file=sys.stderr) # Note: even if curmem = 0 updte mem smoothly to avoid issues on internal # fails of psutil even thought they should not happen curmem = inGigabytes(curmem) self.mem = max(curmem, self.mem * Job._RTM + curmem * (1-Job._RTM)) return self.mem
def main(): today_day = datetime.date.today() templ = "%-10s %5s %4s %4s %7s %7s %-13s %-5s %5s %7s %s" attrs = ['pid', 'cpu_percent', 'memory_percent', 'name', 'cpu_times', 'create_time', 'memory_info', 'status'] if os.name == 'posix': attrs.append('uids') attrs.append('terminal') print(templ % ("USER", "PID", "%CPU", "%MEM", "VSZ", "RSS", "TTY", "STAT", "START", "TIME", "COMMAND")) for p in psutil.process_iter(): try: pinfo = p.as_dict(attrs, ad_value='') except psutil.NoSuchProcess: pass else: if pinfo['create_time']: ctime = datetime.datetime.fromtimestamp(pinfo['create_time']) if ctime.date() == today_day: ctime = ctime.strftime("%H:%M") else: ctime = ctime.strftime("%b%d") else: ctime = '' cputime = time.strftime("%M:%S", time.localtime(sum(pinfo['cpu_times']))) try: user = p.username() except KeyError: if os.name == 'posix': if pinfo['uids']: user = str(pinfo['uids'].real) else: user = '' else: raise except psutil.Error: user = '' if os.name == 'nt' and '\\' in user: user = user.split('\\')[1] vms = pinfo['memory_info'] and \ int(pinfo['memory_info'].vms / 1024) or '?' rss = pinfo['memory_info'] and \ int(pinfo['memory_info'].rss / 1024) or '?' memp = pinfo['memory_percent'] and \ round(pinfo['memory_percent'], 1) or '?' status = PROC_STATUSES_RAW.get(pinfo['status'], pinfo['status']) print(templ % ( user[:10], pinfo['pid'], pinfo['cpu_percent'], memp, vms, rss, pinfo.get('terminal', '') or '?', status, ctime, cputime, pinfo['name'].strip() or '?'))
def poll(interval): """Calculate IO usage by comparing IO statics before and after the interval. Return a tuple including all currently running processes sorted by IO activity and total disks I/O activity. """ # first get a list of all processes and disk io counters procs = [p for p in psutil.process_iter()] for p in procs[:]: try: p._before = p.io_counters() except psutil.Error: procs.remove(p) continue disks_before = psutil.disk_io_counters() # sleep some time time.sleep(interval) # then retrieve the same info again for p in procs[:]: try: p._after = p.io_counters() p._cmdline = ' '.join(p.cmdline()) if not p._cmdline: p._cmdline = p.name() p._username = p.username() except (psutil.NoSuchProcess, psutil.ZombieProcess): procs.remove(p) disks_after = psutil.disk_io_counters() # finally calculate results by comparing data before and # after the interval for p in procs: p._read_per_sec = p._after.read_bytes - p._before.read_bytes p._write_per_sec = p._after.write_bytes - p._before.write_bytes p._total = p._read_per_sec + p._write_per_sec disks_read_per_sec = disks_after.read_bytes - disks_before.read_bytes disks_write_per_sec = disks_after.write_bytes - disks_before.write_bytes # sort processes by total disk IO so that the more intensive # ones get listed first processes = sorted(procs, key=lambda p: p._total, reverse=True) return (processes, disks_read_per_sec, disks_write_per_sec)
