glances.plugins.gpu.PluginModel.msg_curse() - Code Metrics - Inspection of "Glances version 4.0.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ee826a...d9056e )

by Nicolas

created 2024-05-12 08:45 UTC

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

↳ Parent: glances.plugins.gpu

Complexity

Conditions

Size

Total Lines	115
Code Lines	76

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	22
eloc	76
nop	3
dl	0
loc	115
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]>
# Copyright (C) 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""GPU plugin for Glances.

Currently supported:
- NVIDIA GPU (need pynvml lib)
- AMD GPU (no lib needed)
"""

from glances.globals import to_fahrenheit
from glances.plugins.gpu.cards.nvidia import NvidiaGPU
from glances.plugins.gpu.cards.amd import AmdGPU
from glances.plugins.plugin.model import GlancesPluginModel

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'gpu_id': {
        'description': 'GPU identification',
    },
    'name': {
        'description': 'GPU name',
    },
    'mem': {
        'description': 'Memory consumption',
        'unit': 'percent',
    },
    'proc': {
        'description': 'GPU processor consumption',
        'unit': 'percent',
    },
    'temperature': {
        'description': 'GPU temperature',
        'unit': 'celsius',
    },
    'fan_speed': {
        'description': 'GPU fan speed',
        'unit': 'roundperminute',
    },
}

# 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.

    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=[],
            fields_description=fields_description,
        )
        # Init the GPU API
        self.nvidia = NvidiaGPU()
        self.amd = AmdGPU()
        # Just for test purpose (uncomment to test on computer without AMD GPU)
        # self.amd = AmdGPU(drm_root_folder='./test-data/plugins/gpu/amd/sys/class/drm')

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

    def exit(self):
        """Overwrite the exit method to close the GPU API."""
        self.nvidia.exit()
        self.amd.exit()

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

    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()

        # Get the stats
        stats.extend(self.nvidia.get_device_stats())
        stats.extend(self.amd.get_device_stats())

        # !!!
        # Uncomment to test on computer without GPU
        # One GPU sample:
        # stats = [
        #     {
        #         "key": "gpu_id",
        #         "gpu_id": "nvidia0",
        #         "name": "Fake GeForce GTX",
        #         "mem": 5.792331695556641,
        #         "proc": 4,
        #         "temperature": 26,
        #         "fan_speed": 30
        #     }
        # ]
        # Two GPU sample:
        # stats = [
        #     {
        #         "key": "gpu_id",
        #         "gpu_id": "nvidia0",
        #         "name": "Fake GeForce GTX1",
        #         "mem": 5.792331695556641,
        #         "proc": 4,
        #         "temperature": 26,
        #         "fan_speed": 30
        #     },
        #     {
        #         "key": "gpu_id",
        #         "gpu_id": "nvidia1",
        #         "name": "Fake GeForce GTX1",
        #         "mem": 15,
        #         "proc": 8,
        #         "temperature": 65,
        #         "fan_speed": 75
        #     }
        # ]

        # 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_stats['name'])
            else:
                header += ' GPUs'
        elif same_name:
            header += '{}'.format(gpu_stats['name'])
        else:
            header += '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 = '{:>7}'.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


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2020 Kirby Banman <[email protected]>
6			# Copyright (C) 2024 Nicolas Hennion <[email protected]>
7			#
8			# SPDX-License-Identifier: LGPL-3.0-only
9			#
10
11			"""GPU plugin for Glances.
12
13			Currently supported:
14			- NVIDIA GPU (need pynvml lib)
15			- AMD GPU (no lib needed)
16			"""
17
18			from glances.globals import to_fahrenheit
19			from glances.plugins.gpu.cards.nvidia import NvidiaGPU
20			from glances.plugins.gpu.cards.amd import AmdGPU
21			from glances.plugins.plugin.model import GlancesPluginModel
22
23			# Fields description
24			# description: human readable description
25			# short_name: shortname to use un UI
26			# unit: unit type
27			# rate: is it a rate ? If yes, // by time_since_update when displayed,
28			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
29			fields_description = {
30			'gpu_id': {
31			'description': 'GPU identification',
32			},
33			'name': {
34			'description': 'GPU name',
35			},
36			'mem': {
37			'description': 'Memory consumption',
38			'unit': 'percent',
39			},
40			'proc': {
41			'description': 'GPU processor consumption',
42			'unit': 'percent',
43			},
44			'temperature': {
45			'description': 'GPU temperature',
46			'unit': 'celsius',
47			},
48			'fan_speed': {
49			'description': 'GPU fan speed',
50			'unit': 'roundperminute',
51			},
52			}
53
54			# Define the history items list
55			# All items in this list will be historised if the --enable-history tag is set
56			items_history_list = [
57			{'name': 'proc', 'description': 'GPU processor', 'y_unit': '%'},
58			{'name': 'mem', 'description': 'Memory consumption', 'y_unit': '%'},
59			]
60
61
62			class PluginModel(GlancesPluginModel):
63			"""Glances GPU plugin.
64
65			stats is a list of dictionaries with one entry per GPU
66			"""
67
68			def __init__(self, args=None, config=None):
69			"""Init the plugin."""
70			super(PluginModel, self).__init__(
71			args=args,
72			config=config,
73			items_history_list=items_history_list,
74			stats_init_value=[],
75			fields_description=fields_description,
76			)
77			# Init the GPU API
78			self.nvidia = NvidiaGPU()
79			self.amd = AmdGPU()
80			# Just for test purpose (uncomment to test on computer without AMD GPU)
81			# self.amd = AmdGPU(drm_root_folder='./test-data/plugins/gpu/amd/sys/class/drm')
82
83			# We want to display the stat in the curse interface
84			self.display_curse = True
85
86			def exit(self):
87			"""Overwrite the exit method to close the GPU API."""
88			self.nvidia.exit()
89			self.amd.exit()
90
91			# Call the father exit method
92			super(PluginModel, self).exit()
93
94			def get_key(self):
95			"""Return the key of the list."""
96			return 'gpu_id'
97
98			@GlancesPluginModel._check_decorator
99			@GlancesPluginModel._log_result_decorator
100			def update(self):
101			"""Update the GPU stats."""
102			# Init new stats
103			stats = self.get_init_value()
104
105			# Get the stats
106			stats.extend(self.nvidia.get_device_stats())
107			stats.extend(self.amd.get_device_stats())
108
109			# !!!
110			# Uncomment to test on computer without GPU
111			# One GPU sample:
112			# stats = [
113			# {
114			# "key": "gpu_id",
115			# "gpu_id": "nvidia0",
116			# "name": "Fake GeForce GTX",
117			# "mem": 5.792331695556641,
118			# "proc": 4,
119			# "temperature": 26,
120			# "fan_speed": 30
121			# }
122			# ]
123			# Two GPU sample:
124			# stats = [
125			# {
126			# "key": "gpu_id",
127			# "gpu_id": "nvidia0",
128			# "name": "Fake GeForce GTX1",
129			# "mem": 5.792331695556641,
130			# "proc": 4,
131			# "temperature": 26,
132			# "fan_speed": 30
133			# },
134			# {
135			# "key": "gpu_id",
136			# "gpu_id": "nvidia1",
137			# "name": "Fake GeForce GTX1",
138			# "mem": 15,
139			# "proc": 8,
140			# "temperature": 65,
141			# "fan_speed": 75
142			# }
143			# ]
144
145			# Update the stats
146			self.stats = stats
147
148			return self.stats
149
150			def update_views(self):
151			"""Update stats views."""
152			# Call the father's method
153			super(PluginModel, self).update_views()
154
155			# Add specifics information
156			# Alert
157			for i in self.stats:
158			# Init the views for the current GPU
159			self.views[i[self.get_key()]] = {'proc': {}, 'mem': {}, 'temperature': {}}
160			# Processor alert
161			if 'proc' in i:
162			alert = self.get_alert(i['proc'], header='proc')
163			self.views[i[self.get_key()]]['proc']['decoration'] = alert
164			# Memory alert
165			if 'mem' in i:
166			alert = self.get_alert(i['mem'], header='mem')
167			self.views[i[self.get_key()]]['mem']['decoration'] = alert
168			# Temperature alert
169			if 'temperature' in i:
170			alert = self.get_alert(i['temperature'], header='temperature')
171			self.views[i[self.get_key()]]['temperature']['decoration'] = alert
172
173			return True
174
175			def msg_curse(self, args=None, max_width=None):
176			"""Return the dict to display in the curse interface."""
177			# Init the return message
178			ret = []
179
180			# Only process if stats exist, not empty (issue #871) and plugin not disabled
181			if not self.stats or (self.stats == []) or self.is_disabled():
182			return ret
183
184			# Check if all GPU have the same name
185			same_name = all(s['name'] == self.stats[0]['name'] for s in self.stats)
186
187			# gpu_stats contain the first GPU in the list
188			gpu_stats = self.stats[0]
189
190			# Header
191			header = ''
192			if len(self.stats) > 1:
193			header += '{}'.format(len(self.stats))
194			if same_name:
195			header += ' {}'.format(gpu_stats['name'])
196			else:
197			header += ' GPUs'
198			elif same_name:
199			header += '{}'.format(gpu_stats['name'])
200			else:
201			header += 'GPU'
202			msg = header[:17]
203			ret.append(self.curse_add_line(msg, "TITLE"))
204
205			# Build the string message
206			if len(self.stats) == 1 or args.meangpu:
207			# GPU stat summary or mono GPU
208			# New line
209			ret.append(self.curse_new_line())
210			# GPU PROC
211			try:
212			mean_proc = sum(s['proc'] for s in self.stats if s is not None) / len(self.stats)
213			except TypeError:
214			mean_proc_msg = '{:>4}'.format('N/A')
215			else:
216			mean_proc_msg = '{:>3.0f}%'.format(mean_proc)
217			if len(self.stats) > 1:
218			msg = '{:13}'.format('proc mean:')
219			else:
220			msg = '{:13}'.format('proc:')
221			ret.append(self.curse_add_line(msg))
222			ret.append(
223			self.curse_add_line(
224			mean_proc_msg, self.get_views(item=gpu_stats[self.get_key()], key='proc', option='decoration')
225			)
226			)
227			# New line
228			ret.append(self.curse_new_line())
229			# GPU MEM
230			try:
231			mean_mem = sum(s['mem'] for s in self.stats if s is not None) / len(self.stats)
232			except TypeError:
233			mean_mem_msg = '{:>4}'.format('N/A')
234			else:
235			mean_mem_msg = '{:>3.0f}%'.format(mean_mem)
236			if len(self.stats) > 1:
237			msg = '{:13}'.format('mem mean:')
238			else:
239			msg = '{:13}'.format('mem:')
240			ret.append(self.curse_add_line(msg))
241			ret.append(
242			self.curse_add_line(
243			mean_mem_msg, self.get_views(item=gpu_stats[self.get_key()], key='mem', option='decoration')
244			)
245			)
246			# New line
247			ret.append(self.curse_new_line())
248			# GPU TEMPERATURE
249			try:
250			mean_temperature = sum(s['temperature'] for s in self.stats if s is not None) / len(self.stats)
251			except TypeError:
252			mean_temperature_msg = '{:>4}'.format('N/A')
253			else:
254			unit = 'C'
255			if args.fahrenheit:
256			mean_temperature = to_fahrenheit(mean_temperature)
257			unit = 'F'
258			mean_temperature_msg = '{:>3.0f}{}'.format(mean_temperature, unit)
259			if len(self.stats) > 1:
260			msg = '{:13}'.format('temp mean:')
261			else:
262			msg = '{:13}'.format('temperature:')
263			ret.append(self.curse_add_line(msg))
264			ret.append(
265			self.curse_add_line(
266			mean_temperature_msg,
267			self.get_views(item=gpu_stats[self.get_key()], key='temperature', option='decoration'),
268			)
269			)
270			else:
271			# Multi GPU
272			# Temperature is not displayed in this mode...
273			for gpu_stats in self.stats:
274			# New line
275			ret.append(self.curse_new_line())
276			# GPU ID + PROC + MEM + TEMPERATURE
277			id_msg = '{:>7}'.format(gpu_stats['gpu_id'])
278			try:
279			proc_msg = '{:>3.0f}%'.format(gpu_stats['proc'])
280			except (ValueError, TypeError):
281			proc_msg = '{:>4}'.format('N/A')
282			try:
283			mem_msg = '{:>3.0f}%'.format(gpu_stats['mem'])
284			except (ValueError, TypeError):
285			mem_msg = '{:>4}'.format('N/A')
286			msg = '{} {} mem {}'.format(id_msg, proc_msg, mem_msg)
287			ret.append(self.curse_add_line(msg))
288
289			return ret
290

nicolargo / glances

Push — master ( ee826a...d9056e )

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