diff --git a/stubs/psutil/psutil/_common.pyi b/stubs/psutil/psutil/_common.pyi index a16ecda3dca9..97e8c1e49844 100644 --- a/stubs/psutil/psutil/_common.pyi +++ b/stubs/psutil/psutil/_common.pyi @@ -1,66 +1,70 @@ import enum -from _typeshed import Incomplete, StrOrBytesPath, SupportsWrite +import io +import threading +from _typeshed import ConvertibleToFloat, FileDescriptorOrPath, Incomplete, StrOrBytesPath, SupportsWrite from collections.abc import Callable from socket import AF_INET6 as AF_INET6, AddressFamily, SocketKind -from typing import Literal, NamedTuple, TypeVar, overload - -POSIX: bool -WINDOWS: bool -LINUX: bool -MACOS: bool -OSX: bool -FREEBSD: bool -OPENBSD: bool -NETBSD: bool -BSD: bool -SUNOS: bool -AIX: bool - -STATUS_RUNNING: Literal["running"] -STATUS_SLEEPING: Literal["sleeping"] -STATUS_DISK_SLEEP: Literal["disk-sleep"] -STATUS_STOPPED: Literal["stopped"] -STATUS_TRACING_STOP: Literal["tracing-stop"] -STATUS_ZOMBIE: Literal["zombie"] -STATUS_DEAD: Literal["dead"] -STATUS_WAKE_KILL: Literal["wake-kill"] -STATUS_WAKING: Literal["waking"] -STATUS_IDLE: Literal["idle"] -STATUS_LOCKED: Literal["locked"] -STATUS_WAITING: Literal["waiting"] -STATUS_SUSPENDED: Literal["suspended"] -STATUS_PARKED: Literal["parked"] - -CONN_ESTABLISHED: str -CONN_SYN_SENT: str -CONN_SYN_RECV: str -CONN_FIN_WAIT1: str -CONN_FIN_WAIT2: str -CONN_TIME_WAIT: str -CONN_CLOSE: str -CONN_CLOSE_WAIT: str -CONN_LAST_ACK: str -CONN_LISTEN: str -CONN_CLOSING: str -CONN_NONE: str -NIC_DUPLEX_FULL: int -NIC_DUPLEX_HALF: int -NIC_DUPLEX_UNKNOWN: int +from typing import BinaryIO, Final, NamedTuple, SupportsIndex, TypeVar, overload +from typing_extensions import ParamSpec + +POSIX: Final[bool] +WINDOWS: Final[bool] +LINUX: Final[bool] +MACOS: Final[bool] +OSX: Final[bool] +FREEBSD: Final[bool] +OPENBSD: Final[bool] +NETBSD: Final[bool] +BSD: Final[bool] +SUNOS: Final[bool] +AIX: Final[bool] + +STATUS_RUNNING: Final = "running" +STATUS_SLEEPING: Final = "sleeping" +STATUS_DISK_SLEEP: Final = "disk-sleep" +STATUS_STOPPED: Final = "stopped" +STATUS_TRACING_STOP: Final = "tracing-stop" +STATUS_ZOMBIE: Final = "zombie" +STATUS_DEAD: Final = "dead" +STATUS_WAKE_KILL: Final = "wake-kill" +STATUS_WAKING: Final = "waking" +STATUS_IDLE: Final = "idle" +STATUS_LOCKED: Final = "locked" +STATUS_WAITING: Final = "waiting" +STATUS_SUSPENDED: Final = "suspended" +STATUS_PARKED: Final = "parked" + +CONN_ESTABLISHED: Final = "ESTABLISHED" +CONN_SYN_SENT: Final = "SYN_SENT" +CONN_SYN_RECV: Final = "SYN_RECV" +CONN_FIN_WAIT1: Final = "FIN_WAIT1" +CONN_FIN_WAIT2: Final = "FIN_WAIT2" +CONN_TIME_WAIT: Final = "TIME_WAIT" +CONN_CLOSE: Final = "CLOSE" +CONN_CLOSE_WAIT: Final = "CLOSE_WAIT" +CONN_LAST_ACK: Final = "LAST_ACK" +CONN_LISTEN: Final = "LISTEN" +CONN_CLOSING: Final = "CLOSING" +CONN_NONE: Final = "NONE" class NicDuplex(enum.IntEnum): NIC_DUPLEX_FULL = 2 NIC_DUPLEX_HALF = 1 NIC_DUPLEX_UNKNOWN = 0 -POWER_TIME_UNKNOWN: int -POWER_TIME_UNLIMITED: int +NIC_DUPLEX_FULL: Final = NicDuplex.NIC_DUPLEX_FULL +NIC_DUPLEX_HALF: Final = NicDuplex.NIC_DUPLEX_HALF +NIC_DUPLEX_UNKNOWN: Final = NicDuplex.NIC_DUPLEX_UNKNOWN class BatteryTime(enum.IntEnum): POWER_TIME_UNKNOWN = -1 POWER_TIME_UNLIMITED = -2 -ENCODING: str -ENCODING_ERRS: str +POWER_TIME_UNKNOWN: Final = BatteryTime.POWER_TIME_UNKNOWN +POWER_TIME_UNLIMITED: Final = BatteryTime.POWER_TIME_UNLIMITED + +ENCODING: Final[str] +ENCODING_ERRS: Final[str] class sswap(NamedTuple): total: int @@ -210,44 +214,42 @@ class addr(NamedTuple): conn_tmap: dict[str, tuple[list[AddressFamily], list[SocketKind]]] class Error(Exception): - __module__: str - msg: Incomplete + msg: str def __init__(self, msg: str = ...) -> None: ... class NoSuchProcess(Error): - __module__: str - pid: Incomplete - name: Incomplete - msg: Incomplete - def __init__(self, pid, name=None, msg=None) -> None: ... + pid: int + name: str | None + msg: str + def __init__(self, pid: int, name: str | None = None, msg: str | None = None) -> None: ... class ZombieProcess(NoSuchProcess): - __module__: str - pid: Incomplete - ppid: Incomplete - name: Incomplete - msg: Incomplete - def __init__(self, pid, name=None, ppid=None, msg=None) -> None: ... + ppid: int | None + def __init__(self, pid: int, name: str | None = None, ppid: int | None = None, msg: str | None = None) -> None: ... class AccessDenied(Error): - __module__: str - pid: Incomplete - name: Incomplete - msg: Incomplete - def __init__(self, pid=None, name=None, msg=None) -> None: ... + pid: int | None + name: str | None + msg: str + def __init__(self, pid: int | None = None, name: str | None = None, msg: str | None = None) -> None: ... class TimeoutExpired(Error): - __module__: str - seconds: Incomplete - pid: Incomplete - name: Incomplete - def __init__(self, seconds, pid=None, name=None) -> None: ... + seconds: float + pid: int | None + name: str | None + msg: str + def __init__(self, seconds: float, pid: int | None = None, name: str | None = None) -> None: ... + +_P = ParamSpec("_P") +_R = TypeVar("_R") + +def usage_percent(used: ConvertibleToFloat, total: float, round_: SupportsIndex | None = None) -> float: ... -_Func = TypeVar("_Func", bound=Callable[..., Incomplete]) +# returned function has `cache_clear()` attribute: +def memoize(fun: Callable[_P, _R]) -> Callable[_P, _R]: ... -def usage_percent(used, total, round_: int | None = None) -> float: ... -def memoize(fun: _Func) -> _Func: ... -def memoize_when_activated(fun: _Func) -> _Func: ... +# returned function has `cache_activate(proc)` and `cache_deactivate(proc)` attributes: +def memoize_when_activated(fun: Callable[_P, _R]) -> Callable[_P, _R]: ... def isfile_strict(path: StrOrBytesPath) -> bool: ... def path_exists_strict(path: StrOrBytesPath) -> bool: ... def supports_ipv6() -> bool: ... @@ -258,29 +260,30 @@ def socktype_to_enum(num: int) -> SocketKind: ... def conn_to_ntuple(fd: int, fam: int, type_: int, laddr, raddr, status: str, status_map, pid: int) -> sconn: ... @overload def conn_to_ntuple(fd: int, fam: int, type_: int, laddr, raddr, status: str, status_map, pid: None = None) -> pconn: ... -def deprecated_method(replacement: str) -> Callable[[_Func], _Func]: ... +def deprecated_method(replacement: str) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: ... class _WrapNumbers: - lock: Incomplete - cache: Incomplete - reminders: Incomplete - reminder_keys: Incomplete + lock: threading.Lock + cache: dict[Incomplete, Incomplete] + reminders: dict[Incomplete, Incomplete] + reminder_keys: dict[Incomplete, Incomplete] def __init__(self) -> None: ... def run(self, input_dict, name): ... def cache_clear(self, name=None) -> None: ... - def cache_info(self): ... + def cache_info(self) -> tuple[dict[Incomplete, Incomplete], dict[Incomplete, Incomplete], dict[Incomplete, Incomplete]]: ... def wrap_numbers(input_dict, name: str): ... -def open_binary(fname): ... -def open_text(fname): ... -def cat(fname, fallback=..., _open=...): ... -def bcat(fname, fallback=...): ... +def open_binary(fname: FileDescriptorOrPath) -> BinaryIO: ... +def open_text(fname: FileDescriptorOrPath) -> io.TextIOWrapper: ... +def cat(fname: FileDescriptorOrPath, fallback=..., _open=...): ... +def bcat(fname: FileDescriptorOrPath, fallback=...): ... def bytes2human(n: int, format: str = "%(value).1f%(symbol)s") -> str: ... def get_procfs_path() -> str: ... +def decode(s: bytes) -> str: ... def term_supports_colors(file: SupportsWrite[str] = ...) -> bool: ... def hilite(s: str, color: str | None = None, bold: bool = False) -> str: ... def print_color(s: str, color: str | None = None, bold: bool = False, file: SupportsWrite[str] = ...) -> None: ... -def debug(msg) -> None: ... +def debug(msg: str | Exception) -> None: ... __all__ = [ # OS constants diff --git a/stubs/psutil/psutil/_psposix.pyi b/stubs/psutil/psutil/_psposix.pyi index 46c218915a04..57b6ff163616 100644 --- a/stubs/psutil/psutil/_psposix.pyi +++ b/stubs/psutil/psutil/_psposix.pyi @@ -1,6 +1,26 @@ -def pid_exists(pid): ... -def wait_pid(pid, timeout=None, proc_name=None, _waitpid=..., _timer=..., _min=..., _sleep=..., _pid_exists=...): ... -def disk_usage(path): ... -def get_terminal_map(): ... +import sys +from _typeshed import FileDescriptorOrPath, StrOrBytesPath + +from ._common import sdiskusage + +def pid_exists(pid: int) -> bool: ... +def wait_pid( + pid: int, + timeout: float | None = None, + proc_name: str | None = None, + _waitpid=..., + _timer=..., + _min=..., + _sleep=..., + _pid_exists=..., +): ... + +if sys.platform == "darwin": + def disk_usage(path: StrOrBytesPath) -> sdiskusage: ... + +else: + def disk_usage(path: FileDescriptorOrPath) -> sdiskusage: ... + +def get_terminal_map() -> dict[int, str]: ... __all__ = ["pid_exists", "wait_pid", "disk_usage", "get_terminal_map"] diff --git a/stubs/psutil/psutil/_psutil_posix.pyi b/stubs/psutil/psutil/_psutil_posix.pyi index 8a8d0bb58857..4ae2d853e2d8 100644 --- a/stubs/psutil/psutil/_psutil_posix.pyi +++ b/stubs/psutil/psutil/_psutil_posix.pyi @@ -1,33 +1,34 @@ import sys +from typing import Final if sys.platform == "linux": - RLIMIT_AS: int - RLIMIT_CORE: int - RLIMIT_CPU: int - RLIMIT_DATA: int - RLIMIT_FSIZE: int - RLIMIT_LOCKS: int - RLIMIT_MEMLOCK: int - RLIMIT_MSGQUEUE: int - RLIMIT_NICE: int - RLIMIT_NOFILE: int - RLIMIT_NPROC: int - RLIMIT_RSS: int - RLIMIT_RTPRIO: int - RLIMIT_RTTIME: int - RLIMIT_SIGPENDING: int - RLIMIT_STACK: int - RLIM_INFINITY: int + RLIMIT_AS: Final[int] + RLIMIT_CORE: Final[int] + RLIMIT_CPU: Final[int] + RLIMIT_DATA: Final[int] + RLIMIT_FSIZE: Final[int] + RLIMIT_LOCKS: Final[int] + RLIMIT_MEMLOCK: Final[int] + RLIMIT_MSGQUEUE: Final[int] + RLIMIT_NICE: Final[int] + RLIMIT_NOFILE: Final[int] + RLIMIT_NPROC: Final[int] + RLIMIT_RSS: Final[int] + RLIMIT_RTPRIO: Final[int] + RLIMIT_RTTIME: Final[int] + RLIMIT_SIGPENDING: Final[int] + RLIMIT_STACK: Final[int] + RLIM_INFINITY: Final[int] -def getpagesize(*args, **kwargs): ... -def getpriority(*args, **kwargs): ... -def net_if_addrs(*args, **kwargs): ... -def net_if_flags(*args, **kwargs): ... -def net_if_is_running(*args, **kwargs): ... -def net_if_mtu(*args, **kwargs): ... +def getpagesize() -> int: ... +def getpriority(pid: int, /) -> int: ... +def net_if_addrs() -> list[tuple[str, int, str | None, str | None, str | None, str | None]]: ... +def net_if_flags(nic_name: str, /) -> list[str]: ... +def net_if_is_running(nic_name: str, /) -> bool: ... +def net_if_mtu(nic_name: str, /) -> int: ... if sys.platform == "darwin": - AF_LINK: int - def net_if_duplex_speed(*args, **kwargs): ... + AF_LINK: Final[int] + def net_if_duplex_speed(nic_name: str, /) -> list[int]: ... -def setpriority(*args, **kwargs): ... +def setpriority(pid: int, priority: int, /) -> None: ...