nicolargo/glances
1
1
Fork 0
mirror of https://github.com/nicolargo/glances.git synced 2024-11-27 10:42:52 +03:00
Code Issues Projects Releases Wiki Activity

Merge pull request #3006 from ariel-anieli/bump-ruff

Browse Source 
Bump ruff target to Python 3.9
This commit is contained in:
Nicolas Hennion 2024-11-09 14:40:58 +01:00 committed by GitHub
commit a3341b196e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 52 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
glances/cpu_percent.py
View File

@ -8,7 +8,7 @@
"""CPU percent stats shared between CPU and Quicklook plugins."""
from typing import List, Optional, TypedDict
from typing import Optional, TypedDict
import psutil
@ -121,7 +121,7 @@ class CpuPercent:
def _compute_cpu() -> float:
return psutil.cpu_percent(interval=0.0)
def get_percpu(self) -> List[PerCpuPercentInfo]:
def get_percpu(self) -> list[PerCpuPercentInfo]:
"""Update and/or return the per CPU list using the psutil library."""
# Never update more than 1 time per cached_timer_cpu
if self.timer_percpu.finished():
@ -131,7 +131,7 @@ class CpuPercent:
self.percpu_percent = self._compute_percpu()
return self.percpu_percent
def _compute_percpu(self) -> List[PerCpuPercentInfo]:
def _compute_percpu(self) -> list[PerCpuPercentInfo]:
psutil_percpu = enumerate(psutil.cpu_times_percent(interval=0.0, percpu=True))
return [
{

4
glances/globals.py
View File

@ -29,7 +29,7 @@ from configparser import ConfigParser, NoOptionError, NoSectionError
from datetime import datetime
from operator import itemgetter, methodcaller
from statistics import mean
from typing import Any, Dict, List, Union
from typing import Any, Union
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen
@ -360,7 +360,7 @@ def json_dumps(data) -> bytes:
return b(res)
def json_loads(data: Union[str, bytes, bytearray]) -> Union[Dict, List]:
def json_loads(data: Union[str, bytes, bytearray]) -> Union[dict, list]:
"""Load a JSON buffer into memory as a Python object"""
return json.loads(data)

14
glances/plugins/containers/__init__.py
View File

@ -10,7 +10,7 @@
from copy import deepcopy
from functools import partial, reduce
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Optional
from glances.globals import iteritems, itervalues, nativestr
from glances.logger import logger
@ -142,7 +142,7 @@ class PluginModel(GlancesPluginModel):
# We want to display the stat in the curse interface
self.display_curse = True
self.watchers: Dict[str, ContainersExtension] = {}
self.watchers: dict[str, ContainersExtension] = {}
# Init the Docker API
if not disable_plugin_docker:
@ -181,7 +181,7 @@ class PluginModel(GlancesPluginModel):
"""Return the key of the list."""
return 'name'
def get_export(self) -> List[Dict]:
def get_export(self) -> list[dict]:
"""Overwrite the default export method.
- Only exports containers
@ -214,7 +214,7 @@ class PluginModel(GlancesPluginModel):
@GlancesPluginModel._check_decorator
@GlancesPluginModel._log_result_decorator
def update(self) -> List[Dict]:
def update(self) -> list[dict]:
"""Update Docker and podman stats using the input method."""
# Connection should be ok
if not self.watchers:
@ -243,7 +243,7 @@ class PluginModel(GlancesPluginModel):
return self.stats
@staticmethod
def memory_usage_no_cache(mem: Dict[str, float]) -> float:
def memory_usage_no_cache(mem: dict[str, float]) -> float:
"""Return the 'real' memory usage by removing inactive_file to usage"""
# Ref: https://github.com/docker/docker-py/issues/3210
return mem['usage'] - (mem['inactive_file'] if 'inactive_file' in mem else 0)
@ -467,7 +467,7 @@ class PluginModel(GlancesPluginModel):
return ret
def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
def msg_curse(self, args=None, max_width: Optional[int] = None) -> list[str]:
"""Return the dict to display in the curse interface."""
# Init the return message
init = []
@ -529,7 +529,7 @@ class PluginModel(GlancesPluginModel):
return 'INFO'
def sort_docker_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
def sort_docker_stats(stats: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
# Make VM sort related to process sort
if glances_processes.sort_key == 'memory_percent':
sort_by = 'memory_usage'

4
glances/plugins/containers/engines/__init__.py
View File

@ -6,12 +6,12 @@
# SPDX-License-Identifier: LGPL-3.0-only
#
from typing import Any, Dict, List, Protocol, Tuple
from typing import Any, Protocol
class ContainersExtension(Protocol):
def stop(self) -> None:
raise NotImplementedError
def update(self, all_tag) -> Tuple[Dict, List[Dict[str, Any]]]:
def update(self, all_tag) -> tuple[dict, list[dict[str, Any]]]:
raise NotImplementedError

18
glances/plugins/containers/engines/docker.py
View File

@ -9,7 +9,7 @@
"""Docker Extension unit for Glances' Containers plugin."""
import time
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Optional
from glances.globals import iterkeys, itervalues, nativestr, pretty_date, replace_special_chars
from glances.logger import logger
@ -54,7 +54,7 @@ class DockerStatsFetcher:
self._streamer.stop()
@property
def activity_stats(self) -> Dict[str, Dict[str, Any]]:
def activity_stats(self) -> dict[str, dict[str, Any]]:
"""Activity Stats
Each successive access of activity_stats will cause computation of activity_stats
@ -64,7 +64,7 @@ class DockerStatsFetcher:
self._last_stats_computed_time = time.time()
return computed_activity_stats
def _compute_activity_stats(self) -> Dict[str, Dict[str, Any]]:
def _compute_activity_stats(self) -> dict[str, dict[str, Any]]:
with self._streamer.result_lock:
io_stats = self._get_io_stats()
cpu_stats = self._get_cpu_stats()
@ -83,7 +83,7 @@ class DockerStatsFetcher:
# In case no update, default to 1
return max(1, self._streamer.last_update_time - self._last_stats_computed_time)
def _get_cpu_stats(self) -> Optional[Dict[str, float]]:
def _get_cpu_stats(self) -> Optional[dict[str, float]]:
"""Return the container CPU usage.
Output: a dict {'total': 1.49}
@ -117,7 +117,7 @@ class DockerStatsFetcher:
# Return the stats
return stats
def _get_memory_stats(self) -> Optional[Dict[str, float]]:
def _get_memory_stats(self) -> Optional[dict[str, float]]:
"""Return the container MEMORY.
Output: a dict {'usage': ..., 'limit': ..., 'inactive_file': ...}
@ -140,7 +140,7 @@ class DockerStatsFetcher:
# Return the stats
return stats
def _get_network_stats(self) -> Optional[Dict[str, float]]:
def _get_network_stats(self) -> Optional[dict[str, float]]:
"""Return the container network usage using the Docker API (v1.0 or higher).
Output: a dict {'time_since_update': 3000, 'rx': 10, 'tx': 65}.
@ -169,7 +169,7 @@ class DockerStatsFetcher:
# Return the stats
return stats
def _get_io_stats(self) -> Optional[Dict[str, float]]:
def _get_io_stats(self) -> Optional[dict[str, float]]:
"""Return the container IO usage using the Docker API (v1.0 or higher).
Output: a dict {'time_since_update': 3000, 'ior': 10, 'iow': 65}.
@ -245,7 +245,7 @@ class DockerExtension:
for t in itervalues(self.stats_fetchers):
t.stop()
def update(self, all_tag) -> Tuple[Dict, List[Dict]]:
def update(self, all_tag) -> tuple[dict, list[dict]]:
"""Update Docker stats using the input method."""
if not self.client or self.disable:
@ -293,7 +293,7 @@ class DockerExtension:
"""Return the key of the list."""
return 'name'
def generate_stats(self, container) -> Dict[str, Any]:
def generate_stats(self, container) -> dict[str, Any]:
# Init the stats for the current container
stats = {
'key': self.key,

20
glances/plugins/containers/engines/podman.py
View File

@ -9,7 +9,7 @@
import time
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
from typing import Any, Optional
from glances.globals import iterkeys, itervalues, nativestr, pretty_date, replace_special_chars, string_value_to_float
from glances.logger import logger
@ -47,7 +47,7 @@ class PodmanContainerStatsFetcher:
def stop(self):
self._streamer.stop()
def get_streamed_stats(self) -> Dict[str, Any]:
def get_streamed_stats(self) -> dict[str, Any]:
stats = self._streamer.stats
if stats["Error"]:
logger.error(f"containers (Podman) Container({self._container.id}): Stats fetching failed")
@ -56,7 +56,7 @@ class PodmanContainerStatsFetcher:
return stats["Stats"][0]
@property
def activity_stats(self) -> Dict[str, Any]:
def activity_stats(self) -> dict[str, Any]:
"""Activity Stats
Each successive access of activity_stats will cause computation of activity_stats
@ -66,7 +66,7 @@ class PodmanContainerStatsFetcher:
self._last_stats_computed_time = time.time()
return computed_activity_stats
def _compute_activity_stats(self) -> Dict[str, Dict[str, Any]]:
def _compute_activity_stats(self) -> dict[str, dict[str, Any]]:
stats = {"cpu": {}, "memory": {}, "io": {}, "network": {}}
api_stats = self.get_streamed_stats()
@ -157,7 +157,7 @@ class PodmanPodStatsFetcher:
return result_stats
def _get_cpu_stats(self, stats: Dict) -> Optional[Dict]:
def _get_cpu_stats(self, stats: dict) -> Optional[dict]:
"""Return the container CPU usage.
Output: a dict {'total': 1.49}
@ -169,7 +169,7 @@ class PodmanPodStatsFetcher:
cpu_usage = string_value_to_float(stats["CPU"].rstrip("%"))
return {"total": cpu_usage}
def _get_memory_stats(self, stats) -> Optional[Dict]:
def _get_memory_stats(self, stats) -> Optional[dict]:
"""Return the container MEMORY.
Output: a dict {'usage': ..., 'limit': ...}
@ -190,7 +190,7 @@ class PodmanPodStatsFetcher:
return {'usage': usage, 'limit': limit, 'inactive_file': 0}
def _get_network_stats(self, stats) -> Optional[Dict]:
def _get_network_stats(self, stats) -> Optional[dict]:
"""Return the container network usage using the Docker API (v1.0 or higher).
Output: a dict {'time_since_update': 3000, 'rx': 10, 'tx': 65}.
@ -216,7 +216,7 @@ class PodmanPodStatsFetcher:
# Hardcode `time_since_update` to 1 as podman docs don't specify the rate calculation procedure
return {"rx": rx, "tx": tx, "time_since_update": 1}
def _get_io_stats(self, stats) -> Optional[Dict]:
def _get_io_stats(self, stats) -> Optional[dict]:
"""Return the container IO usage using the Docker API (v1.0 or higher).
Output: a dict {'time_since_update': 3000, 'ior': 10, 'iow': 65}.
@ -287,7 +287,7 @@ class PodmanExtension:
if self.pods_stats_fetcher:
self.pods_stats_fetcher.stop()
def update(self, all_tag) -> Tuple[Dict, list[Dict[str, Any]]]:
def update(self, all_tag) -> tuple[dict, list[dict[str, Any]]]:
"""Update Podman stats using the input method."""
if not self.client or self.disable:
@ -344,7 +344,7 @@ class PodmanExtension:
"""Return the key of the list."""
return 'name'
def generate_stats(self, container) -> Dict[str, Any]:
def generate_stats(self, container) -> dict[str, Any]:
# Init the stats for the current container
stats = {
'key': self.key,

14
glances/plugins/sensors/__init__.py
View File

@ -11,7 +11,7 @@
import warnings
from concurrent.futures import ThreadPoolExecutor
from enum import Enum
from typing import Any, Dict, List, Literal
from typing import Any, Literal
import psutil
@ -105,7 +105,7 @@ class PluginModel(GlancesPluginModel):
batpercent_plugin = BatPercentPluginModel(args=args, config=config)
logger.debug(f"Battery sensor plugin init duration: {start_duration.get()} seconds")
self.sensors_grab_map: Dict[SensorType, Any] = {}
self.sensors_grab_map: dict[SensorType, Any] = {}
if glances_grab_sensors_cpu_temp.init:
self.sensors_grab_map[SensorType.CPU_TEMP] = glances_grab_sensors_cpu_temp
@ -116,7 +116,7 @@ class PluginModel(GlancesPluginModel):
self.sensors_grab_map[SensorType.HDD_TEMP] = hddtemp_plugin
self.sensors_grab_map[SensorType.BATTERY] = batpercent_plugin
self.sensors_grab_map: Dict[SensorType, Any] = {}
self.sensors_grab_map: dict[SensorType, Any] = {}
if glances_grab_sensors_cpu_temp.init:
self.sensors_grab_map[SensorType.CPU_TEMP] = glances_grab_sensors_cpu_temp
@ -138,7 +138,7 @@ class PluginModel(GlancesPluginModel):
"""Return the key of the list."""
return 'label'
def __get_sensor_data(self, sensor_type: SensorType) -> List[Dict]:
def __get_sensor_data(self, sensor_type: SensorType) -> list[dict]:
try:
data = self.sensors_grab_map[sensor_type].update()
data = self.__set_type(data, sensor_type)
@ -203,7 +203,7 @@ class PluginModel(GlancesPluginModel):
return self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower())
return stats["label"]
def __set_type(self, stats: List[Dict[str, Any]], sensor_type: SensorType) -> List[Dict[str, Any]]:
def __set_type(self, stats: list[dict[str, Any]], sensor_type: SensorType) -> list[dict[str, Any]]:
"""Set the plugin type.
4 types of stats is possible in the sensors plugin:
@ -340,7 +340,7 @@ class GlancesGrabSensors:
except AttributeError:
logger.debug(f"Cannot grab {sensor_type}. Platform not supported.")
def __fetch_psutil(self) -> Dict[str, list]:
def __fetch_psutil(self) -> dict[str, list]:
if self.sensor_type == SensorType.CPU_TEMP:
# Solve an issue #1203 concerning a RunTimeError warning message displayed
# in the curses interface.
@ -355,7 +355,7 @@ class GlancesGrabSensors:
raise ValueError(f"Unsupported sensor_type: {self.sensor_type}")
def update(self) -> List[dict]:
def update(self) -> list[dict]:
"""Update the stats."""
if not self.init:
return []

12
glances/plugins/vms/__init__.py
View File

@ -9,7 +9,7 @@
"""Vms plugin."""
from copy import deepcopy
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Optional
from glances.globals import iteritems
from glances.logger import logger
@ -106,7 +106,7 @@ class PluginModel(GlancesPluginModel):
# We want to display the stat in the curse interface
self.display_curse = True
self.watchers: Dict[str, VmsExtension] = {}
self.watchers: dict[str, VmsExtension] = {}
# Init the Multipass API
self.watchers['multipass'] = VmExtension()
@ -118,7 +118,7 @@ class PluginModel(GlancesPluginModel):
"""Return the key of the list."""
return 'name'
def get_export(self) -> List[Dict]:
def get_export(self) -> list[dict]:
"""Overwrite the default export method.
- Only exports vms
@ -151,7 +151,7 @@ class PluginModel(GlancesPluginModel):
@GlancesPluginModel._check_decorator
@GlancesPluginModel._log_result_decorator
def update(self) -> List[Dict]:
def update(self) -> list[dict]:
"""Update VMs stats using the input method."""
# Connection should be ok
if not self.watchers or self.input_method != 'local':
@ -187,7 +187,7 @@ class PluginModel(GlancesPluginModel):
return True
def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
def msg_curse(self, args=None, max_width: Optional[int] = None) -> list[str]:
"""Return the dict to display in the curse interface."""
# Init the return message
ret = []
@ -293,7 +293,7 @@ class PluginModel(GlancesPluginModel):
return 'INFO'
def sort_vm_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
def sort_vm_stats(stats: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
# Make VM sort related to process sort
if glances_processes.sort_key == 'memory_percent':
sort_by = 'memory_usage'

4
glances/plugins/vms/engines/__init__.py
View File

@ -6,12 +6,12 @@
# SPDX-License-Identifier: LGPL-3.0-only
#
from typing import Any, Dict, List, Protocol, Tuple
from typing import Any, Protocol
class VmsExtension(Protocol):
def stop(self) -> None:
raise NotImplementedError
def update(self, all_tag) -> Tuple[Dict, List[Dict[str, Any]]]:
def update(self, all_tag) -> tuple[dict, list[dict[str, Any]]]:
raise NotImplementedError

6
glances/plugins/vms/engines/multipass.py
View File

@ -10,7 +10,7 @@
import json
import os
from typing import Any, Dict, List, Tuple
from typing import Any
from glances.globals import json_loads, nativestr
from glances.secure import secure_popen
@ -89,7 +89,7 @@ class VmExtension:
else:
return ret.get('info', {})
def update(self, all_tag) -> Tuple[Dict, List[Dict]]:
def update(self, all_tag) -> tuple[dict, list[dict]]:
"""Update Vm stats using the input method."""
# Can not run multipass on this system then...
if import_multipass_error_tag:
@ -115,7 +115,7 @@ class VmExtension:
def _want_display(self, vm_stats, key, values):
return vm_stats.get(key).lower() in [v.lower() for v in values]
def generate_stats(self, vm_name, vm_stats) -> Dict[str, Any]:
def generate_stats(self, vm_name, vm_stats) -> dict[str, Any]:
# Init the stats for the current vm
return {
'key': self.key,

2
pyproject.toml
View File

@ -93,7 +93,7 @@ include = ["glances*"]
[tool.ruff]
line-length = 120
target-version = "py38"
target-version = "py39"
[tool.ruff.format]
quote-style = "preserve"