glances.plugins.gpu.PluginModel.msg_curse() - Code Metrics - Inspection of "Refactor plugin folder in the same way than export..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 30cc9f...48251c )

by Nicolas

created 2023-10-15 16:05 UTC

glances.plugins.gpu.PluginModel.msg_curse() F

↳ Parent: glances.plugins.gpu

Complexity

Conditions

Size

Total Lines	111
Code Lines	73

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	21
eloc	73
nop	3
dl	0
loc	111
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2020 Kirby Banman <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""GPU plugin (limited to NVIDIA chipsets)."""

from glances.globals import nativestr, to_fahrenheit
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

# In Glances 3.1.4 or higher, we use the py3nvml lib (see issue #1523)
try:
    import py3nvml.py3nvml as pynvml
except Exception as e:
    import_error_tag = True
    # Display debug message if import KeyError
    logger.warning("Missing Python Lib ({}), Nvidia GPU plugin is disabled".format(e))
else:
    import_error_tag = False

# Define the history items list
# All items in this list will be historised if the --enable-history tag is set
items_history_list = [
    {'name': 'proc', 'description': 'GPU processor', 'y_unit': '%'},
    {'name': 'mem', 'description': 'Memory consumption', 'y_unit': '%'},
]


class PluginModel(GlancesPluginModel):
    """Glances GPU plugin (limited to NVIDIA chipsets).

    stats is a list of dictionaries with one entry per GPU
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super(PluginModel, self).__init__(args=args,
                                          config=config,
                                          items_history_list=items_history_list,
                                          stats_init_value=[])

        # Init the Nvidia API
        self.init_nvidia()

        # We want to display the stat in the curse interface
        self.display_curse = True

    def init_nvidia(self):
        """Init the NVIDIA API."""
        if import_error_tag:
            self.nvml_ready = False

        try:
            pynvml.nvmlInit()
            self.device_handles = get_device_handles()
            self.nvml_ready = True
        except Exception:
            logger.debug("pynvml could not be initialized.")
            self.nvml_ready = False

        return self.nvml_ready

    def get_key(self):
        """Return the key of the list."""
        return 'gpu_id'

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the GPU stats."""
        # Init new stats
        stats = self.get_init_value()

        if not self.nvml_ready:
            # !!!
            # Uncomment to test on computer without GPU
            # One GPU sample:
            # self.stats = [
            #     {
            #         "key": "gpu_id",
            #         "gpu_id": 0,
            #         "name": "Fake GeForce GTX",
            #         "mem": 5.792331695556641,
            #         "proc": 4,
            #         "temperature": 26,
            #         "fan_speed": 30
            #     }
            # ]
            # Two GPU sample:
            # self.stats = [
            #     {
            #         "key": "gpu_id",
            #         "gpu_id": 0,
            #         "name": "Fake GeForce GTX1",
            #         "mem": 5.792331695556641,
            #         "proc": 4,
            #         "temperature": 26,
            #         "fan_speed": 30
            #     },
            #     {
            #         "key": "gpu_id",
            #         "gpu_id": 1,
            #         "name": "Fake GeForce GTX2",
            #         "mem": 15,
            #         "proc": 8,
            #         "temperature": 65,
            #         "fan_speed": 75
            #     }
            # ]
            return self.stats

        if self.input_method == 'local':
            stats = self.get_device_stats()
        elif self.input_method == 'snmp':
            # not available
            pass

        # Update the stats
        self.stats = stats

        return self.stats

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super(PluginModel, self).update_views()

        # Add specifics information
        # Alert
        for i in self.stats:
            # Init the views for the current GPU
            self.views[i[self.get_key()]] = {'proc': {}, 'mem': {}, 'temperature': {}}
            # Processor alert
            if 'proc' in i:
                alert = self.get_alert(i['proc'], header='proc')
                self.views[i[self.get_key()]]['proc']['decoration'] = alert
            # Memory alert
            if 'mem' in i:
                alert = self.get_alert(i['mem'], header='mem')
                self.views[i[self.get_key()]]['mem']['decoration'] = alert
            # Temperature alert
            if 'temperature' in i:
                alert = self.get_alert(i['temperature'], header='temperature')
                self.views[i[self.get_key()]]['temperature']['decoration'] = alert

        return True

    def msg_curse(self, args=None, max_width=None):
        """Return the dict to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist, not empty (issue #871) and plugin not disabled
        if not self.stats or (self.stats == []) or self.is_disabled():
            return ret

        # Check if all GPU have the same name
        same_name = all(s['name'] == self.stats[0]['name'] for s in self.stats)

        # gpu_stats contain the first GPU in the list
        gpu_stats = self.stats[0]

        # Header
        header = ''
        if len(self.stats) > 1:
            header += '{} '.format(len(self.stats))
        if same_name:
            header += '{} {}'.format('GPU', gpu_stats['name'])
        else:
            header += '{}'.format('GPU')
        msg = header[:17]
        ret.append(self.curse_add_line(msg, "TITLE"))

        # Build the string message
        if len(self.stats) == 1 or args.meangpu:
            # GPU stat summary or mono GPU
            # New line
            ret.append(self.curse_new_line())
            # GPU PROC
            try:
                mean_proc = sum(s['proc'] for s in self.stats if s is not None) / len(self.stats)
            except TypeError:
                mean_proc_msg = '{:>4}'.format('N/A')
            else:
                mean_proc_msg = '{:>3.0f}%'.format(mean_proc)
            if len(self.stats) > 1:
                msg = '{:13}'.format('proc mean:')
            else:
                msg = '{:13}'.format('proc:')
            ret.append(self.curse_add_line(msg))
            ret.append(
                self.curse_add_line(
                    mean_proc_msg, self.get_views(item=gpu_stats[self.get_key()], key='proc', option='decoration')
                )
            )
            # New line
            ret.append(self.curse_new_line())
            # GPU MEM
            try:
                mean_mem = sum(s['mem'] for s in self.stats if s is not None) / len(self.stats)
            except TypeError:
                mean_mem_msg = '{:>4}'.format('N/A')
            else:
                mean_mem_msg = '{:>3.0f}%'.format(mean_mem)
            if len(self.stats) > 1:
                msg = '{:13}'.format('mem mean:')
            else:
                msg = '{:13}'.format('mem:')
            ret.append(self.curse_add_line(msg))
            ret.append(
                self.curse_add_line(
                    mean_mem_msg, self.get_views(item=gpu_stats[self.get_key()], key='mem', option='decoration')
                )
            )
            # New line
            ret.append(self.curse_new_line())
            # GPU TEMPERATURE
            try:
                mean_temperature = sum(s['temperature'] for s in self.stats if s is not None) / len(self.stats)
            except TypeError:
                mean_temperature_msg = '{:>4}'.format('N/A')
            else:
                unit = 'C'
                if args.fahrenheit:
                    mean_temperature = to_fahrenheit(mean_temperature)
                    unit = 'F'
                mean_temperature_msg = '{:>3.0f}{}'.format(mean_temperature, unit)
            if len(self.stats) > 1:
                msg = '{:13}'.format('temp mean:')
            else:
                msg = '{:13}'.format('temperature:')
            ret.append(self.curse_add_line(msg))
            ret.append(
                self.curse_add_line(
                    mean_temperature_msg,
                    self.get_views(item=gpu_stats[self.get_key()], key='temperature', option='decoration'),
                )
            )
        else:
            # Multi GPU
            # Temperature is not displayed in this mode...
            for gpu_stats in self.stats:
                # New line
                ret.append(self.curse_new_line())
                # GPU ID + PROC + MEM + TEMPERATURE
                id_msg = '{}'.format(gpu_stats['gpu_id'])
                try:
                    proc_msg = '{:>3.0f}%'.format(gpu_stats['proc'])
                except (ValueError, TypeError):
                    proc_msg = '{:>4}'.format('N/A')
                try:
                    mem_msg = '{:>3.0f}%'.format(gpu_stats['mem'])
                except (ValueError, TypeError):
                    mem_msg = '{:>4}'.format('N/A')
                msg = '{}: {} mem: {}'.format(id_msg, proc_msg, mem_msg)
                ret.append(self.curse_add_line(msg))

        return ret

    def get_device_stats(self):
        """Get GPU stats."""
        stats = []

        for index, device_handle in enumerate(self.device_handles):
            device_stats = dict()
            # Dictionary key is the GPU_ID
            device_stats['key'] = self.get_key()
            # GPU id (for multiple GPU, start at 0)
            device_stats['gpu_id'] = index
            # GPU name
            device_stats['name'] = get_device_name(device_handle)
            # Memory consumption in % (not available on all GPU)
            device_stats['mem'] = get_mem(device_handle)
            # Processor consumption in %
            device_stats['proc'] = get_proc(device_handle)
            # Processor temperature in °C
            device_stats['temperature'] = get_temperature(device_handle)
            # Fan speed in %
            device_stats['fan_speed'] = get_fan_speed(device_handle)
            stats.append(device_stats)

        return stats

    def exit(self):
        """Overwrite the exit method to close the GPU API."""
        if self.nvml_ready:
            try:
                pynvml.nvmlShutdown()
            except Exception as e:
                logger.debug("pynvml failed to shutdown correctly ({})".format(e))

        # Call the father exit method
        super(PluginModel, self).exit()


def get_device_handles():
    """Get a list of NVML device handles, one per device.

    Can throw NVMLError.
    """
    return [pynvml.nvmlDeviceGetHandleByIndex(i) for i in range(pynvml.nvmlDeviceGetCount())]


def get_device_name(device_handle):
    """Get GPU device name."""
    try:
        return nativestr(pynvml.nvmlDeviceGetName(device_handle))
    except pynvml.NVMLError:
        return "NVIDIA"


def get_mem(device_handle):
    """Get GPU device memory consumption in percent."""
    try:
        memory_info = pynvml.nvmlDeviceGetMemoryInfo(device_handle)
        return memory_info.used * 100.0 / memory_info.total
    except pynvml.NVMLError:
        return None


def get_proc(device_handle):
    """Get GPU device CPU consumption in percent."""
    try:
        return pynvml.nvmlDeviceGetUtilizationRates(device_handle).gpu
    except pynvml.NVMLError:
        return None


def get_temperature(device_handle):
    """Get GPU device CPU temperature in Celsius."""
    try:
        return pynvml.nvmlDeviceGetTemperature(device_handle, pynvml.NVML_TEMPERATURE_GPU)
    except pynvml.NVMLError:
        return None


def get_fan_speed(device_handle):
    """Get GPU device fan speed in percent."""
    try:
        return pynvml.nvmlDeviceGetFanSpeed(device_handle)
    except pynvml.NVMLError:
        return None


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2020 Kirby Banman <[email protected]>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8			#
9
10			"""GPU plugin (limited to NVIDIA chipsets)."""
11
12			from glances.globals import nativestr, to_fahrenheit
13			from glances.logger import logger
14			from glances.plugins.plugin.model import GlancesPluginModel
15
16			# In Glances 3.1.4 or higher, we use the py3nvml lib (see issue #1523)
17			try:
18			import py3nvml.py3nvml as pynvml
19			except Exception as e:
20			import_error_tag = True
21			# Display debug message if import KeyError
22			logger.warning("Missing Python Lib ({}), Nvidia GPU plugin is disabled".format(e))
23			else:
24			import_error_tag = False
25
26			# Define the history items list
27			# All items in this list will be historised if the --enable-history tag is set
28			items_history_list = [
29			{'name': 'proc', 'description': 'GPU processor', 'y_unit': '%'},
30			{'name': 'mem', 'description': 'Memory consumption', 'y_unit': '%'},
31			]
32
33
34			class PluginModel(GlancesPluginModel):
35			"""Glances GPU plugin (limited to NVIDIA chipsets).
36
37			stats is a list of dictionaries with one entry per GPU
38			"""
39
40			def __init__(self, args=None, config=None):
41			"""Init the plugin."""
42			super(PluginModel, self).__init__(args=args,
43			config=config,
44			items_history_list=items_history_list,
45			stats_init_value=[])
46
47			# Init the Nvidia API
48			self.init_nvidia()
49
50			# We want to display the stat in the curse interface
51			self.display_curse = True
52
53			def init_nvidia(self):
54			"""Init the NVIDIA API."""
55			if import_error_tag:
56			self.nvml_ready = False
57
58			try:
59			pynvml.nvmlInit()
60			self.device_handles = get_device_handles()
61			self.nvml_ready = True
62			except Exception:
63			logger.debug("pynvml could not be initialized.")
64			self.nvml_ready = False
65
66			return self.nvml_ready
67
68			def get_key(self):
69			"""Return the key of the list."""
70			return 'gpu_id'
71
72			@GlancesPluginModel._check_decorator
73			@GlancesPluginModel._log_result_decorator
74			def update(self):
75			"""Update the GPU stats."""
76			# Init new stats
77			stats = self.get_init_value()
78
79			if not self.nvml_ready:
80			# !!!
81			# Uncomment to test on computer without GPU
82			# One GPU sample:
83			# self.stats = [
84			# {
85			# "key": "gpu_id",
86			# "gpu_id": 0,
87			# "name": "Fake GeForce GTX",
88			# "mem": 5.792331695556641,
89			# "proc": 4,
90			# "temperature": 26,
91			# "fan_speed": 30
92			# }
93			# ]
94			# Two GPU sample:
95			# self.stats = [
96			# {
97			# "key": "gpu_id",
98			# "gpu_id": 0,
99			# "name": "Fake GeForce GTX1",
100			# "mem": 5.792331695556641,
101			# "proc": 4,
102			# "temperature": 26,
103			# "fan_speed": 30
104			# },
105			# {
106			# "key": "gpu_id",
107			# "gpu_id": 1,
108			# "name": "Fake GeForce GTX2",
109			# "mem": 15,
110			# "proc": 8,
111			# "temperature": 65,
112			# "fan_speed": 75
113			# }
114			# ]
115			return self.stats
116
117			if self.input_method == 'local':
118			stats = self.get_device_stats()
119			elif self.input_method == 'snmp':
120			# not available
121			pass
122
123			# Update the stats
124			self.stats = stats
125
126			return self.stats
127
128			def update_views(self):
129			"""Update stats views."""
130			# Call the father's method
131			super(PluginModel, self).update_views()
132
133			# Add specifics information
134			# Alert
135			for i in self.stats:
136			# Init the views for the current GPU
137			self.views[i[self.get_key()]] = {'proc': {}, 'mem': {}, 'temperature': {}}
138			# Processor alert
139			if 'proc' in i:
140			alert = self.get_alert(i['proc'], header='proc')
141			self.views[i[self.get_key()]]['proc']['decoration'] = alert
142			# Memory alert
143			if 'mem' in i:
144			alert = self.get_alert(i['mem'], header='mem')
145			self.views[i[self.get_key()]]['mem']['decoration'] = alert
146			# Temperature alert
147			if 'temperature' in i:
148			alert = self.get_alert(i['temperature'], header='temperature')
149			self.views[i[self.get_key()]]['temperature']['decoration'] = alert
150
151			return True
152
153			def msg_curse(self, args=None, max_width=None):
154			"""Return the dict to display in the curse interface."""
155			# Init the return message
156			ret = []
157
158			# Only process if stats exist, not empty (issue #871) and plugin not disabled
159			if not self.stats or (self.stats == []) or self.is_disabled():
160			return ret
161
162			# Check if all GPU have the same name
163			same_name = all(s['name'] == self.stats[0]['name'] for s in self.stats)
164
165			# gpu_stats contain the first GPU in the list
166			gpu_stats = self.stats[0]
167
168			# Header
169			header = ''
170			if len(self.stats) > 1:
171			header += '{} '.format(len(self.stats))
172			if same_name:
173			header += '{} {}'.format('GPU', gpu_stats['name'])
174			else:
175			header += '{}'.format('GPU')
176			msg = header[:17]
177			ret.append(self.curse_add_line(msg, "TITLE"))
178
179			# Build the string message
180			if len(self.stats) == 1 or args.meangpu:
181			# GPU stat summary or mono GPU
182			# New line
183			ret.append(self.curse_new_line())
184			# GPU PROC
185			try:
186			mean_proc = sum(s['proc'] for s in self.stats if s is not None) / len(self.stats)
187			except TypeError:
188			mean_proc_msg = '{:>4}'.format('N/A')
189			else:
190			mean_proc_msg = '{:>3.0f}%'.format(mean_proc)
191			if len(self.stats) > 1:
192			msg = '{:13}'.format('proc mean:')
193			else:
194			msg = '{:13}'.format('proc:')
195			ret.append(self.curse_add_line(msg))
196			ret.append(
197			self.curse_add_line(
198			mean_proc_msg, self.get_views(item=gpu_stats[self.get_key()], key='proc', option='decoration')
199			)
200			)
201			# New line
202			ret.append(self.curse_new_line())
203			# GPU MEM
204			try:
205			mean_mem = sum(s['mem'] for s in self.stats if s is not None) / len(self.stats)
206			except TypeError:
207			mean_mem_msg = '{:>4}'.format('N/A')
208			else:
209			mean_mem_msg = '{:>3.0f}%'.format(mean_mem)
210			if len(self.stats) > 1:
211			msg = '{:13}'.format('mem mean:')
212			else:
213			msg = '{:13}'.format('mem:')
214			ret.append(self.curse_add_line(msg))
215			ret.append(
216			self.curse_add_line(
217			mean_mem_msg, self.get_views(item=gpu_stats[self.get_key()], key='mem', option='decoration')
218			)
219			)
220			# New line
221			ret.append(self.curse_new_line())
222			# GPU TEMPERATURE
223			try:
224			mean_temperature = sum(s['temperature'] for s in self.stats if s is not None) / len(self.stats)
225			except TypeError:
226			mean_temperature_msg = '{:>4}'.format('N/A')
227			else:
228			unit = 'C'
229			if args.fahrenheit:
230			mean_temperature = to_fahrenheit(mean_temperature)
231			unit = 'F'
232			mean_temperature_msg = '{:>3.0f}{}'.format(mean_temperature, unit)
233			if len(self.stats) > 1:
234			msg = '{:13}'.format('temp mean:')
235			else:
236			msg = '{:13}'.format('temperature:')
237			ret.append(self.curse_add_line(msg))
238			ret.append(
239			self.curse_add_line(
240			mean_temperature_msg,
241			self.get_views(item=gpu_stats[self.get_key()], key='temperature', option='decoration'),
242			)
243			)
244			else:
245			# Multi GPU
246			# Temperature is not displayed in this mode...
247			for gpu_stats in self.stats:
248			# New line
249			ret.append(self.curse_new_line())
250			# GPU ID + PROC + MEM + TEMPERATURE
251			id_msg = '{}'.format(gpu_stats['gpu_id'])
252			try:
253			proc_msg = '{:>3.0f}%'.format(gpu_stats['proc'])
254			except (ValueError, TypeError):
255			proc_msg = '{:>4}'.format('N/A')
256			try:
257			mem_msg = '{:>3.0f}%'.format(gpu_stats['mem'])
258			except (ValueError, TypeError):
259			mem_msg = '{:>4}'.format('N/A')
260			msg = '{}: {} mem: {}'.format(id_msg, proc_msg, mem_msg)
261			ret.append(self.curse_add_line(msg))
262
263			return ret
264
265			def get_device_stats(self):
266			"""Get GPU stats."""
267			stats = []
268
269			for index, device_handle in enumerate(self.device_handles):
270			device_stats = dict()
271			# Dictionary key is the GPU_ID
272			device_stats['key'] = self.get_key()
273			# GPU id (for multiple GPU, start at 0)
274			device_stats['gpu_id'] = index
275			# GPU name
276			device_stats['name'] = get_device_name(device_handle)
277			# Memory consumption in % (not available on all GPU)
278			device_stats['mem'] = get_mem(device_handle)
279			# Processor consumption in %
280			device_stats['proc'] = get_proc(device_handle)
281			# Processor temperature in °C
282			device_stats['temperature'] = get_temperature(device_handle)
283			# Fan speed in %
284			device_stats['fan_speed'] = get_fan_speed(device_handle)
285			stats.append(device_stats)
286
287			return stats
288
289			def exit(self):
290			"""Overwrite the exit method to close the GPU API."""
291			if self.nvml_ready:
292			try:
293			pynvml.nvmlShutdown()
294			except Exception as e:
295			logger.debug("pynvml failed to shutdown correctly ({})".format(e))
296
297			# Call the father exit method
298			super(PluginModel, self).exit()
299
300
301			def get_device_handles():
302			"""Get a list of NVML device handles, one per device.
303
304			Can throw NVMLError.
305			"""
306			return [pynvml.nvmlDeviceGetHandleByIndex(i) for i in range(pynvml.nvmlDeviceGetCount())]
307
308
309			def get_device_name(device_handle):
310			"""Get GPU device name."""
311			try:
312			return nativestr(pynvml.nvmlDeviceGetName(device_handle))
313			except pynvml.NVMLError:
314			return "NVIDIA"
315
316
317			def get_mem(device_handle):
318			"""Get GPU device memory consumption in percent."""
319			try:
320			memory_info = pynvml.nvmlDeviceGetMemoryInfo(device_handle)
321			return memory_info.used * 100.0 / memory_info.total
322			except pynvml.NVMLError:
323			return None
324
325
326			def get_proc(device_handle):
327			"""Get GPU device CPU consumption in percent."""
328			try:
329			return pynvml.nvmlDeviceGetUtilizationRates(device_handle).gpu
330			except pynvml.NVMLError:
331			return None
332
333
334			def get_temperature(device_handle):
335			"""Get GPU device CPU temperature in Celsius."""
336			try:
337			return pynvml.nvmlDeviceGetTemperature(device_handle, pynvml.NVML_TEMPERATURE_GPU)
338			except pynvml.NVMLError:
339			return None
340
341
342			def get_fan_speed(device_handle):
343			"""Get GPU device fan speed in percent."""
344			try:
345			return pynvml.nvmlDeviceGetFanSpeed(device_handle)
346			except pynvml.NVMLError:
347			return None
348

nicolargo / glances

Push — develop ( 30cc9f...48251c )

glances.plugins.gpu.PluginModel.msg_curse() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like