glances.plugins.diskio.DiskioPlugin.msg_curse() - Code Metrics - Inspection of "Merge branch 'develop' of github.com:nicolargo/gla..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 3f5c1f...948bc9 )

by Nicolas

created 2025-08-17 14:56 UTC

glances.plugins.diskio.DiskioPlugin.msg_curse() D

↳ Parent: glances.plugins.diskio

Complexity

Conditions

Size

Total Lines	99
Code Lines	66

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	66
nop	3
dl	0
loc	99
rs	4.4727
c	0
b	0
f	0

How to fix Long Method Complexity

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Disk I/O plugin."""

import psutil

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

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: if True then compute and add *_gauge and *_rate_per_is fields
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'disk_name': {'description': 'Disk name.'},
    'read_count': {
        'description': 'Number of reads.',
        'rate': True,
        'unit': 'number',
    },
    'write_count': {
        'description': 'Number of writes.',
        'rate': True,
        'unit': 'number',
    },
    'read_bytes': {
        'description': 'Number of bytes read.',
        'rate': True,
        'unit': 'byte',
    },
    'write_bytes': {
        'description': 'Number of bytes written.',
        'rate': True,
        'unit': 'byte',
    },
    'read_time': {
        'description': 'Time spent reading.',
        'rate': True,
        'unit': 'millisecond',
    },
    'write_time': {
        'description': 'Time spent writing.',
        'rate': True,
        'unit': 'millisecond',
    },
    'read_latency': {
        'description': 'Mean time spent reading per operation.',
        'unit': 'millisecond',
    },
    'write_latency': {
        'description': 'Mean time spent writing per operation.',
        'unit': 'millisecond',
    },
}

# Define the history items list
items_history_list = [
    {'name': 'read_bytes_rate_per_sec', 'description': 'Bytes read per second', 'y_unit': 'B/s'},
    {'name': 'write_bytes_rate_per_sec', 'description': 'Bytes write per second', 'y_unit': 'B/s'},
]


class DiskioPlugin(GlancesPluginModel):
    """Glances disks I/O plugin.

    stats is a list
    """

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

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

        # Hide stats if it has never been != 0
        if config is not None:
            self.hide_zero = config.get_bool_value(self.plugin_name, 'hide_zero', default=False)
            self.hide_threshold_bytes = config.get_int_value(self.plugin_name, 'hide_threshold_bytes', default=0)
        else:
            self.hide_zero = False
            self.hide_threshold_bytes = 0
        self.hide_zero_fields = ['read_bytes_rate_per_sec', 'write_bytes_rate_per_sec']

        # Force a first update because we need two updates to have the first stat
        self.update()
        self.refresh_timer.set(0)

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

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update disk I/O stats using the input method."""
        # Update the stats
        if self.input_method == 'local':
            stats = self.update_local()

            # Compute latency (need rate stats, so should be done after decorator)
            stats = self.update_latency(stats)
        else:
            stats = self.get_init_value()

        # Update the stats
        self.stats = stats

        return self.stats

    def update_latency(self, stats):
        """Update the latency stats."""
        # Compute read/write latency if we have the rate stats
        for stat in stats:
            # Compute read/write latency if we have the rate stats
            if stat.get("read_count_rate_per_sec", 0) > 0:
                stat["read_latency"] = int(stat["read_time_rate_per_sec"] / stat["read_count_rate_per_sec"])
            else:
                stat["read_latency"] = 0

            if stat.get("write_count_rate_per_sec", 0) > 0:
                stat["write_latency"] = int(stat["write_time_rate_per_sec"] / stat["write_count_rate_per_sec"])
            else:
                stat["write_latency"] = 0

        return stats

    @GlancesPluginModel._manage_rate
    def update_local(self):
        stats = self.get_init_value()

        try:
            diskio = psutil.disk_io_counters(perdisk=True)
        except Exception:
            return stats

        for disk_name, disk_stat in diskio.items():
            # By default, RamFS is not displayed (issue #714)
            if self.args is not None and not self.args.diskio_show_ramfs and disk_name.startswith('ram'):
                continue

            # Shall we display the stats ?
            if not self.is_display(disk_name):
                continue

            # Filter stats to keep only the fields we want (define in fields_description)
            # It will also convert psutil objects to a standard Python dict
            stat = self.filter_stats(disk_stat)

            # Add the key
            stat['key'] = self.get_key()

            # Add disk name
            stat['disk_name'] = disk_name

            # Add alias if exist (define in the configuration file)
            if self.has_alias(disk_name) is not None:
                stat['alias'] = self.has_alias(disk_name)

            stats.append(stat)

        return stats

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

        # Alert
        for i in self.get_raw():
            disk_real_name = i['disk_name']

            # # Skip alert if no timespan to measure
            # if not i.get('read_bytes_rate_per_sec') or not i.get('write_bytes_rate_per_sec'):
            #     continue

            # Decorate the bitrate with the configuration file
            alert_rx = self.get_alert(i['read_bytes'], header=disk_real_name + '_rx')
            alert_tx = self.get_alert(i['write_bytes'], header=disk_real_name + '_tx')
            self.views[i[self.get_key()]]['read_bytes']['decoration'] = alert_rx
            self.views[i[self.get_key()]]['write_bytes']['decoration'] = alert_tx

            # Decorate the latency with the configuration file
            # Try to get the read/write latency for the current disk
            alert_latency_rx = self.get_alert(i['read_latency'], header=disk_real_name + '_rx_latency')
            alert_latency_tx = self.get_alert(i['write_latency'], header=disk_real_name + '_tx_latency')
            # If the alert is not defined, use the default one
            if alert_latency_rx == 'DEFAULT':
                alert_latency_rx = self.get_alert(i['read_latency'], header='rx_latency')
            if alert_latency_tx == 'DEFAULT':
                alert_latency_tx = self.get_alert(i['write_latency'], header='tx_latency')
            self.views[i[self.get_key()]]['read_latency']['decoration'] = alert_latency_rx
            self.views[i[self.get_key()]]['write_latency']['decoration'] = alert_latency_tx

    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 and display plugin enable...
        if not self.stats or self.is_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 13
        else:
            # No max_width defined, return an empty curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return ret

        # Header
        msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.diskio_iops:
            msg = '{:>8}'.format('IOR/s')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('IOW/s')
            ret.append(self.curse_add_line(msg))
        elif args.diskio_latency:
            msg = '{:>8}'.format('ms/opR')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('ms/opW')
            ret.append(self.curse_add_line(msg))
        else:
            msg = '{:>8}'.format('R/s')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('W/s')
            ret.append(self.curse_add_line(msg))
        # Disk list (sorted by name)
        for i in self.sorted_stats():
            # Hide stats if never be different from 0 (issue #1787)
            if all(self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields):
                continue
            # Is there an alias for the disk name ?
            disk_name = i['alias'] if 'alias' in i else i['disk_name']
            # New line
            ret.append(self.curse_new_line())
            if len(disk_name) > name_max_width:
                # Cut disk name if it is too long
                disk_name = disk_name[:name_max_width] + '_'
            msg = '{:{width}}'.format(nativestr(disk_name), width=name_max_width + 1)
            ret.append(self.curse_add_line(msg))
            if args.diskio_iops:
                # count
                txps = self.auto_unit(i.get('read_count_rate_per_sec', None))
                rxps = self.auto_unit(i.get('write_count_rate_per_sec', None))
                msg = f'{txps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='read_count', option='decoration')
                    )
                )
                msg = f'{rxps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='write_count', option='decoration')
                    )
                )
            elif args.diskio_latency:
                # latency (mean time spent reading/writing per operation)
                txps = self.auto_unit(i.get('read_latency', None), low_precision=True)
                rxps = self.auto_unit(i.get('write_latency', None), low_precision=True)
                msg = f'{txps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='read_latency', option='decoration')
                    )
                )
                msg = f'{rxps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='write_latency', option='decoration')
                    )
                )
            else:
                # Bitrate
                txps = self.auto_unit(i.get('read_bytes_rate_per_sec', None))
                rxps = self.auto_unit(i.get('write_bytes_rate_per_sec', None))
                msg = f'{txps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='read_bytes', option='decoration')
                    )
                )
                msg = f'{rxps:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='write_bytes', option='decoration')
                    )
                )

        return ret


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""Disk I/O plugin."""
10
11			import psutil
12
13			from glances.globals import nativestr
14			from glances.logger import logger
15			from glances.plugins.plugin.model import GlancesPluginModel
16
17			# Fields description
18			# description: human readable description
19			# short_name: shortname to use un UI
20			# unit: unit type
21			# rate: if True then compute and add _gauge and _rate_per_is fields
22			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
23			fields_description = {
24			'disk_name': {'description': 'Disk name.'},
25			'read_count': {
26			'description': 'Number of reads.',
27			'rate': True,
28			'unit': 'number',
29			},
30			'write_count': {
31			'description': 'Number of writes.',
32			'rate': True,
33			'unit': 'number',
34			},
35			'read_bytes': {
36			'description': 'Number of bytes read.',
37			'rate': True,
38			'unit': 'byte',
39			},
40			'write_bytes': {
41			'description': 'Number of bytes written.',
42			'rate': True,
43			'unit': 'byte',
44			},
45			'read_time': {
46			'description': 'Time spent reading.',
47			'rate': True,
48			'unit': 'millisecond',
49			},
50			'write_time': {
51			'description': 'Time spent writing.',
52			'rate': True,
53			'unit': 'millisecond',
54			},
55			'read_latency': {
56			'description': 'Mean time spent reading per operation.',
57			'unit': 'millisecond',
58			},
59			'write_latency': {
60			'description': 'Mean time spent writing per operation.',
61			'unit': 'millisecond',
62			},
63			}
64
65			# Define the history items list
66			items_history_list = [
67			{'name': 'read_bytes_rate_per_sec', 'description': 'Bytes read per second', 'y_unit': 'B/s'},
68			{'name': 'write_bytes_rate_per_sec', 'description': 'Bytes write per second', 'y_unit': 'B/s'},
69			]
70
71
72			class DiskioPlugin(GlancesPluginModel):
73			"""Glances disks I/O plugin.
74
75			stats is a list
76			"""
77
78			def __init__(self, args=None, config=None):
79			"""Init the plugin."""
80			super().__init__(
81			args=args,
82			config=config,
83			items_history_list=items_history_list,
84			stats_init_value=[],
85			fields_description=fields_description,
86			)
87
88			# We want to display the stat in the curse interface
89			self.display_curse = True
90
91			# Hide stats if it has never been != 0
92			if config is not None:
93			self.hide_zero = config.get_bool_value(self.plugin_name, 'hide_zero', default=False)
94			self.hide_threshold_bytes = config.get_int_value(self.plugin_name, 'hide_threshold_bytes', default=0)
95			else:
96			self.hide_zero = False
97			self.hide_threshold_bytes = 0
98			self.hide_zero_fields = ['read_bytes_rate_per_sec', 'write_bytes_rate_per_sec']
99
100			# Force a first update because we need two updates to have the first stat
101			self.update()
102			self.refresh_timer.set(0)
103
104			def get_key(self):
105			"""Return the key of the list."""
106			return 'disk_name'
107
108			@GlancesPluginModel._check_decorator
109			@GlancesPluginModel._log_result_decorator
110			def update(self):
111			"""Update disk I/O stats using the input method."""
112			# Update the stats
113			if self.input_method == 'local':
114			stats = self.update_local()
115
116			# Compute latency (need rate stats, so should be done after decorator)
117			stats = self.update_latency(stats)
118			else:
119			stats = self.get_init_value()
120
121			# Update the stats
122			self.stats = stats
123
124			return self.stats
125
126			def update_latency(self, stats):
127			"""Update the latency stats."""
128			# Compute read/write latency if we have the rate stats
129			for stat in stats:
130			# Compute read/write latency if we have the rate stats
131			if stat.get("read_count_rate_per_sec", 0) > 0:
132			stat["read_latency"] = int(stat["read_time_rate_per_sec"] / stat["read_count_rate_per_sec"])
133			else:
134			stat["read_latency"] = 0
135
136			if stat.get("write_count_rate_per_sec", 0) > 0:
137			stat["write_latency"] = int(stat["write_time_rate_per_sec"] / stat["write_count_rate_per_sec"])
138			else:
139			stat["write_latency"] = 0
140
141			return stats
142
143			@GlancesPluginModel._manage_rate
144			def update_local(self):
145			stats = self.get_init_value()
146
147			try:
148			diskio = psutil.disk_io_counters(perdisk=True)
149			except Exception:
150			return stats
151
152			for disk_name, disk_stat in diskio.items():
153			# By default, RamFS is not displayed (issue #714)
154			if self.args is not None and not self.args.diskio_show_ramfs and disk_name.startswith('ram'):
155			continue
156
157			# Shall we display the stats ?
158			if not self.is_display(disk_name):
159			continue
160
161			# Filter stats to keep only the fields we want (define in fields_description)
162			# It will also convert psutil objects to a standard Python dict
163			stat = self.filter_stats(disk_stat)
164
165			# Add the key
166			stat['key'] = self.get_key()
167
168			# Add disk name
169			stat['disk_name'] = disk_name
170
171			# Add alias if exist (define in the configuration file)
172			if self.has_alias(disk_name) is not None:
173			stat['alias'] = self.has_alias(disk_name)
174
175			stats.append(stat)
176
177			return stats
178
179			def update_views(self):
180			"""Update stats views."""
181			# Call the father's method
182			super().update_views()
183
184			# Alert
185			for i in self.get_raw():
186			disk_real_name = i['disk_name']
187
188			# # Skip alert if no timespan to measure
189			# if not i.get('read_bytes_rate_per_sec') or not i.get('write_bytes_rate_per_sec'):
190			# continue
191
192			# Decorate the bitrate with the configuration file
193			alert_rx = self.get_alert(i['read_bytes'], header=disk_real_name + '_rx')
194			alert_tx = self.get_alert(i['write_bytes'], header=disk_real_name + '_tx')
195			self.views[i[self.get_key()]]['read_bytes']['decoration'] = alert_rx
196			self.views[i[self.get_key()]]['write_bytes']['decoration'] = alert_tx
197
198			# Decorate the latency with the configuration file
199			# Try to get the read/write latency for the current disk
200			alert_latency_rx = self.get_alert(i['read_latency'], header=disk_real_name + '_rx_latency')
201			alert_latency_tx = self.get_alert(i['write_latency'], header=disk_real_name + '_tx_latency')
202			# If the alert is not defined, use the default one
203			if alert_latency_rx == 'DEFAULT':
204			alert_latency_rx = self.get_alert(i['read_latency'], header='rx_latency')
205			if alert_latency_tx == 'DEFAULT':
206			alert_latency_tx = self.get_alert(i['write_latency'], header='tx_latency')
207			self.views[i[self.get_key()]]['read_latency']['decoration'] = alert_latency_rx
208			self.views[i[self.get_key()]]['write_latency']['decoration'] = alert_latency_tx
209
210			def msg_curse(self, args=None, max_width=None):
211			"""Return the dict to display in the curse interface."""
212			# Init the return message
213			ret = []
214
215			# Only process if stats exist and display plugin enable...
216			if not self.stats or self.is_disabled():
217			return ret
218
219			# Max size for the interface name
220			if max_width:
221			name_max_width = max_width - 13
222			else:
223			# No max_width defined, return an empty curse message
224			logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
225			return ret
226
227			# Header
228			msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
229			ret.append(self.curse_add_line(msg, "TITLE"))
230			if args.diskio_iops:
231			msg = '{:>8}'.format('IOR/s')
232			ret.append(self.curse_add_line(msg))
233			msg = '{:>7}'.format('IOW/s')
234			ret.append(self.curse_add_line(msg))
235			elif args.diskio_latency:
236			msg = '{:>8}'.format('ms/opR')
237			ret.append(self.curse_add_line(msg))
238			msg = '{:>7}'.format('ms/opW')
239			ret.append(self.curse_add_line(msg))
240			else:
241			msg = '{:>8}'.format('R/s')
242			ret.append(self.curse_add_line(msg))
243			msg = '{:>7}'.format('W/s')
244			ret.append(self.curse_add_line(msg))
245			# Disk list (sorted by name)
246			for i in self.sorted_stats():
247			# Hide stats if never be different from 0 (issue #1787)
248			if all(self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields):
249			continue
250			# Is there an alias for the disk name ?
251			disk_name = i['alias'] if 'alias' in i else i['disk_name']
252			# New line
253			ret.append(self.curse_new_line())
254			if len(disk_name) > name_max_width:
255			# Cut disk name if it is too long
256			disk_name = disk_name[:name_max_width] + '_'
257			msg = '{:{width}}'.format(nativestr(disk_name), width=name_max_width + 1)
258			ret.append(self.curse_add_line(msg))
259			if args.diskio_iops:
260			# count
261			txps = self.auto_unit(i.get('read_count_rate_per_sec', None))
262			rxps = self.auto_unit(i.get('write_count_rate_per_sec', None))
263			msg = f'{txps:>7}'
264			ret.append(
265			self.curse_add_line(
266			msg, self.get_views(item=i[self.get_key()], key='read_count', option='decoration')
267			)
268			)
269			msg = f'{rxps:>7}'
270			ret.append(
271			self.curse_add_line(
272			msg, self.get_views(item=i[self.get_key()], key='write_count', option='decoration')
273			)
274			)
275			elif args.diskio_latency:
276			# latency (mean time spent reading/writing per operation)
277			txps = self.auto_unit(i.get('read_latency', None), low_precision=True)
278			rxps = self.auto_unit(i.get('write_latency', None), low_precision=True)
279			msg = f'{txps:>7}'
280			ret.append(
281			self.curse_add_line(
282			msg, self.get_views(item=i[self.get_key()], key='read_latency', option='decoration')
283			)
284			)
285			msg = f'{rxps:>7}'
286			ret.append(
287			self.curse_add_line(
288			msg, self.get_views(item=i[self.get_key()], key='write_latency', option='decoration')
289			)
290			)
291			else:
292			# Bitrate
293			txps = self.auto_unit(i.get('read_bytes_rate_per_sec', None))
294			rxps = self.auto_unit(i.get('write_bytes_rate_per_sec', None))
295			msg = f'{txps:>7}'
296			ret.append(
297			self.curse_add_line(
298			msg, self.get_views(item=i[self.get_key()], key='read_bytes', option='decoration')
299			)
300			)
301			msg = f'{rxps:>7}'
302			ret.append(
303			self.curse_add_line(
304			msg, self.get_views(item=i[self.get_key()], key='write_bytes', option='decoration')
305			)
306			)
307
308			return ret
309

nicolargo / glances

Push — develop ( 3f5c1f...948bc9 )

glances.plugins.diskio.DiskioPlugin.msg_curse() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like