github-starred/nvitop
2
0
Fork 0
mirror of https://github.com/XuehaiPan/nvitop.git synced 2026-05-15 14:15:55 -06:00
Code Issues 29 Projects Packages Wiki Activity Actions

deps(python): drop Python 3.6 support (#56)

Browse source
This commit is contained in:
Xuehai Pan 2023-03-15 16:56:24 +08:00 • committed by GitHub
parent 970d515eaf
commit c5ce570c72
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 285 additions and 286 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/PULL_REQUEST_TEMPLATE.md vendored
View file

@ -14,7 +14,7 @@
- Operating system and version: [e.g. Ubuntu 20.04 LTS / Windows 10 Build 19043.1110]
- Terminal emulator and version: [e.g. GNOME Terminal 3.36.2 / Windows Terminal 1.8.1521.0]
- Python version: [e.g. `3.6.6` / `3.9.6`]
- Python version: [e.g. `3.7.2` / `3.9.6`]
- NVML version (driver version): [e.g. `460.84`]
- `nvitop` version or commit: [e.g. `0.10.0` / `0.10.1.dev7+ga083321` / `main@75ae3c`]
- `python-ml-py` version: [e.g. `11.450.51`]

14
.github/workflows/build.yaml vendored
View file

@ -50,19 +50,19 @@ jobs:
id: py
uses: actions/setup-python@v4
with:
python-version: "3.6 - 3.11"
python-version: "3.7 - 3.11"
update-environment: true
- name: Set up Python 3.6
id: py36
- name: Set up Python 3.7
id: py37
uses: actions/setup-python@v4
with:
python-version: "3.6"
python-version: "3.7"
update-environment: false
- name: Check syntax (Python 3.6)
- name: Check syntax (Python 3.7)
run: |
"${{ steps.py36.outputs.python-path }}" -m compileall nvitop
"${{ steps.py37.outputs.python-path }}" -m compileall nvitop
- name: Upgrade build dependencies
run: python -m pip install --upgrade pip setuptools wheel build
@ -123,7 +123,7 @@ jobs:
uses: actions/setup-python@v4
if: startsWith(github.ref, 'refs/tags/')
with:
python-version: "3.6 - 3.11"
python-version: "3.7 - 3.11"
update-environment: true
- name: Set __release__

12
.github/workflows/lint.yaml vendored
View file

@ -27,19 +27,19 @@ jobs:
id: py
uses: actions/setup-python@v4
with:
python-version: "3.6 - 3.11"
python-version: "3.7 - 3.11"
update-environment: true
- name: Set up Python 3.6
id: py36
- name: Set up Python 3.7
id: py37
uses: actions/setup-python@v4
with:
python-version: "3.6"
python-version: "3.7"
update-environment: false
- name: Check syntax (Python 3.6)
- name: Check syntax (Python 3.7)
run: |
"${{ steps.py36.outputs.python-path }}" -m compileall nvitop
"${{ steps.py37.outputs.python-path }}" -m compileall nvitop
- name: Upgrade pip
run: |

2
.pre-commit-config.yaml
View file

@ -37,7 +37,7 @@ repos:
rev: v3.3.1
hooks:
- id: pyupgrade
args: [--py36-plus]
args: [--py37-plus]
stages: [commit, push, manual]
- repo: https://github.com/pycqa/flake8
rev: 6.0.0

2
.pylintrc
View file

@ -84,7 +84,7 @@ persistent=yes
# Minimum Python version to use for version dependent checks. Will default to
# the version used to run pylint.
py-version=3.6
py-version=3.7
# Discover python modules and packages in the file system subtree.
recursive=no

2
CHANGELOG.md
View file

@ -25,7 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Removed
-
- Drop Python 3.6 support by [@XuehaiPan](https://github.com/XuehaiPan) in [#56](https://github.com/XuehaiPan/nvitop/pull/56).
------

4
README.md
View file

@ -2,7 +2,7 @@
<!-- markdownlint-disable html -->
![Python 3.6+](https://img.shields.io/badge/Python-3.6%2B-brightgreen)
![Python 3.7+](https://img.shields.io/badge/Python-3.7%2B-brightgreen)
[![PyPI](https://img.shields.io/pypi/v/nvitop?label=pypi&logo=pypi)](https://pypi.org/project/nvitop)
[![conda-forge](https://img.shields.io/conda/vn/conda-forge/nvitop?label=conda&logo=condaforge)](https://anaconda.org/conda-forge/nvitop)
[![Documentation Status](https://img.shields.io/readthedocs/nvitop?label=docs&logo=readthedocs)](https://nvitop.readthedocs.io)
@ -103,7 +103,7 @@ An interactive NVIDIA-GPU process viewer and beyond, the one-stop solution for G
## Requirements
- Python 3.6+
- Python 3.7+
- NVIDIA Management Library (NVML)
- nvidia-ml-py
- psutil

4
docs/source/index.rst
View file

@ -13,7 +13,7 @@ An interactive NVIDIA-GPU process viewer and beyond, the one-stop solution for G
.. |GitHub| image:: https://img.shields.io/badge/GitHub-Homepage-blue?logo=github
.. _GitHub: https://github.com/XuehaiPan/nvitop
.. |Python Version| image:: https://img.shields.io/badge/Python-3.6%2B-brightgreen
.. |Python Version| image:: https://img.shields.io/badge/Python-3.7%2B-brightgreen
.. _Python Version: https://pypi.org/project/nvitop
.. |PyPI Package| image:: https://img.shields.io/pypi/v/nvitop?label=pypi&logo=pypi
@ -56,7 +56,7 @@ Install from PyPI (|PyPI Package|_):
.. note::
Python 3.6+ is required, and Python versions lower than 3.6 is not supported.
Python 3.7+ is required, and Python versions lower than 3.7 is not supported.
Install from conda-forge (|Conda-forge Package|_):

58
nvitop/api/collector.py
View file

@ -17,6 +17,8 @@
"""Resource metrics collectors."""
from __future__ import annotations
import contextlib
import itertools
import math
@ -24,7 +26,7 @@ import os
import threading
import time
from collections import OrderedDict, defaultdict
from typing import Callable, Dict, Hashable, Iterable, List, NamedTuple, Optional, Tuple, Union
from typing import Callable, Hashable, Iterable, NamedTuple
from weakref import WeakSet
from nvitop.api import host
@ -37,22 +39,22 @@ __all__ = ['take_snapshots', 'collect_in_background', 'ResourceMetricCollector']
class SnapshotResult(NamedTuple): # pylint: disable=missing-class-docstring
devices: List[Snapshot]
gpu_processes: List[Snapshot]
devices: list[Snapshot]
gpu_processes: list[Snapshot]
timer = time.monotonic
def _unique(iterable: Iterable[Hashable]) -> List[Hashable]:
def _unique(iterable: Iterable[Hashable]) -> list[Hashable]:
return list(OrderedDict.fromkeys(iterable).keys())
# pylint: disable-next=too-many-branches
def take_snapshots(
devices: Optional[Union[Device, Iterable[Device]]] = None,
devices: Device | Iterable[Device] | None = None,
*,
gpu_processes: Optional[Union[bool, GpuProcess, Iterable[GpuProcess]]] = None,
gpu_processes: bool | GpuProcess | Iterable[GpuProcess] | None = None,
) -> SnapshotResult:
"""Retrieve status of demanded devices and GPU processes.
@ -182,12 +184,12 @@ def take_snapshots(
# pylint: disable-next=too-many-arguments
def collect_in_background(
on_collect: Callable[[Dict[str, float]], bool],
collector: Optional['ResourceMetricCollector'] = None,
interval: Optional[float] = None,
on_collect: Callable[[dict[str, float]], bool],
collector: ResourceMetricCollector | None = None,
interval: float | None = None,
*,
on_start: Optional[Callable[['ResourceMetricCollector'], None]] = None,
on_stop: Optional[Callable[['ResourceMetricCollector'], None]] = None,
on_start: Callable[[ResourceMetricCollector], None] | None = None,
on_stop: Callable[[ResourceMetricCollector], None] | None = None,
tag: str = 'metrics-daemon',
start: bool = True,
) -> threading.Thread:
@ -389,9 +391,9 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
def __init__(
self,
devices: Optional[Iterable[Device]] = None,
root_pids: Optional[Iterable[int]] = None,
interval: Union[int, float] = 1.0,
devices: Iterable[Device] | None = None,
root_pids: Iterable[int] | None = None,
interval: int | float = 1.0,
) -> None:
"""Initialize the resource metric collector."""
if isinstance(interval, (int, float)) and interval > 0:
@ -432,7 +434,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
)
self._daemon_running = threading.Event()
def activate(self, tag: str) -> 'ResourceMetricCollector':
def activate(self, tag: str) -> ResourceMetricCollector:
"""Start a new metric collection with the given tag.
Args:
@ -467,7 +469,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
start = activate
def deactivate(self, tag: Optional[str] = None) -> 'ResourceMetricCollector':
def deactivate(self, tag: str | None = None) -> ResourceMetricCollector:
"""Stop the current collection with the given tag and remove all sub-tags.
If the tag is not specified, deactivate the current active collection. For nested
@ -506,7 +508,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
stop = deactivate
@contextlib.contextmanager
def context(self, tag: str) -> 'ResourceMetricCollector':
def context(self, tag: str) -> ResourceMetricCollector:
"""A context manager for starting and stopping resource metric collection.
Args:
@ -529,7 +531,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
__call__ = context # alias for `with collector(tag='<tag>')`
def clear(self, tag: Optional[str] = None) -> None:
def clear(self, tag: str | None = None) -> None:
"""Reset the metric collection with the given tag.
If the tag is not specified, reset the current active collection. For nested collections,
@ -576,7 +578,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
break
buffer = buffer.prev
def collect(self) -> Dict[str, float]:
def collect(self) -> dict[str, float]:
"""Get the average resource consumption during collection."""
with self._lock:
if self._metric_buffer is None:
@ -589,11 +591,11 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
# pylint: disable-next=too-many-arguments
def daemonize(
self,
on_collect: Callable[[Dict[str, float]], bool],
interval: Optional[float] = None,
on_collect: Callable[[dict[str, float]], bool],
interval: float | None = None,
*,
on_start: Optional[Callable[['ResourceMetricCollector'], None]] = None,
on_stop: Optional[Callable[['ResourceMetricCollector'], None]] = None,
on_start: Callable[[ResourceMetricCollector], None] | None = None,
on_stop: Callable[[ResourceMetricCollector], None] | None = None,
tag: str = 'metrics-daemon',
start: bool = True,
) -> threading.Thread:
@ -753,7 +755,7 @@ class ResourceMetricCollector: # pylint: disable=too-many-instance-attributes
class _MetricBuffer: # pylint: disable=missing-class-docstring,missing-function-docstring,too-many-instance-attributes
def __init__(
self, tag: str, collector: 'ResourceMetricCollector', prev: Optional['_MetricBuffer'] = None
self, tag: str, collector: ResourceMetricCollector, prev: _MetricBuffer | None = None
) -> None:
self.collector = collector
self.prev = prev
@ -769,7 +771,7 @@ class _MetricBuffer: # pylint: disable=missing-class-docstring,missing-function
self.len = 0
def add(self, metrics: Dict[str, float], timestamp: Optional[float] = None) -> None:
def add(self, metrics: dict[str, float], timestamp: float | None = None) -> None:
if timestamp is None:
timestamp = timer()
@ -788,7 +790,7 @@ class _MetricBuffer: # pylint: disable=missing-class-docstring,missing-function
self.buffer.clear()
self.len = 0
def collect(self) -> Dict[str, float]:
def collect(self) -> dict[str, float]:
metrics = {
f'{self.key_prefix}/{key}/{name}': value
for key, stats in self.buffer.items()
@ -820,7 +822,7 @@ class _StatisticsMaintainer: # pylint: disable=missing-class-docstring,missing-
self.max_value = None
self.has_nan = False
def add(self, value: float, timestamp: Optional[float] = None) -> None:
def add(self, value: float, timestamp: float | None = None) -> None:
if timestamp is None:
timestamp = timer()
@ -859,7 +861,7 @@ class _StatisticsMaintainer: # pylint: disable=missing-class-docstring,missing-
return math.nan
return self.max_value
def items(self) -> Iterable[Tuple[str, float]]:
def items(self) -> Iterable[tuple[str, float]]:
yield ('mean', self.mean())
yield ('min', self.min())
yield ('max', self.max())

230
nvitop/api/device.py
View file

@ -101,13 +101,15 @@ Examples:
# pylint: disable=too-many-lines
from __future__ import annotations
import contextlib
import multiprocessing as mp
import os
import re
import threading
from collections import OrderedDict
from typing import Any, Callable, Dict, Iterable, List, NamedTuple, Optional, Tuple, Type, Union
from typing import Any, Callable, Iterable, NamedTuple
from cachetools.func import ttl_cache
@ -130,16 +132,16 @@ __all__ = [
class MemoryInfo(NamedTuple): # in bytes # pylint: disable=missing-class-docstring
total: Union[int, NaType]
free: Union[int, NaType]
used: Union[int, NaType]
total: int | NaType
free: int | NaType
used: int | NaType
class ClockInfos(NamedTuple): # in MHz # pylint: disable=missing-class-docstring
graphics: Union[int, NaType]
sm: Union[int, NaType]
memory: Union[int, NaType]
video: Union[int, NaType]
graphics: int | NaType
sm: int | NaType
memory: int | NaType
video: int | NaType
class ClockSpeedInfos(NamedTuple): # pylint: disable=missing-class-docstring
@ -148,10 +150,10 @@ class ClockSpeedInfos(NamedTuple): # pylint: disable=missing-class-docstring
class UtilizationRates(NamedTuple): # in percentage # pylint: disable=missing-class-docstring
gpu: Union[int, NaType]
memory: Union[int, NaType]
encoder: Union[int, NaType]
decoder: Union[int, NaType]
gpu: int | NaType
memory: int | NaType
encoder: int | NaType
decoder: int | NaType
_VALUE_OMITTED = object()
@ -252,7 +254,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return False
@staticmethod
def driver_version() -> Union[str, NaType]:
def driver_version() -> str | NaType:
"""The version of the installed NVIDIA display driver. This is an alphanumeric string.
Command line equivalent:
@ -273,7 +275,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return libnvml.nvmlQuery('nvmlSystemGetDriverVersion')
@staticmethod
def cuda_driver_version() -> Union[str, NaType]:
def cuda_driver_version() -> str | NaType:
"""The maximum CUDA version supported by the NVIDIA display driver. This is an alphanumeric string.
This can be different from the version of the CUDA Runtime. See also :meth:`cuda_runtime_version`.
@ -303,7 +305,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
max_cuda_version = cuda_driver_version
@staticmethod
def cuda_runtime_version() -> Union[str, NaType]:
def cuda_runtime_version() -> str | NaType:
"""The CUDA Runtime version. This is an alphanumeric string.
This can be different from the CUDA driver version. See also :meth:`cuda_driver_version`.
@ -341,14 +343,14 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return libnvml.nvmlQuery('nvmlDeviceGetCount', default=0)
@classmethod
def all(cls) -> List['PhysicalDevice']:
def all(cls) -> list[PhysicalDevice]:
"""Return a list of all physical devices in the system."""
return cls.from_indices()
@classmethod
def from_indices(
cls, indices: Optional[Union[int, Iterable[Union[int, Tuple[int, int]]]]] = None
) -> List[Union['PhysicalDevice', 'MigDevice']]:
cls, indices: int | Iterable[int | tuple[int, int]] | None = None
) -> list[PhysicalDevice | MigDevice]:
"""Return a list of devices of the given indices.
Args:
@ -386,7 +388,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return list(map(cls, indices))
@staticmethod
def from_cuda_visible_devices() -> List['CudaDevice']:
def from_cuda_visible_devices() -> list[CudaDevice]:
"""Return a list of all CUDA visible devices.
The CUDA ordinal will be enumerate from the ``CUDA_VISIBLE_DEVICES`` environment variable.
@ -410,9 +412,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return cuda_devices
@staticmethod
def from_cuda_indices(
cuda_indices: Optional[Union[int, Iterable[int]]] = None
) -> List['CudaDevice']:
def from_cuda_indices(cuda_indices: int | Iterable[int] | None = None) -> list[CudaDevice]:
"""Return a list of CUDA devices of the given CUDA indices.
The CUDA ordinal will be enumerate from the ``CUDA_VISIBLE_DEVICES`` environment variable.
@ -460,8 +460,8 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
@staticmethod
def parse_cuda_visible_devices(
cuda_visible_devices: Optional[str] = _VALUE_OMITTED,
) -> Union[List[int], List[Tuple[int, int]]]:
cuda_visible_devices: str | None = _VALUE_OMITTED,
) -> list[int] | list[tuple[int, int]]:
"""Parse the given ``CUDA_VISIBLE_DEVICES`` value into a list of NVML device indices.
This is a alias of :func:`parse_cuda_visible_devices`.
@ -486,7 +486,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return parse_cuda_visible_devices(cuda_visible_devices)
@staticmethod
def normalize_cuda_visible_devices(cuda_visible_devices: Optional[str] = _VALUE_OMITTED) -> str:
def normalize_cuda_visible_devices(cuda_visible_devices: str | None = _VALUE_OMITTED) -> str:
"""Parse the given ``CUDA_VISIBLE_DEVICES`` value and convert it into a comma-separated string of UUIDs.
This is an alias of :func:`normalize_cuda_visible_devices`.
@ -511,11 +511,11 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
def __new__(
cls,
index: Optional[Union[int, Tuple[int, int], str]] = None,
index: int | tuple[int, int] | str | None = None,
*,
uuid: Optional[str] = None,
bus_id: Optional[str] = None,
) -> 'Device':
uuid: str | None = None,
bus_id: str | None = None,
) -> Device:
"""Create a new instance of Device.
The type of the result is determined by the given argument.
@ -576,10 +576,10 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
def __init__(
self,
index: Optional[Union[int, str]] = None,
index: int | str | None = None,
*,
uuid: Optional[str] = None,
bus_id: Optional[str] = None,
uuid: str | None = None,
bus_id: str | None = None,
) -> None:
"""Initialize the instance created by :meth:`__new__()`.
@ -672,7 +672,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
self._hash = hash(self._ident)
return self._hash
def __getattr__(self, name: str) -> Union[Any, Callable[..., Any]]:
def __getattr__(self, name: str) -> Any | Callable[..., Any]:
"""Get the object attribute.
If the attribute is not defined, make a method from ``pynvml.nvmlDeviceGet<AttributeName>(handle)``.
@ -734,12 +734,12 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
setattr(self, name, attribute)
return attribute
def __reduce__(self) -> Tuple[Type['Device'], Tuple[Union[int, Tuple[int, int]]]]:
def __reduce__(self) -> tuple[type[Device], tuple[int | tuple[int, int]]]:
"""Return state information for pickling."""
return self.__class__, (self._nvml_index,)
@property
def index(self) -> Union[int, Tuple[int, int]]:
def index(self) -> int | tuple[int, int]:
"""The NVML index of the device.
Returns: Union[int, Tuple[int, int]]
@ -748,7 +748,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return self._nvml_index
@property
def nvml_index(self) -> Union[int, Tuple[int, int]]:
def nvml_index(self) -> int | tuple[int, int]:
"""The NVML index of the device.
Returns: Union[int, Tuple[int, int]]
@ -793,7 +793,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return self._cuda_index
def name(self) -> Union[str, NaType]:
def name(self) -> str | NaType:
"""The official product name of the GPU. This is an alphanumeric string. For all products.
Returns: Union[str, NaType]
@ -809,7 +809,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
self._name = libnvml.nvmlQuery('nvmlDeviceGetName', self.handle)
return self._name
def uuid(self) -> Union[str, NaType]:
def uuid(self) -> str | NaType:
"""This value is the globally unique immutable alphanumeric identifier of the GPU.
It does not correspond to any physical label on the board.
@ -827,7 +827,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
self._uuid = libnvml.nvmlQuery('nvmlDeviceGetUUID', self.handle)
return self._uuid
def bus_id(self) -> Union[str, NaType]:
def bus_id(self) -> str | NaType:
"""PCI bus ID as "domain:bus:device.function", in hex.
Returns: Union[str, NaType]
@ -845,7 +845,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
return self._bus_id
def serial(self) -> Union[str, NaType]:
def serial(self) -> str | NaType:
"""This number matches the serial number physically printed on each board.
It is a globally unique immutable alphanumeric value.
@ -874,7 +874,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return MemoryInfo(total=memory_info.total, free=memory_info.free, used=memory_info.used)
return MemoryInfo(total=NA, free=NA, used=NA)
def memory_total(self) -> Union[int, NaType]: # in bytes
def memory_total(self) -> int | NaType: # in bytes
"""Total installed GPU memory in bytes.
Returns: Union[int, NaType]
@ -890,7 +890,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
self._memory_total = self.memory_info().total
return self._memory_total
def memory_used(self) -> Union[int, NaType]: # in bytes
def memory_used(self) -> int | NaType: # in bytes
"""Total memory allocated by active contexts in bytes.
Returns: Union[int, NaType]
@ -904,7 +904,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.memory_info().used
def memory_free(self) -> Union[int, NaType]: # in bytes
def memory_free(self) -> int | NaType: # in bytes
"""Total free memory in bytes.
Returns: Union[int, NaType]
@ -918,7 +918,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.memory_info().free
def memory_total_human(self) -> Union[str, NaType]: # in human readable
def memory_total_human(self) -> str | NaType: # in human readable
"""Total installed GPU memory in human readable format.
Returns: Union[str, NaType]
@ -928,7 +928,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
self._memory_total_human = bytes2human(self.memory_total())
return self._memory_total_human
def memory_used_human(self) -> Union[str, NaType]: # in human readable
def memory_used_human(self) -> str | NaType: # in human readable
"""Total memory allocated by active contexts in human readable format.
Returns: Union[int, NaType]
@ -936,7 +936,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return bytes2human(self.memory_used())
def memory_free_human(self) -> Union[str, NaType]: # in human readable
def memory_free_human(self) -> str | NaType: # in human readable
"""Total free memory in human readable format.
Returns: Union[int, NaType]
@ -944,7 +944,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return bytes2human(self.memory_free())
def memory_percent(self) -> Union[float, NaType]: # in percentage
def memory_percent(self) -> float | NaType: # in percentage
"""The percentage of used memory over total memory (``0 <= p <= 100``).
Returns: Union[float, NaType]
@ -980,7 +980,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
return MemoryInfo(total=NA, free=NA, used=NA)
def bar1_memory_total(self) -> Union[int, NaType]: # in bytes
def bar1_memory_total(self) -> int | NaType: # in bytes
"""Total BAR1 memory in bytes.
Returns: Union[int, NaType]
@ -988,7 +988,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.bar1_memory_info().total
def bar1_memory_used(self) -> Union[int, NaType]: # in bytes
def bar1_memory_used(self) -> int | NaType: # in bytes
"""Total used BAR1 memory in bytes.
Returns: Union[int, NaType]
@ -996,7 +996,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.bar1_memory_info().used
def bar1_memory_free(self) -> Union[int, NaType]: # in bytes
def bar1_memory_free(self) -> int | NaType: # in bytes
"""Total free BAR1 memory in bytes.
Returns: Union[int, NaType]
@ -1004,7 +1004,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.bar1_memory_info().free
def bar1_memory_total_human(self) -> Union[str, NaType]: # in human readable
def bar1_memory_total_human(self) -> str | NaType: # in human readable
"""Total BAR1 memory in human readable format.
Returns: Union[int, NaType]
@ -1012,7 +1012,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return bytes2human(self.bar1_memory_total())
def bar1_memory_used_human(self) -> Union[str, NaType]: # in human readable
def bar1_memory_used_human(self) -> str | NaType: # in human readable
"""Total used BAR1 memory in human readable format.
Returns: Union[int, NaType]
@ -1020,7 +1020,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return bytes2human(self.bar1_memory_used())
def bar1_memory_free_human(self) -> Union[str, NaType]: # in human readable
def bar1_memory_free_human(self) -> str | NaType: # in human readable
"""Total free BAR1 memory in human readable format.
Returns: Union[int, NaType]
@ -1028,7 +1028,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return bytes2human(self.bar1_memory_free())
def bar1_memory_percent(self) -> Union[float, NaType]: # in percentage
def bar1_memory_percent(self) -> float | NaType: # in percentage
"""The percentage of used BAR1 memory over total BAR1 memory (0 <= p <= 100).
Returns: Union[float, NaType]
@ -1073,7 +1073,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return UtilizationRates(gpu=gpu, memory=memory, encoder=encoder, decoder=decoder)
def gpu_utilization(self) -> Union[int, NaType]: # in percentage
def gpu_utilization(self) -> int | NaType: # in percentage
"""Percent of time over the past sample period during which one or more kernels was executing on the GPU.
The sample period may be between 1 second and 1/6 second depending on the product.
@ -1091,7 +1091,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
gpu_percent = gpu_utilization # in percentage
def memory_utilization(self) -> Union[float, NaType]: # in percentage
def memory_utilization(self) -> float | NaType: # in percentage
"""Percent of time over the past sample period during which global (device) memory was being read or written.
The sample period may be between 1 second and 1/6 second depending on the product.
@ -1107,7 +1107,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.utilization_rates().memory
def encoder_utilization(self) -> Union[float, NaType]: # in percentage
def encoder_utilization(self) -> float | NaType: # in percentage
"""The encoder utilization rate in percentage.
Returns: Union[int, NaType]
@ -1115,7 +1115,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.utilization_rates().encoder
def decoder_utilization(self) -> Union[float, NaType]: # in percentage\
def decoder_utilization(self) -> float | NaType: # in percentage\
"""The decoder utilization rate in percentage.
Returns: Union[int, NaType]
@ -1173,7 +1173,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return ClockSpeedInfos(current=self.clock_infos(), max=self.max_clock_infos())
def graphics_clock(self) -> Union[int, NaType]: # in MHz
def graphics_clock(self) -> int | NaType: # in MHz
"""Current frequency of graphics (shader) clock in MHz.
Returns: Union[int, NaType]
@ -1187,7 +1187,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.clock_infos().graphics
def sm_clock(self) -> Union[int, NaType]: # in MHz
def sm_clock(self) -> int | NaType: # in MHz
"""Current frequency of SM (Streaming Multiprocessor) clock in MHz.
Returns: Union[int, NaType]
@ -1201,7 +1201,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.clock_infos().sm
def memory_clock(self) -> Union[int, NaType]: # in MHz
def memory_clock(self) -> int | NaType: # in MHz
"""Current frequency of memory clock in MHz.
Returns: Union[int, NaType]
@ -1215,7 +1215,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.clock_infos().memory
def video_clock(self) -> Union[int, NaType]: # in MHz
def video_clock(self) -> int | NaType: # in MHz
"""Current frequency of video encoder/decoder clock in MHz.
Returns: Union[int, NaType]
@ -1229,7 +1229,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.clock_infos().video
def max_graphics_clock(self) -> Union[int, NaType]: # in MHz
def max_graphics_clock(self) -> int | NaType: # in MHz
"""Maximum frequency of graphics (shader) clock in MHz.
Returns: Union[int, NaType]
@ -1243,7 +1243,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.max_clock_infos().graphics
def max_sm_clock(self) -> Union[int, NaType]: # in MHz
def max_sm_clock(self) -> int | NaType: # in MHz
"""Maximum frequency of SM (Streaming Multiprocessor) clock in MHz.
Returns: Union[int, NaType]
@ -1257,7 +1257,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
""" # pylint: disable=line-too-long
return self.max_clock_infos().sm
def max_memory_clock(self) -> Union[int, NaType]: # in MHz
def max_memory_clock(self) -> int | NaType: # in MHz
"""Maximum frequency of memory clock in MHz.
Returns: Union[int, NaType]
@ -1271,7 +1271,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.max_clock_infos().memory
def max_video_clock(self) -> Union[int, NaType]: # in MHz
def max_video_clock(self) -> int | NaType: # in MHz
"""Maximum frequency of video encoder/decoder clock in MHz.
Returns: Union[int, NaType]
@ -1286,7 +1286,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return self.max_clock_infos().video
@ttl_cache(ttl=5.0)
def fan_speed(self) -> Union[int, NaType]: # in percentage
def fan_speed(self) -> int | NaType: # in percentage
"""The fan speed value is the percent of the product's maximum noise tolerance fan speed that the device's fan is currently intended to run at.
This value may exceed 100% in certain cases. Note: The reported speed is the intended fan
@ -1306,7 +1306,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return libnvml.nvmlQuery('nvmlDeviceGetFanSpeed', self.handle)
@ttl_cache(ttl=5.0)
def temperature(self) -> Union[int, NaType]: # in Celsius
def temperature(self) -> int | NaType: # in Celsius
"""Core GPU temperature in degrees C.
Returns: Union[int, NaType]
@ -1324,7 +1324,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
@memoize_when_activated
@ttl_cache(ttl=5.0)
def power_usage(self) -> Union[int, NaType]: # in milliwatts (mW)
def power_usage(self) -> int | NaType: # in milliwatts (mW)
"""The last measured power draw for the entire board in milliwatts.
Returns: Union[int, NaType]
@ -1342,7 +1342,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
@memoize_when_activated
@ttl_cache(ttl=60.0)
def power_limit(self) -> Union[int, NaType]: # in milliwatts (mW)
def power_limit(self) -> int | NaType: # in milliwatts (mW)
"""The software power limit in milliwatts.
Set by software like nvidia-smi.
@ -1373,7 +1373,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return f'{power_usage} / {power_limit}'
@ttl_cache(ttl=60.0)
def display_active(self) -> Union[str, NaType]:
def display_active(self) -> str | NaType:
"""A flag that indicates whether a display is initialized on the GPU's (e.g. memory is allocated on the device for display).
Display can be active even when no monitor is physically attached. "Enabled" indicates an
@ -1395,7 +1395,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
@ttl_cache(ttl=60.0)
def display_mode(self) -> Union[str, NaType]:
def display_mode(self) -> str | NaType:
"""A flag that indicates whether a physical display (e.g. monitor) is currently connected to any of the GPU's connectors.
"Enabled" indicates an attached display. "Disabled" indicates otherwise.
@ -1416,7 +1416,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
@ttl_cache(ttl=60.0)
def current_driver_model(self) -> Union[str, NaType]:
def current_driver_model(self) -> str | NaType:
"""The driver model currently in use.
Always "N/A" on Linux. On Windows, the TCC (WDM) and WDDM driver models are supported. The
@ -1443,7 +1443,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
driver_model = current_driver_model
@ttl_cache(ttl=60.0)
def persistence_mode(self) -> Union[str, NaType]:
def persistence_mode(self) -> str | NaType:
"""A flag that indicates whether persistence mode is enabled for the GPU. Value is either "Enabled" or "Disabled".
When persistence mode is enabled the NVIDIA driver remains loaded even when no active
@ -1466,7 +1466,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
@ttl_cache(ttl=5.0)
def performance_state(self) -> Union[str, NaType]:
def performance_state(self) -> str | NaType:
"""The current performance state for the GPU. States range from P0 (maximum performance) to P12 (minimum performance).
Returns: Union[str, NaType]
@ -1484,7 +1484,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return performance_state
@ttl_cache(ttl=5.0)
def total_volatile_uncorrected_ecc_errors(self) -> Union[int, NaType]:
def total_volatile_uncorrected_ecc_errors(self) -> int | NaType:
"""Total errors detected across entire chip.
Returns: Union[int, NaType]
@ -1504,7 +1504,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
)
@ttl_cache(ttl=60.0)
def compute_mode(self) -> Union[str, NaType]:
def compute_mode(self) -> str | NaType:
"""The compute mode flag indicates whether individual or multiple compute applications may run on the GPU.
Returns: Union[str, NaType]
@ -1527,7 +1527,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
libnvml.NVML_COMPUTEMODE_EXCLUSIVE_PROCESS: 'Exclusive Process',
}.get(libnvml.nvmlQuery('nvmlDeviceGetComputeMode', self.handle), NA)
def cuda_compute_capability(self) -> Union[Tuple[int, int], NaType]:
def cuda_compute_capability(self) -> tuple[int, int] | NaType:
"""The CUDA compute capability for the device.
Returns: Union[Tuple[int, int], NaType]
@ -1558,7 +1558,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return self._is_mig_device
@ttl_cache(ttl=60.0)
def mig_mode(self) -> Union[str, NaType]:
def mig_mode(self) -> str | NaType:
"""The MIG mode that the GPU is currently operating under.
Returns: Union[str, NaType]
@ -1594,7 +1594,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return 0 # implemented in PhysicalDevice
def mig_devices(self) -> List['MigDevice']:
def mig_devices(self) -> list[MigDevice]:
"""Return a list of children MIG devices of the current device.
This method will return an empty list if the MIG mode is disabled or the device does not
@ -1610,7 +1610,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
"""
return self.is_mig_device() or not self.is_mig_mode_enabled()
def to_leaf_devices(self) -> List[Union['PhysicalDevice', 'MigDevice', 'CudaDevice']]:
def to_leaf_devices(self) -> list[PhysicalDevice | MigDevice | CudaDevice]:
"""Return a list of leaf devices.
Note that a CUDA device is always a leaf device.
@ -1620,7 +1620,7 @@ class Device: # pylint: disable=too-many-instance-attributes,too-many-public-me
return self.mig_devices()
@ttl_cache(ttl=2.0)
def processes(self) -> Dict[int, GpuProcess]:
def processes(self) -> dict[int, GpuProcess]:
"""Return a dictionary of processes running on the GPU.
Returns: Dict[int, GpuProcess]
@ -1808,7 +1808,7 @@ class PhysicalDevice(Device):
)
@ttl_cache(ttl=60.0)
def mig_device(self, mig_index: int) -> 'MigDevice':
def mig_device(self, mig_index: int) -> MigDevice:
"""Return a child MIG device of the given index.
Raises:
@ -1819,7 +1819,7 @@ class PhysicalDevice(Device):
return MigDevice(index=(self.index, mig_index))
@ttl_cache(ttl=60.0)
def mig_devices(self) -> List['MigDevice']:
def mig_devices(self) -> list[MigDevice]:
"""Return a list of children MIG devices of the current device.
This method will return an empty list if the MIG mode is disabled or the device does not
@ -1850,7 +1850,7 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
return len(cls.all())
@classmethod
def all(cls) -> List['MigDevice']:
def all(cls) -> list[MigDevice]:
"""Return a list of MIG devices aggregated over all physical devices."""
mig_devices = []
for device in PhysicalDevice.all():
@ -1859,8 +1859,8 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
@classmethod
def from_indices( # pylint: disable=signature-differs
cls, indices: Iterable[Tuple[int, int]]
) -> List['MigDevice']:
cls, indices: Iterable[tuple[int, int]]
) -> list[MigDevice]:
"""Return a list of MIG devices of the given indices.
Args:
@ -1885,7 +1885,7 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
# pylint: disable-next=super-init-not-called
def __init__(
self, index: Optional[Union[Tuple[int, int], str]] = None, *, uuid: Optional[str] = None
self, index: tuple[int, int] | str | None = None, *, uuid: str | None = None
) -> None:
"""Initialize the instance created by :meth:`__new__()`.
@ -1962,7 +1962,7 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
self._hash = None
@property
def index(self) -> Tuple[int, int]:
def index(self) -> tuple[int, int]:
"""The index of the MIG device. This is a tuple of two integers."""
return self._nvml_index
@ -1981,7 +1981,7 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
"""The parent physical device."""
return self._parent
def gpu_instance_id(self) -> Union[int, NaType]:
def gpu_instance_id(self) -> int | NaType:
"""The gpu instance ID of the MIG device.
Returns: Union[int, NaType]
@ -1995,7 +1995,7 @@ class MigDevice(Device): # pylint: disable=too-many-instance-attributes
self._gpu_instance_id = NA
return self._gpu_instance_id
def compute_instance_id(self) -> Union[int, NaType]:
def compute_instance_id(self) -> int | NaType:
"""The compute instance ID of the MIG device.
Returns: Union[int, NaType]
@ -2105,7 +2105,7 @@ class CudaDevice(Device):
return 0
@classmethod
def all(cls) -> List['CudaDevice']:
def all(cls) -> list[CudaDevice]:
"""All CUDA visible devices.
Note:
@ -2114,9 +2114,7 @@ class CudaDevice(Device):
return cls.from_indices()
@classmethod
def from_indices(
cls, indices: Optional[Union[int, Iterable[int]]] = None
) -> List['CudaDevice']:
def from_indices(cls, indices: int | Iterable[int] | None = None) -> list[CudaDevice]:
"""Return a list of CUDA devices of the given CUDA indices.
The CUDA ordinal will be enumerate from the ``CUDA_VISIBLE_DEVICES`` environment variable.
@ -2147,11 +2145,11 @@ class CudaDevice(Device):
def __new__(
cls,
cuda_index: Optional[int] = None,
cuda_index: int | None = None,
*,
nvml_index: Optional[Union[int, Tuple[int, int]]] = None,
uuid: Optional[str] = None,
) -> 'Device':
nvml_index: int | tuple[int, int] | None = None,
uuid: str | None = None,
) -> Device:
"""Create a new instance of CudaDevice.
The type of the result is determined by the given argument.
@ -2189,10 +2187,10 @@ class CudaDevice(Device):
def __init__(
self,
cuda_index: Optional[int] = None,
cuda_index: int | None = None,
*,
nvml_index: Optional[Union[int, Tuple[int, int]]] = None,
uuid: Optional[str] = None,
nvml_index: int | tuple[int, int] | None = None,
uuid: str | None = None,
) -> None:
"""Initialize the instance created by :meth:`__new__()`.
@ -2238,7 +2236,7 @@ class CudaDevice(Device):
__repr__ = __str__
def __reduce__(self) -> Tuple[Type['CudaDevice'], Tuple[int]]:
def __reduce__(self) -> tuple[type[CudaDevice], tuple[int]]:
"""Return state information for pickling."""
return self.__class__, (self._cuda_index,)
@ -2261,7 +2259,7 @@ class CudaMigDevice(CudaDevice, MigDevice):
"""Class for CUDA devices that are MIG devices."""
def is_mig_device_uuid(uuid: Optional[str]) -> bool:
def is_mig_device_uuid(uuid: str | None) -> bool:
"""Return :data:`True` if the argument is a MIG device UUID, otherwise, return :data:`False`."""
if isinstance(uuid, str):
match = Device.UUID_PATTERN.match(uuid)
@ -2271,8 +2269,8 @@ def is_mig_device_uuid(uuid: Optional[str]) -> bool:
def parse_cuda_visible_devices(
cuda_visible_devices: Optional[str] = _VALUE_OMITTED,
) -> Union[List[int], List[Tuple[int, int]]]:
cuda_visible_devices: str | None = _VALUE_OMITTED,
) -> list[int] | list[tuple[int, int]]:
"""Parse the given ``CUDA_VISIBLE_DEVICES`` value into a list of NVML device indices.
This function is aliased by :meth:`Device.parse_cuda_visible_devices`.
@ -2329,7 +2327,7 @@ def parse_cuda_visible_devices(
return _parse_cuda_visible_devices(cuda_visible_devices, format='index')
def normalize_cuda_visible_devices(cuda_visible_devices: Optional[str] = _VALUE_OMITTED) -> str:
def normalize_cuda_visible_devices(cuda_visible_devices: str | None = _VALUE_OMITTED) -> str:
"""Parse the given ``CUDA_VISIBLE_DEVICES`` value and convert it into a comma-separated string of UUIDs.
This function is aliased by :meth:`Device.normalize_cuda_visible_devices`.
@ -2400,7 +2398,7 @@ _GLOBAL_PHYSICAL_DEVICE = None
_GLOBAL_PHYSICAL_DEVICE_LOCK = threading.RLock()
def _get_all_physical_device_attrs() -> Dict[str, _PhysicalDeviceAttrs]:
def _get_all_physical_device_attrs() -> dict[str, _PhysicalDeviceAttrs]:
global _PHYSICAL_DEVICE_ATTRS # pylint: disable=global-statement
with _GLOBAL_PHYSICAL_DEVICE_LOCK:
@ -2422,14 +2420,14 @@ def _get_all_physical_device_attrs() -> Dict[str, _PhysicalDeviceAttrs]:
return _PHYSICAL_DEVICE_ATTRS
def _does_any_device_support_mig_mode(uuids: Optional[Iterable[str]] = None) -> bool:
def _does_any_device_support_mig_mode(uuids: Iterable[str] | None = None) -> bool:
physical_device_attrs = _get_all_physical_device_attrs()
uuids = uuids or physical_device_attrs.keys()
return any(physical_device_attrs[uuid].support_mig_mode for uuid in uuids)
@contextlib.contextmanager
def _global_physical_device(device: 'PhysicalDevice') -> 'PhysicalDevice':
def _global_physical_device(device: PhysicalDevice) -> PhysicalDevice:
global _GLOBAL_PHYSICAL_DEVICE # pylint: disable=global-statement
with _GLOBAL_PHYSICAL_DEVICE_LOCK:
@ -2440,16 +2438,16 @@ def _global_physical_device(device: 'PhysicalDevice') -> 'PhysicalDevice':
_GLOBAL_PHYSICAL_DEVICE = None
def _get_global_physical_device() -> 'PhysicalDevice':
def _get_global_physical_device() -> PhysicalDevice:
with _GLOBAL_PHYSICAL_DEVICE_LOCK:
return _GLOBAL_PHYSICAL_DEVICE
@ttl_cache(ttl=300.0)
def _parse_cuda_visible_devices( # pylint: disable=too-many-branches,too-many-statements
cuda_visible_devices: Optional[str] = None,
cuda_visible_devices: str | None = None,
format: str = 'index', # pylint: disable=redefined-builtin
) -> Union[List[int], List[Tuple[int, int]], List[str]]:
) -> list[int] | list[tuple[int, int]] | list[str]:
"""The underlining implementation for :meth:`parse_cuda_visible_devices`. The result will be cached."""
assert format in ('index', 'uuid')
@ -2477,7 +2475,7 @@ def _parse_cuda_visible_devices( # pylint: disable=too-many-branches,too-many-s
if cuda_visible_devices is None:
cuda_visible_devices = ','.join(physical_device_attrs.keys())
def from_index_or_uuid(index_or_uuid: Union[int, str]) -> 'Device':
def from_index_or_uuid(index_or_uuid: int | str) -> Device:
nonlocal use_integer_identifiers
if isinstance(index_or_uuid, str):
@ -2549,9 +2547,9 @@ def _parse_cuda_visible_devices( # pylint: disable=too-many-branches,too-many-s
def _parse_cuda_visible_devices_to_uuids(
cuda_visible_devices: Optional[str] = _VALUE_OMITTED,
cuda_visible_devices: str | None = _VALUE_OMITTED,
verbose: bool = True,
) -> List[str]:
) -> list[str]:
"""Parse the given ``CUDA_VISIBLE_DEVICES`` environment variable in a separate process and return a list of device UUIDs.
The UUIDs do not have a prefix ``GPU-`` or ``MIG-``.

10
nvitop/api/libcuda.py
View file

@ -18,6 +18,8 @@
# pylint: disable=invalid-name
from __future__ import annotations
import ctypes as _ctypes
import itertools as _itertools
import platform as _platform
@ -25,8 +27,6 @@ import string as _string
import sys as _sys
import threading as _threading
from typing import Any as _Any
from typing import Tuple as _Tuple
from typing import Type as _Type
# pylint: disable-next=missing-class-docstring,too-few-public-methods
@ -229,7 +229,7 @@ class CUDAError(Exception):
} # fmt:skip
_errcode_to_name = {}
def __new__(cls, value: int) -> 'CUDAError':
def __new__(cls, value: int) -> CUDAError:
"""Map value to a proper subclass of :class:`CUDAError`."""
if cls is CUDAError:
# pylint: disable-next=self-cls-assignment
@ -264,12 +264,12 @@ class CUDAError(Exception):
return NotImplemented
return self.value == other.value # pylint: disable=no-member
def __reduce__(self) -> _Tuple[_Type['CUDAError'], _Tuple[int]]:
def __reduce__(self) -> tuple[type[CUDAError], tuple[int]]:
"""Return state information for pickling."""
return CUDAError, (self.value,) # pylint: disable=no-member
def cudaExceptionClass(cudaErrorCode: int) -> _Type[CUDAError]:
def cudaExceptionClass(cudaErrorCode: int) -> type[CUDAError]:
"""Map value to a proper subclass of :class:`CUDAError`.
Raises:

10
nvitop/api/libcudart.py
View file

@ -18,6 +18,8 @@
# pylint: disable=invalid-name
from __future__ import annotations
import ctypes as _ctypes
import glob as _glob
import os as _os
@ -25,8 +27,6 @@ import platform as _platform
import sys as _sys
import threading as _threading
from typing import Any as _Any
from typing import Tuple as _Tuple
from typing import Type as _Type
_cudaError_t = _ctypes.c_int
@ -280,7 +280,7 @@ class cudaError(Exception):
} # fmt:skip
_errcode_to_name = {}
def __new__(cls, value: int) -> 'cudaError':
def __new__(cls, value: int) -> cudaError:
"""Map value to a proper subclass of :class:`cudaError`."""
if cls is cudaError:
# pylint: disable-next=self-cls-assignment
@ -315,12 +315,12 @@ class cudaError(Exception):
return NotImplemented
return self.value == other.value # pylint: disable=no-member
def __reduce__(self) -> _Tuple[_Type['cudaError'], _Tuple[int]]:
def __reduce__(self) -> tuple[type[cudaError], tuple[int]]:
"""Return state information for pickling."""
return cudaError, (self.value,) # pylint: disable=no-member
def cudaExceptionClass(cudaErrorCode: int) -> _Type[cudaError]:
def cudaExceptionClass(cudaErrorCode: int) -> type[cudaError]:
"""Map value to a proper subclass of :class:`cudaError`.
Raises:

22
nvitop/api/libnvml.py
View file

@ -18,6 +18,8 @@
# pylint: disable=invalid-name
from __future__ import annotations
import ctypes as _ctypes
import functools as _functools
import inspect as _inspect
@ -26,15 +28,10 @@ import os as _os
import re as _re
import sys as _sys
import threading as _threading
from collections import OrderedDict as _OrderedDict
from types import FunctionType as _FunctionType
from types import ModuleType as _ModuleType
from typing import Any as _Any
from typing import Callable as _Callable
from typing import Optional as _Optional
from typing import Tuple as _Tuple
from typing import Type as _Type
from typing import Union as _Union
# Python Bindings for the NVIDIA Management Library (NVML)
# https://pypi.org/project/nvidia-ml-py
@ -334,7 +331,7 @@ def nvmlShutdown() -> None: # pylint: disable=function-redefined
def nvmlQuery(
func: _Union[_Callable[..., _Any], str],
func: _Callable[..., _Any] | str,
*args,
default: _Any = NA,
ignore_errors: bool = True,
@ -419,9 +416,7 @@ def nvmlQuery(
return retval
def nvmlCheckReturn(
retval: _Any, types: _Optional[_Union[_Type, _Tuple[_Type, ...]]] = None
) -> bool:
def nvmlCheckReturn(retval: _Any, types: type | tuple[type, ...] | None = None) -> bool:
"""Check whether the return value is not :const:`nvitop.NA` and is one of the given types."""
if types is None:
return retval != NA
@ -690,14 +685,14 @@ class _CustomModule(_ModuleType):
... # The NVML context has been shutdown
"""
def __getattribute__(self, name: str) -> _Union[_Any, _Callable[..., _Any]]:
def __getattribute__(self, name: str) -> _Any | _Callable[..., _Any]:
"""Get a member from the current module. Fallback to the original package if missing."""
try:
return super().__getattribute__(name)
except AttributeError:
return getattr(_pynvml, name)
def __enter__(self) -> '_CustomModule': # noqa: F405
def __enter__(self) -> _CustomModule: # noqa: F405
"""Entry of the context manager for ``with`` statement."""
_lazy_init()
return self
@ -718,8 +713,3 @@ class _CustomModule(_ModuleType):
__modself = _sys.modules[__name__]
__modself.__class__ = _CustomModule
del _CustomModule
# Delete imported references
del _logging, _os, _re, _sys, _threading
del _OrderedDict, _FunctionType, _ModuleType
del _Tuple, _Callable, _Type, _Union, _Optional, _Any

110
nvitop/api/process.py
View file

@ -18,6 +18,8 @@
# pylint: disable=too-many-lines
from __future__ import annotations
import contextlib
import datetime
import functools
@ -25,7 +27,7 @@ import os
import threading
from abc import ABCMeta
from types import FunctionType
from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional, Tuple, Type, Union
from typing import TYPE_CHECKING, Any, Callable, Iterable
from weakref import WeakValueDictionary
from nvitop.api import host, libnvml
@ -85,7 +87,7 @@ else:
return '"{}"'.format(s.replace('\n', r'\n'))
def command_join(cmdline: List[str]) -> str:
def command_join(cmdline: list[str]) -> str:
"""Return a shell-escaped string from command line arguments."""
if len(cmdline) == 1 and not (
# May be modified by `setproctitle`
@ -111,7 +113,7 @@ def auto_garbage_clean(
def wrapper(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
def wrapped(self: 'GpuProcess', *args, **kwargs):
def wrapped(self: GpuProcess, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except host.PsutilError as ex:
@ -183,7 +185,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
INSTANCE_LOCK = threading.RLock()
INSTANCES = WeakValueDictionary()
def __new__(cls, pid: Optional[int] = None) -> 'HostProcess':
def __new__(cls, pid: int | None = None) -> HostProcess:
"""Return the cached instance of :class:`HostProcess`."""
if pid is None:
pid = os.getpid()
@ -211,7 +213,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
return instance
# pylint: disable-next=unused-argument,super-init-not-called
def __init__(self, pid: Optional[int] = None) -> None:
def __init__(self, pid: int | None = None) -> None:
"""Initialize the instance."""
@property
@ -234,7 +236,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
__repr__ = __str__
def __reduce__(self) -> Tuple[Type['HostProcess'], Tuple[int]]:
def __reduce__(self) -> tuple[type[HostProcess], tuple[int]]:
"""Return state information for pickling."""
return self.__class__, (self.pid,)
@ -277,7 +279,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
return self._username
@memoize_when_activated
def cmdline(self) -> List[str]:
def cmdline(self) -> list[str]:
"""The command line this process has been called with.
Raises:
@ -351,7 +353,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
"""
return self.memory_info().rss
def parent(self) -> Optional['HostProcess']:
def parent(self) -> HostProcess | None:
"""Return the parent process as a :class:`HostProcess` instance or :data:`None` if there is no parent.
Raises:
@ -365,7 +367,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
return HostProcess(parent.pid)
return None
def children(self, recursive: bool = False) -> List['HostProcess']:
def children(self, recursive: bool = False) -> list[HostProcess]:
"""Return the children of this process as a list of :class:`HostProcess` instances.
If *recursive* is :data:`True` return all the descendants.
@ -414,7 +416,7 @@ class HostProcess(host.Process, metaclass=ABCMeta):
self.running_time.cache_deactivate(self)
def as_snapshot(
self, attrs: Optional[Iterable[str]] = None, ad_value: Optional[Any] = None
self, attrs: Iterable[str] | None = None, ad_value: Any | None = None
) -> Snapshot:
"""Return a onetime snapshot of the process."""
with self.oneshot():
@ -447,14 +449,14 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
def __new__(
cls,
pid: int,
device: 'Device',
device: Device,
# pylint: disable=unused-argument
gpu_memory: Optional[Union[int, NaType]] = None,
gpu_instance_id: Optional[Union[int, NaType]] = None,
compute_instance_id: Optional[Union[int, NaType]] = None,
type: Optional[Union[str, NaType]] = None, # pylint: disable=redefined-builtin
gpu_memory: int | NaType | None = None,
gpu_instance_id: int | NaType | None = None,
compute_instance_id: int | NaType | None = None,
type: str | NaType | None = None, # pylint: disable=redefined-builtin
# pylint: enable=unused-argument
) -> 'GpuProcess':
) -> GpuProcess:
"""Return the cached instance of :class:`GpuProcess`."""
if pid is None:
pid = os.getpid()
@ -485,11 +487,11 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
def __init__(
self,
pid: int, # pylint: disable=unused-argument
device: 'Device',
gpu_memory: Optional[Union[int, NaType]] = None,
gpu_instance_id: Optional[Union[int, NaType]] = None,
compute_instance_id: Optional[Union[int, NaType]] = None,
type: Optional[Union[str, NaType]] = None, # pylint: disable=redefined-builtin
device: Device,
gpu_memory: int | NaType | None = None,
gpu_instance_id: int | NaType | None = None,
compute_instance_id: int | NaType | None = None,
type: str | NaType | None = None, # pylint: disable=redefined-builtin
) -> None:
"""Initialize the instance returned by :meth:`__new__()`."""
if gpu_memory is None and not hasattr(self, '_gpu_memory'):
@ -542,7 +544,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
self._hash = hash(self._ident) # pylint: disable=attribute-defined-outside-init
return self._hash
def __getattr__(self, name: str) -> Union[Any, Callable[..., Any]]:
def __getattr__(self, name: str) -> Any | Callable[..., Any]:
"""Get a member from the instance or fallback to the host process instance if missing.
Raises:
@ -576,7 +578,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self._host
@property
def device(self) -> 'Device':
def device(self) -> Device:
"""The GPU device the process running on.
The same host process can use multiple GPU devices. The :class:`GpuProcess` instances
@ -584,43 +586,43 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
"""
return self._device
def gpu_instance_id(self) -> Union[int, NaType]:
def gpu_instance_id(self) -> int | NaType:
"""The GPU instance ID of the MIG device, or :const:`nvitop.NA` if not applicable."""
return self._gpu_instance_id
def compute_instance_id(self) -> Union[int, NaType]:
def compute_instance_id(self) -> int | NaType:
"""The compute instance ID of the MIG device, or :const:`nvitop.NA` if not applicable."""
return self._compute_instance_id
def gpu_memory(self) -> Union[int, NaType]: # in bytes
def gpu_memory(self) -> int | NaType: # in bytes
"""The used GPU memory in bytes, or :const:`nvitop.NA` if not applicable."""
return self._gpu_memory
def gpu_memory_human(self) -> Union[str, NaType]: # in human readable
def gpu_memory_human(self) -> str | NaType: # in human readable
"""The used GPU memory in human readable format, or :const:`nvitop.NA` if not applicable."""
return self._gpu_memory_human
def gpu_memory_percent(self) -> Union[float, NaType]: # in percentage
def gpu_memory_percent(self) -> float | NaType: # in percentage
"""The percentage of used GPU memory by the process, or :const:`nvitop.NA` if not applicable."""
return self._gpu_memory_percent
def gpu_sm_utilization(self) -> Union[int, NaType]: # in percentage
def gpu_sm_utilization(self) -> int | NaType: # in percentage
"""The utilization rate of SM (Streaming Multiprocessor), or :const:`nvitop.NA` if not applicable."""
return self._gpu_sm_utilization
def gpu_memory_utilization(self) -> Union[int, NaType]: # in percentage
def gpu_memory_utilization(self) -> int | NaType: # in percentage
"""The utilization rate of GPU memory bandwidth, or :const:`nvitop.NA` if not applicable."""
return self._gpu_memory_utilization
def gpu_encoder_utilization(self) -> Union[int, NaType]: # in percentage
def gpu_encoder_utilization(self) -> int | NaType: # in percentage
"""The utilization rate of the encoder, or :const:`nvitop.NA` if not applicable."""
return self._gpu_encoder_utilization
def gpu_decoder_utilization(self) -> Union[int, NaType]: # in percentage
def gpu_decoder_utilization(self) -> int | NaType: # in percentage
"""The utilization rate of the decoder, or :const:`nvitop.NA` if not applicable."""
return self._gpu_decoder_utilization
def set_gpu_memory(self, value: Union[int, NaType]) -> None:
def set_gpu_memory(self, value: int | NaType) -> None:
"""Set the used GPU memory in bytes."""
# pylint: disable=attribute-defined-outside-init
self._gpu_memory = memory_used = value
@ -633,10 +635,10 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
def set_gpu_utilization(
self,
gpu_sm_utilization: Optional[int] = None,
gpu_memory_utilization: Optional[int] = None,
gpu_encoder_utilization: Optional[int] = None,
gpu_decoder_utilization: Optional[int] = None,
gpu_sm_utilization: int | None = None,
gpu_memory_utilization: int | None = None,
gpu_encoder_utilization: int | None = None,
gpu_decoder_utilization: int | None = None,
) -> None:
"""Set the GPU utilization rates."""
# pylint: disable=attribute-defined-outside-init
@ -649,7 +651,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
if gpu_decoder_utilization is not None:
self._gpu_decoder_utilization = gpu_decoder_utilization
def update_gpu_status(self) -> Union[int, NaType]:
def update_gpu_status(self) -> int | NaType:
"""Update the GPU consumption status from a new NVML query."""
self.set_gpu_memory(NA)
self.set_gpu_utilization(NA, NA, NA, NA)
@ -658,7 +660,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self.gpu_memory()
@property
def type(self) -> Union[str, NaType]:
def type(self) -> str | NaType:
"""The type of the GPU context.
The type is one of the following:
@ -670,7 +672,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self._type
@type.setter
def type(self, value: Union[str, NaType]) -> None:
def type(self, value: str | NaType) -> None:
if 'C' in value and 'G' in value:
self._type = 'C+G'
elif 'C' in value:
@ -702,7 +704,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self.host.status()
@auto_garbage_clean(fallback=NA)
def create_time(self) -> Union[float, NaType]:
def create_time(self) -> float | NaType:
"""The process creation time as a floating point number expressed in seconds since the epoch.
Raises:
@ -718,7 +720,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self.host.create_time()
@auto_garbage_clean(fallback=NA)
def running_time(self) -> Union[datetime.timedelta, NaType]:
def running_time(self) -> datetime.timedelta | NaType:
"""The elapsed time this process has been running in :class:`datetime.timedelta`.
Raises:
@ -733,7 +735,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
"""
return self.host.running_time()
def running_time_human(self) -> Union[str, NaType]:
def running_time_human(self) -> str | NaType:
"""The elapsed time this process has been running in human readable format.
Raises:
@ -748,7 +750,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
"""
return timedelta2human(self.running_time())
def running_time_in_seconds(self) -> Union[float, NaType]:
def running_time_in_seconds(self) -> float | NaType:
"""The elapsed time this process has been running in seconds.
Raises:
@ -771,7 +773,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
elapsed_time_in_seconds = running_time_in_seconds
@auto_garbage_clean(fallback=NA)
def username(self) -> Union[str, NaType]:
def username(self) -> str | NaType:
"""The name of the user that owns the process.
Raises:
@ -789,7 +791,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self._username
@auto_garbage_clean(fallback=NA)
def name(self) -> Union[str, NaType]:
def name(self) -> str | NaType:
"""The process name.
Raises:
@ -805,7 +807,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self.host.name()
@auto_garbage_clean(fallback=NA)
def cpu_percent(self) -> Union[float, NaType]: # in percentage
def cpu_percent(self) -> float | NaType: # in percentage
"""Return a float representing the current process CPU utilization as a percentage.
Raises:
@ -821,7 +823,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
return self.host.cpu_percent()
@auto_garbage_clean(fallback=NA)
def memory_percent(self) -> Union[float, NaType]: # in percentage
def memory_percent(self) -> float | NaType: # in percentage
"""Compare process RSS memory to total physical system memory and calculate process memory utilization as a percentage.
Raises:
@ -839,7 +841,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
host_memory_percent = memory_percent # in percentage
@auto_garbage_clean(fallback=NA)
def host_memory(self) -> Union[int, NaType]: # in bytes
def host_memory(self) -> int | NaType: # in bytes
"""The used resident set size (RSS) memory of the process in bytes.
Raises:
@ -854,7 +856,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
"""
return self.host.rss_memory()
def host_memory_human(self) -> Union[str, NaType]:
def host_memory_human(self) -> str | NaType:
"""The used resident set size (RSS) memory of the process in human readable format.
Raises:
@ -873,7 +875,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
# For `AccessDenied` error the fallback value is `['No Permissions']`
@auto_garbage_clean(fallback=('No Such Process',))
def cmdline(self) -> List[str]:
def cmdline(self) -> list[str]:
"""The command line this process has been called with.
Raises:
@ -931,7 +933,7 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
@auto_garbage_clean(fallback=_RAISE)
def as_snapshot(
self, *, host_process_snapshot_cache: Optional[Dict[int, Snapshot]] = None
self, *, host_process_snapshot_cache: dict[int, Snapshot] | None = None
) -> Snapshot:
"""Return a onetime snapshot of the process on the GPU device.
@ -981,8 +983,8 @@ class GpuProcess: # pylint: disable=too-many-instance-attributes,too-many-publi
@classmethod
def take_snapshots( # batched version of `as_snapshot`
cls, gpu_processes: Iterable['GpuProcess'], *, failsafe: bool = False
) -> List[Snapshot]:
cls, gpu_processes: Iterable[GpuProcess], *, failsafe: bool = False
) -> list[Snapshot]:
"""Take snapshots for a list of :class:`GpuProcess` instances.
If *failsafe* is :data:`True`, then if any method fails, the fallback value in

32
nvitop/api/utils.py
View file

@ -18,6 +18,8 @@
# pylint: disable=invalid-name
from __future__ import annotations
import datetime
import functools
import math
@ -25,7 +27,7 @@ import os
import re
import sys
import time
from typing import Any, Callable, Iterable, Optional, Tuple, Union
from typing import Any, Callable, Iterable
from psutil import WINDOWS
@ -66,8 +68,8 @@ except ImportError:
def _colored( # pylint: disable=unused-argument
text: str,
color: Optional[str] = None,
on_color: Optional[str] = None,
color: str | None = None,
on_color: str | None = None,
attrs: Iterable[str] = None,
) -> str:
return text
@ -90,8 +92,8 @@ def set_color(value: bool) -> None:
def colored(
text: str,
color: Optional[str] = None,
on_color: Optional[str] = None,
color: str | None = None,
on_color: str | None = None,
attrs: Iterable[str] = None,
) -> str:
"""Colorize text with ANSI color escape codes.
@ -144,7 +146,7 @@ class NaType(str):
nan
"""
def __new__(cls) -> 'NaType':
def __new__(cls) -> NaType:
"""Get the singleton instance (:const:`nvitop.NA`)."""
if not hasattr(cls, '_instance'):
cls._instance = super().__new__(cls, 'N/A')
@ -176,7 +178,7 @@ class NaType(str):
"""
return math.nan
def __add__(self, other: object) -> Union[str, float]:
def __add__(self, other: object) -> str | float:
"""Return :data:`math.nan` if the operand is a number or uses string concatenation if the operand is a string (``NA + other``).
A special case is when the operand is :const:`nvitop.NA` itself, the result is
@ -195,7 +197,7 @@ class NaType(str):
return float(self) + other
return super().__add__(other)
def __radd__(self, other: object) -> Union[str, float]:
def __radd__(self, other: object) -> str | float:
"""Return :data:`math.nan` if the operand is a number or uses string concatenation if the operand is a string (``other + NA``).
>>> 'str' + NA
@ -351,7 +353,7 @@ class NaType(str):
return other % float(self)
return NotImplemented
def __divmod__(self, other: object) -> Tuple[float, float]:
def __divmod__(self, other: object) -> tuple[float, float]:
"""The pair ``(NA // other, NA % other)`` (``divmod(NA, other)``).
>>> divmod(NA, 1024)
@ -365,7 +367,7 @@ class NaType(str):
"""
return (self // other, self % other)
def __rdivmod__(self, other: object) -> Tuple[float, float]:
def __rdivmod__(self, other: object) -> tuple[float, float]:
"""The pair ``(other // NA, other % NA)`` (``divmod(other, NA)``).
>>> divmod(1024, NA)
@ -399,7 +401,7 @@ class NaType(str):
"""
return abs(float(self))
def __round__(self, ndigits: Optional[int] = None) -> Union[int, float]:
def __round__(self, ndigits: int | None = None) -> int | float:
"""Round :const:`nvitop.NA` to ``ndigits`` decimal places, defaulting to :const:`0`.
If ``ndigits`` is omitted or :data:`None`, returns :const:`0`, otherwise returns :data:`math.nan`.
@ -494,7 +496,7 @@ SIZE_PATTERN = re.compile(
"""The regex pattern for human readable size."""
def bytes2human(b: Union[int, float, NaType]) -> str: # pylint: disable=too-many-return-statements
def bytes2human(b: int | float | NaType) -> str: # pylint: disable=too-many-return-statements
"""Convert bytes to a human readable string."""
if b == NA:
return NA
@ -524,7 +526,7 @@ def bytes2human(b: Union[int, float, NaType]) -> str: # pylint: disable=too-man
return f'{round(b / PiB, 1):.1f}PiB'
def human2bytes(s: Union[int, str]) -> int:
def human2bytes(s: int | str) -> int:
"""Convert a human readable size string (*case insensitive*) to bytes.
Raises:
@ -558,7 +560,7 @@ def human2bytes(s: Union[int, str]) -> int:
return int(float(size) * SIZE_UNITS[unit])
def timedelta2human(dt: Union[int, float, datetime.timedelta, NaType]) -> str:
def timedelta2human(dt: int | float | datetime.timedelta | NaType) -> str:
"""Convert a number in seconds or a :class:`datetime.timedelta` instance to a human readable string."""
if isinstance(dt, (int, float)):
dt = datetime.timedelta(seconds=dt)
@ -575,7 +577,7 @@ def timedelta2human(dt: Union[int, float, datetime.timedelta, NaType]) -> str:
return '{:d}:{:02d}'.format(*divmod(seconds, 60))
def utilization2string(utilization: Union[int, float, NaType]) -> str:
def utilization2string(utilization: int | float | NaType) -> str:
"""Convert a utilization rate to string."""
if utilization != NA:
if isinstance(utilization, int):

7
nvitop/callbacks/keras.py
View file

@ -18,9 +18,10 @@
# pylint: disable=missing-module-docstring,missing-function-docstring
# pylint: disable=unused-argument,attribute-defined-outside-init
from __future__ import annotations
import re
import time
from typing import Dict, List, Tuple, Union
from tensorflow.python.keras.callbacks import ( # pylint: disable=import-error,no-name-in-module
Callback,
@ -94,7 +95,7 @@ class GpuStatsLogger(Callback): # pylint: disable=too-many-instance-attributes
def __init__( # pylint: disable=too-many-arguments
self,
gpus: Union[int, Union[List[Union[int, str]], Tuple[Union[int, str], ...]]],
gpus: int | list[int | str] | tuple[int | str, ...],
memory_utilization: bool = True,
gpu_utilization: bool = True,
intra_step_time: bool = False,
@ -166,7 +167,7 @@ class GpuStatsLogger(Callback): # pylint: disable=too-many-instance-attributes
time.monotonic() - self._snap_intra_step_time
)
def _get_gpu_stats(self) -> Dict[str, float]:
def _get_gpu_stats(self) -> dict[str, float]:
"""Get the gpu status from NVML queries."""
return get_gpu_stats(
devices=self._devices,

5
nvitop/callbacks/pytorch_lightning.py
View file

@ -18,8 +18,9 @@
# pylint: disable=missing-module-docstring,missing-function-docstring
# pylint: disable=unused-argument,attribute-defined-outside-init
from __future__ import annotations
import time
from typing import Dict
from pytorch_lightning.callbacks import Callback # pylint: disable=import-error
from pytorch_lightning.utilities import rank_zero_only # pylint: disable=import-error
@ -161,7 +162,7 @@ class GpuStatsLogger(Callback): # pylint: disable=too-many-instance-attributes
trainer.logger.log_metrics(logs, step=trainer.global_step)
def _get_gpu_stats(self) -> Dict[str, float]:
def _get_gpu_stats(self) -> dict[str, float]:
"""Get the gpu status from NVML queries."""
return get_gpu_stats(
devices=self._devices,

12
nvitop/callbacks/tensorboard.py
View file

@ -17,7 +17,9 @@
# pylint: disable=missing-module-docstring
from typing import TYPE_CHECKING, Dict, Optional, Union
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@ -33,11 +35,11 @@ if TYPE_CHECKING:
def add_scalar_dict(
writer: 'SummaryWriter',
writer: SummaryWriter,
main_tag: str,
tag_scalar_dict: Dict[str, Union[int, float, 'np.floating']],
global_step: Optional[Union[int, 'np.integer']] = None,
walltime: Optional[float] = None,
tag_scalar_dict: dict[str, int | float | np.floating],
global_step: int | np.integer | None = None,
walltime: float | None = None,
) -> None:
"""Add a batch of scalars to the writer.

8
nvitop/callbacks/utils.py
View file

@ -17,12 +17,12 @@
# pylint: disable=missing-module-docstring,missing-function-docstring
from typing import Dict, List
from __future__ import annotations
from nvitop.api import CudaDevice, Device, MiB
def get_devices_by_logical_ids(device_ids: List[int], unique: bool = True) -> List[CudaDevice]:
def get_devices_by_logical_ids(device_ids: list[int], unique: bool = True) -> list[CudaDevice]:
cuda_devices = CudaDevice.from_indices(device_ids)
devices = []
@ -37,12 +37,12 @@ def get_devices_by_logical_ids(device_ids: List[int], unique: bool = True) -> Li
def get_gpu_stats(
devices: List[Device],
devices: list[Device],
memory_utilization: bool = True,
gpu_utilization: bool = True,
fan_speed: bool = False,
temperature: bool = False,
) -> Dict[str, float]:
) -> dict[str, float]:
"""Get the GPU status from NVML queries."""
stats = {}
for device in devices:

18
nvitop/select.py
View file

@ -54,13 +54,15 @@ Python API:
)
""" # pylint: disable=line-too-long
from __future__ import annotations
import argparse
import getpass
import math
import os
import sys
import warnings
from typing import Any, Iterable, List, Optional, Tuple, Union
from typing import Any, Iterable
from nvitop.api import Device, GpuProcess, colored, human2bytes, libnvml
from nvitop.version import __version__
@ -82,16 +84,16 @@ def select_devices( # pylint: disable=too-many-branches,too-many-statements,too
format: str = 'index', # pylint: disable=redefined-builtin
force_index: bool = False,
min_count: int = 0,
max_count: Optional[int] = None,
min_free_memory: Optional[Union[int, str]] = None, # in bytes or human readable
min_total_memory: Optional[Union[int, str]] = None, # in bytes or human readable
max_gpu_utilization: Optional[int] = None, # in percentage
max_memory_utilization: Optional[int] = None, # in percentage
max_count: int | None = None,
min_free_memory: int | str | None = None, # in bytes or human readable
min_total_memory: int | str | None = None, # in bytes or human readable
max_gpu_utilization: int | None = None, # in percentage
max_memory_utilization: int | None = None, # in percentage
tolerance: int = 0, # in percentage
free_accounts: List[str] = None,
free_accounts: list[str] = None,
sort: bool = True,
**kwargs: Any,
) -> Union[List[int], List[Tuple[int, int]], List[str]]:
) -> list[int] | list[tuple[int, int]] | list[str]:
"""Select a subset of devices satisfying the specified criteria.
Note:

2
nvitop/version.py
View file

@ -32,7 +32,7 @@ if not __release__:
['git', 'describe', '--abbrev=7'],
cwd=os.path.dirname(os.path.abspath(__file__)),
stderr=subprocess.DEVNULL,
universal_newlines=True,
text=True,
)
.strip()
.lstrip('v')

5
pyproject.toml
View file

@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
name = "nvitop"
description = "An interactive NVIDIA-GPU process viewer and beyond, the one-stop solution for GPU process management."
readme = "README.md"
requires-python = ">= 3.6"
requires-python = ">= 3.7"
authors = [{ name = "Xuehai Pan", email = "XuehaiPan@pku.edu.cn" }]
license = { text = "Apache License, Version 2.0 (Apache-2.0) & GNU General Public License, Version 3 (GPL-3.0)" }
keywords = [
@ -24,7 +24,6 @@ classifiers = [
"License :: OSI Approved :: Apache Software License",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
@ -72,7 +71,7 @@ include = ["nvitop", "nvitop.*"]
safe = true
line-length = 100
skip-string-normalization = true
target-version = ["py36", "py37", "py38", "py39", "py310", "py311"]
target-version = ["py37", "py38", "py39", "py310", "py311"]
[tool.isort]
atomic = true