# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Common objects shared by __init__.py and _ps*.py modules."""
# Note: this module is imported by setup.py so it should not import
# psutil or third-party modules.
from __future__ import division
from __future__ import print_function
import collections
import contextlib
import errno
import functools
import os
import socket
import stat
import sys
import threading
import warnings
from collections import namedtuple
from socket import AF_INET
from socket import SOCK_DGRAM
from socket import SOCK_STREAM
try:
from socket import AF_INET6
except ImportError:
AF_INET6 = None
try:
from socket import AF_UNIX
except ImportError:
AF_UNIX = None
if sys.version_info >= (3, 4):
import enum
else:
enum = None
# can't take it from _common.py as this script is imported by setup.py
PY3 = sys.version_info[0] == 3
PSUTIL_DEBUG = bool(os.getenv('PSUTIL_DEBUG', 0))
_DEFAULT = object()
__all__ = [
# OS constants
'FREEBSD', 'BSD', 'LINUX', 'NETBSD', 'OPENBSD', 'MACOS', 'OSX', 'POSIX',
'SUNOS', 'WINDOWS',
# connection constants
'CONN_CLOSE', 'CONN_CLOSE_WAIT', 'CONN_CLOSING', 'CONN_ESTABLISHED',
'CONN_FIN_WAIT1', 'CONN_FIN_WAIT2', 'CONN_LAST_ACK', 'CONN_LISTEN',
'CONN_NONE', 'CONN_SYN_RECV', 'CONN_SYN_SENT', 'CONN_TIME_WAIT',
# net constants
'NIC_DUPLEX_FULL', 'NIC_DUPLEX_HALF', 'NIC_DUPLEX_UNKNOWN',
# process status constants
'STATUS_DEAD', 'STATUS_DISK_SLEEP', 'STATUS_IDLE', 'STATUS_LOCKED',
'STATUS_RUNNING', 'STATUS_SLEEPING', 'STATUS_STOPPED', 'STATUS_SUSPENDED',
'STATUS_TRACING_STOP', 'STATUS_WAITING', 'STATUS_WAKE_KILL',
'STATUS_WAKING', 'STATUS_ZOMBIE', 'STATUS_PARKED',
# other constants
'ENCODING', 'ENCODING_ERRS', 'AF_INET6',
# named tuples
'pconn', 'pcputimes', 'pctxsw', 'pgids', 'pio', 'pionice', 'popenfile',
'pthread', 'puids', 'sconn', 'scpustats', 'sdiskio', 'sdiskpart',
'sdiskusage', 'snetio', 'snicaddr', 'snicstats', 'sswap', 'suser',
# utility functions
'conn_tmap', 'deprecated_method', 'isfile_strict', 'memoize',
'parse_environ_block', 'path_exists_strict', 'usage_percent',
'supports_ipv6', 'sockfam_to_enum', 'socktype_to_enum', "wrap_numbers",
'open_text', 'open_binary', 'cat', 'bcat',
'bytes2human', 'conn_to_ntuple', 'debug',
# shell utils
'hilite', 'term_supports_colors', 'print_color',
]
# ===================================================================
# --- OS constants
# ===================================================================
POSIX = os.name == "posix"
WINDOWS = os.name == "nt"
LINUX = sys.platform.startswith("linux")
MACOS = sys.platform.startswith("darwin")
OSX = MACOS # deprecated alias
FREEBSD = sys.platform.startswith(("freebsd", "midnightbsd"))
OPENBSD = sys.platform.startswith("openbsd")
NETBSD = sys.platform.startswith("netbsd")
BSD = FREEBSD or OPENBSD or NETBSD
SUNOS = sys.platform.startswith(("sunos", "solaris"))
AIX = sys.platform.startswith("aix")
# ===================================================================
# --- API constants
# ===================================================================
# Process.status()
STATUS_RUNNING = "running"
STATUS_SLEEPING = "sleeping"
STATUS_DISK_SLEEP = "disk-sleep"
STATUS_STOPPED = "stopped"
STATUS_TRACING_STOP = "tracing-stop"
STATUS_ZOMBIE = "zombie"
STATUS_DEAD = "dead"
STATUS_WAKE_KILL = "wake-kill"
STATUS_WAKING = "waking"
STATUS_IDLE = "idle" # Linux, macOS, FreeBSD
STATUS_LOCKED = "locked" # FreeBSD
STATUS_WAITING = "waiting" # FreeBSD
STATUS_SUSPENDED = "suspended" # NetBSD
STATUS_PARKED = "parked" # Linux
# Process.connections() and psutil.net_connections()
CONN_ESTABLISHED = "ESTABLISHED"
CONN_SYN_SENT = "SYN_SENT"
CONN_SYN_RECV = "SYN_RECV"
CONN_FIN_WAIT1 = "FIN_WAIT1"
CONN_FIN_WAIT2 = "FIN_WAIT2"
CONN_TIME_WAIT = "TIME_WAIT"
CONN_CLOSE = "CLOSE"
CONN_CLOSE_WAIT = "CLOSE_WAIT"
CONN_LAST_ACK = "LAST_ACK"
CONN_LISTEN = "LISTEN"
CONN_CLOSING = "CLOSING"
CONN_NONE = "NONE"
# net_if_stats()
if enum is None:
NIC_DUPLEX_FULL = 2
NIC_DUPLEX_HALF = 1
NIC_DUPLEX_UNKNOWN = 0
else:
class NicDuplex(enum.IntEnum):
NIC_DUPLEX_FULL = 2
NIC_DUPLEX_HALF = 1
NIC_DUPLEX_UNKNOWN = 0
globals().update(NicDuplex.__members__)
# sensors_battery()
if enum is None:
POWER_TIME_UNKNOWN = -1
POWER_TIME_UNLIMITED = -2
else:
class BatteryTime(enum.IntEnum):
POWER_TIME_UNKNOWN = -1
POWER_TIME_UNLIMITED = -2
globals().update(BatteryTime.__members__)
# --- others
ENCODING = sys.getfilesystemencoding()
if not PY3:
ENCODING_ERRS = "replace"
else:
try:
ENCODING_ERRS = sys.getfilesystemencodeerrors() # py 3.6
except AttributeError:
ENCODING_ERRS = "surrogateescape" if POSIX else "replace"
# ===================================================================
# --- namedtuples
# ===================================================================
# --- for system functions
# psutil.swap_memory()
sswap = namedtuple('sswap', ['total', 'used', 'free', 'percent', 'sin',
'sout'])
# psutil.disk_usage()
sdiskusage = namedtuple('sdiskusage', ['total', 'used', 'free', 'percent'])
# psutil.disk_io_counters()
sdiskio = namedtuple('sdiskio', ['read_count', 'write_count',
'read_bytes', 'write_bytes',
'read_time', 'write_time'])
# psutil.disk_partitions()
sdiskpart = namedtuple('sdiskpart', ['device', 'mountpoint', 'fstype', 'opts',
'maxfile', 'maxpath'])
# psutil.net_io_counters()
snetio = namedtuple('snetio', ['bytes_sent', 'bytes_recv',
'packets_sent', 'packets_recv',
'errin', 'errout',
'dropin', 'dropout'])
# psutil.users()
suser = namedtuple('suser', ['name', 'terminal', 'host', 'started', 'pid'])
# psutil.net_connections()
sconn = namedtuple('sconn', ['fd', 'family', 'type', 'laddr', 'raddr',
'status', 'pid'])
# psutil.net_if_addrs()
snicaddr = namedtuple('snicaddr',
['family', 'address', 'netmask', 'broadcast', 'ptp'])
# psutil.net_if_stats()
snicstats = namedtuple('snicstats',
['isup', 'duplex', 'speed', 'mtu', 'flags'])
# psutil.cpu_stats()
scpustats = namedtuple(
'scpustats', ['ctx_switches', 'interrupts', 'soft_interrupts', 'syscalls'])
# psutil.cpu_freq()
scpufreq = namedtuple('scpufreq', ['current', 'min', 'max'])
# psutil.sensors_temperatures()
shwtemp = namedtuple(
'shwtemp', ['label', 'current', 'high', 'critical'])
# psutil.sensors_battery()
sbattery = namedtuple('sbattery', ['percent', 'secsleft', 'power_plugged'])
# psutil.sensors_fans()
sfan = namedtuple('sfan', ['label', 'current'])
# --- for Process methods
# psutil.Process.cpu_times()
pcputimes = namedtuple('pcputimes',
['user', 'system', 'children_user', 'children_system'])
# psutil.Process.open_files()
popenfile = namedtuple('popenfile', ['path', 'fd'])
# psutil.Process.threads()
pthread = namedtuple('pthread', ['id', 'user_time', 'system_time'])
# psutil.Process.uids()
puids = namedtuple('puids', ['real', 'effective', 'saved'])
# psutil.Process.gids()
pgids = namedtuple('pgids', ['real', 'effective', 'saved'])
# psutil.Process.io_counters()
pio = namedtuple('pio', ['read_count', 'write_count',
'read_bytes', 'write_bytes'])
# psutil.Process.ionice()
pionice = namedtuple('pionice', ['ioclass', 'value'])
# psutil.Process.ctx_switches()
pctxsw = namedtuple('pctxsw', ['voluntary', 'involuntary'])
# psutil.Process.connections()
pconn = namedtuple('pconn', ['fd', 'family', 'type', 'laddr', 'raddr',
'status'])
# psutil.connections() and psutil.Process.connections()
addr = namedtuple('addr', ['ip', 'port'])
# ===================================================================
# --- Process.connections() 'kind' parameter mapping
# ===================================================================
conn_tmap = {
"all": ([AF_INET, AF_INET6, AF_UNIX], [SOCK_STREAM, SOCK_DGRAM]),
"tcp": ([AF_INET, AF_INET6], [SOCK_STREAM]),
"tcp4": ([AF_INET], [SOCK_STREAM]),
"udp": ([AF_INET, AF_INET6], [SOCK_DGRAM]),
"udp4": ([AF_INET], [SOCK_DGRAM]),
"inet": ([AF_INET, AF_INET6], [SOCK_STREAM, SOCK_DGRAM]),
"inet4": ([AF_INET], [SOCK_STREAM, SOCK_DGRAM]),
"inet6": ([AF_INET6], [SOCK_STREAM, SOCK_DGRAM]),
}
if AF_INET6 is not None:
conn_tmap.update({
"tcp6": ([AF_INET6], [SOCK_STREAM]),
"udp6": ([AF_INET6], [SOCK_DGRAM]),
})
if AF_UNIX is not None:
conn_tmap.update({
"unix": ([AF_UNIX], [SOCK_STREAM, SOCK_DGRAM]),
})
# =====================================================================
# --- Exceptions
# =====================================================================
class Error(Exception):
"""Base exception class. All other psutil exceptions inherit
from this one.
"""
__module__ = 'psutil'
def _infodict(self, attrs):
info = collections.OrderedDict()
for name in attrs:
value = getattr(self, name, None)
if value:
info[name] = value
elif name == "pid" and value == 0:
info[name] = value
return info
def __str__(self):
# invoked on `raise Error`
info = self._infodict(("pid", "ppid", "name"))
if info:
details = "(%s)" % ", ".join(
["%s=%r" % (k, v) for k, v in info.items()])
else:
details = None
return " ".join([x for x in (getattr(self, "msg", ""), details) if x])
def __repr__(self):
# invoked on `repr(Error)`
info = self._infodict(("pid", "ppid", "name", "seconds", "msg"))
details = ", ".join(["%s=%r" % (k, v) for k, v in info.items()])
return "psutil.%s(%s)" % (self.__class__.__name__, details)
class NoSuchProcess(Error):
"""Exception raised when a process with a certain PID doesn't
or no longer exists.
"""
__module__ = 'psutil'
def __init__(self, pid, name=None, msg=None):
Error.__init__(self)
self.pid = pid
self.name = name
self.msg = msg or "process no longer exists"
class ZombieProcess(NoSuchProcess):
"""Exception raised when querying a zombie process. This is
raised on macOS, BSD and Solaris only, and not always: depending
on the query the OS may be able to succeed anyway.
On Linux all zombie processes are querable (hence this is never
raised). Windows doesn't have zombie processes.
"""
__module__ = 'psutil'
def __init__(self, pid, name=None, ppid=None, msg=None):
NoSuchProcess.__init__(self, pid, name, msg)
self.ppid = ppid
self.msg = msg or "PID still exists but it's a zombie"
class AccessDenied(Error):
"""Exception raised when permission to perform an action is denied."""
__module__ = 'psutil'
def __init__(self, pid=None, name=None, msg=None):
Error.__init__(self)
self.pid = pid
Loading ...