glances.plugins.glances_diskio.Plugin.msg_curse() - Code Metrics - Inspection of "version 3.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 69b639...e7fa0a )

by Nicolas

created 2021-07-03 08:40 UTC

glances.plugins.glances_diskio.Plugin.msg_curse() C

↳ Parent: glances.plugins.glances_diskio

Complexity

Conditions

Size

Total Lines	77
Code Lines	59

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	59
nop	3
dl	0
loc	77
rs	6.0084
c	0
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""Disk I/O plugin."""
from __future__ import unicode_literals

from glances.compat import nativestr, n
from glances.timer import getTimeSinceLastUpdate
from glances.plugins.glances_plugin import GlancesPlugin
from glances.logger import logger

import psutil


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


class Plugin(GlancesPlugin):
    """Glances disks I/O plugin.

    stats is a list
    """

    def __init__(self, args=None, config=None):

        """Init the plugin."""
        super(Plugin, self).__init__(args=args,
                                     config=config,
                                     items_history_list=items_history_list,
                                     stats_init_value=[])

        # 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)
        else:
            self.hide_zero = False
        self.hide_zero_fields = ['read_bytes', 'write_bytes']

        # Force a first update because we need two update 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'

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update disk I/O stats using the input method."""
        # Init new stats
        stats = self.get_init_value()

        if self.input_method == 'local':
            # Update stats using the standard system lib
            # Grab the stat using the psutil disk_io_counters method
            # read_count: number of reads
            # write_count: number of writes
            # read_bytes: number of bytes read
            # write_bytes: number of bytes written
            # read_time: time spent reading from disk (in milliseconds)
            # write_time: time spent writing to disk (in milliseconds)
            try:
                diskio = psutil.disk_io_counters(perdisk=True)
            except Exception:
                return stats

            # Previous disk IO stats are stored in the diskio_old variable
            # By storing time data we enable Rx/s and Tx/s calculations in the
            # XML/RPC API, which would otherwise be overly difficult work
            # for users of the API
            time_since_update = getTimeSinceLastUpdate('disk')

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

                # Do not take hide disk into account
                if self.is_hide(disk):
                    continue

                # Compute count and bit rate
                try:
                    diskstat = {
                        'time_since_update': time_since_update,
                        'disk_name': n(disk),
                        'read_count': diskio[disk].read_count - \
                                      self.diskio_old[disk].read_count,
                        'write_count': diskio[disk].write_count - \
                                       self.diskio_old[disk].write_count,
                        'read_bytes': diskio[disk].read_bytes - \
                                      self.diskio_old[disk].read_bytes,
                        'write_bytes': diskio[disk].write_bytes - \
                                       self.diskio_old[disk].write_bytes
                    }
                except (KeyError, AttributeError):
                    diskstat = {
                        'time_since_update': time_since_update,
                        'disk_name': n(disk),
                        'read_count': 0,
                        'write_count': 0,
                        'read_bytes': 0,
                        'write_bytes': 0}

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

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

                # Ad dthe current disk stat to the list
                stats.append(diskstat)

            # Save stats to compute next bitrate
            try:
                self.diskio_old = diskio
            except (IOError, UnboundLocalError):
                pass
        elif self.input_method == 'snmp':
            # Update stats using SNMP
            # No standard way for the moment...
            pass

        # Update the stats
        self.stats = stats

        return self.stats

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

        # Check if the stats should be hidden
        self.update_views_hidden()

        # Add specifics informations
        # Alert
        for i in self.get_raw():
            disk_real_name = i['disk_name']
            self.views[i[self.get_key()]]['read_bytes']['decoration'] = self.get_alert(int(i['read_bytes'] // i['time_since_update']),
                                                                                       header=disk_real_name + '_rx')
            self.views[i[self.get_key()]]['write_bytes']['decoration'] = self.get_alert(int(i['write_bytes'] // i['time_since_update']),
                                                                                        header=disk_real_name + '_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_disable():
            return ret

        # Max size for the interface name
        name_max_width = max_width - 13

        # Header
        msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.diskio_iops:
            msg = '{:>7}'.format('IOR/s')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('IOW/s')
            ret.append(self.curse_add_line(msg))
        else:
            msg = '{:>7}'.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_real_name = i['disk_name']
            disk_name = self.has_alias(i['disk_name'])
            if disk_name is None:
                disk_name = disk_real_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+1:]
            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(
                    int(i['read_count'] // i['time_since_update']))
                rxps = self.auto_unit(
                    int(i['write_count'] // i['time_since_update']))
                msg = '{:>7}'.format(txps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='read_count',
                                                              option='decoration')))
                msg = '{:>7}'.format(rxps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='write_count',
                                                              option='decoration')))
            else:
                # Bitrate
                txps = self.auto_unit(
                    int(i['read_bytes'] // i['time_since_update']))
                rxps = self.auto_unit(
                    int(i['write_bytes'] // i['time_since_update']))
                msg = '{:>7}'.format(txps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='read_bytes',
                                                              option='decoration')))
                msg = '{:>7}'.format(rxps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='write_bytes',
                                                              option='decoration')))

        return ret


1		# -- coding: utf-8 --
2		#
3		# This file is part of Glances.
4		#
5		# Copyright (C) 2019 Nicolargo <[email protected]>
6		#
7		# Glances is free software; you can redistribute it and/or modify
8		# it under the terms of the GNU Lesser General Public License as published by
9		# the Free Software Foundation, either version 3 of the License, or
10		# (at your option) any later version.
11		#
12		# Glances is distributed in the hope that it will be useful,
13		# but WITHOUT ANY WARRANTY; without even the implied warranty of
14		# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15		# GNU Lesser General Public License for more details.
16		#
17		# You should have received a copy of the GNU Lesser General Public License
18		# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20		"""Disk I/O plugin."""
21		from __future__ import unicode_literals
22
23		from glances.compat import nativestr, n
24		from glances.timer import getTimeSinceLastUpdate
25		from glances.plugins.glances_plugin import GlancesPlugin
26		from glances.logger import logger
27
28		import psutil
29
30
31		# Define the history items list
32		items_history_list = [{'name': 'read_bytes',
33		'description': 'Bytes read per second',
34		'y_unit': 'B/s'},
35		{'name': 'write_bytes',
36		'description': 'Bytes write per second',
37		'y_unit': 'B/s'}]
38
39
40		class Plugin(GlancesPlugin):
41		"""Glances disks I/O plugin.
42
43		stats is a list
44		"""
45
46	View Code Duplication	def __init__(self, args=None, config=None):
		0 ignored issues – show Duplication introduced 2019-08-17 09:24 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
47		"""Init the plugin."""
48		super(Plugin, self).__init__(args=args,
49		config=config,
50		items_history_list=items_history_list,
51		stats_init_value=[])
52
53		# We want to display the stat in the curse interface
54		self.display_curse = True
55
56		# Hide stats if it has never been != 0
57		if config is not None:
58		self.hide_zero = config.get_bool_value(
59		self.plugin_name, 'hide_zero', default=False)
60		else:
61		self.hide_zero = False
62		self.hide_zero_fields = ['read_bytes', 'write_bytes']
63
64		# Force a first update because we need two update to have the first stat
65		self.update()
66		self.refresh_timer.set(0)
67
68		def get_key(self):
69		"""Return the key of the list."""
70		return 'disk_name'
71
72		@GlancesPlugin._check_decorator
73		@GlancesPlugin._log_result_decorator
74		def update(self):
75		"""Update disk I/O stats using the input method."""
76		# Init new stats
77		stats = self.get_init_value()
78
79		if self.input_method == 'local':
80		# Update stats using the standard system lib
81		# Grab the stat using the psutil disk_io_counters method
82		# read_count: number of reads
83		# write_count: number of writes
84		# read_bytes: number of bytes read
85		# write_bytes: number of bytes written
86		# read_time: time spent reading from disk (in milliseconds)
87		# write_time: time spent writing to disk (in milliseconds)
88		try:
89		diskio = psutil.disk_io_counters(perdisk=True)
90		except Exception:
91		return stats
92
93		# Previous disk IO stats are stored in the diskio_old variable
94		# By storing time data we enable Rx/s and Tx/s calculations in the
95		# XML/RPC API, which would otherwise be overly difficult work
96		# for users of the API
97		time_since_update = getTimeSinceLastUpdate('disk')
98
99		diskio = diskio
100		for disk in diskio:
101		# By default, RamFS is not displayed (issue #714)
102		if self.args is not None and not self.args.diskio_show_ramfs and disk.startswith('ram'):
103		continue
104
105		# Do not take hide disk into account
106		if self.is_hide(disk):
107		continue
108
109		# Compute count and bit rate
110		try:
111		diskstat = {
112		'time_since_update': time_since_update,
113		'disk_name': n(disk),
114		'read_count': diskio[disk].read_count - \
115		self.diskio_old[disk].read_count,
116		'write_count': diskio[disk].write_count - \
117		self.diskio_old[disk].write_count,
118		'read_bytes': diskio[disk].read_bytes - \
119		self.diskio_old[disk].read_bytes,
120		'write_bytes': diskio[disk].write_bytes - \
121		self.diskio_old[disk].write_bytes
122		}
123		except (KeyError, AttributeError):
124		diskstat = {
125		'time_since_update': time_since_update,
126		'disk_name': n(disk),
127		'read_count': 0,
128		'write_count': 0,
129		'read_bytes': 0,
130		'write_bytes': 0}
131
132		# Add alias if exist (define in the configuration file)
133		if self.has_alias(disk) is not None:
134		diskstat['alias'] = self.has_alias(disk)
135
136		# Add the dict key
137		diskstat['key'] = self.get_key()
138
139		# Ad dthe current disk stat to the list
140		stats.append(diskstat)
141
142		# Save stats to compute next bitrate
143		try:
144		self.diskio_old = diskio
145		except (IOError, UnboundLocalError):
146		pass
147		elif self.input_method == 'snmp':
148		# Update stats using SNMP
149		# No standard way for the moment...
150		pass
151
152		# Update the stats
153		self.stats = stats
154
155		return self.stats
156
157		def update_views(self):
158		"""Update stats views."""
159		# Call the father's method
160		super(Plugin, self).update_views()
161
162		# Check if the stats should be hidden
163		self.update_views_hidden()
164
165		# Add specifics informations
166		# Alert
167		for i in self.get_raw():
168		disk_real_name = i['disk_name']
169		self.views[i[self.get_key()]]['read_bytes']['decoration'] = self.get_alert(int(i['read_bytes'] // i['time_since_update']),
170		header=disk_real_name + '_rx')
171		self.views[i[self.get_key()]]['write_bytes']['decoration'] = self.get_alert(int(i['write_bytes'] // i['time_since_update']),
172		header=disk_real_name + '_tx')
173
174		def msg_curse(self, args=None, max_width=None):
175		"""Return the dict to display in the curse interface."""
176		# Init the return message
177		ret = []
178
179		# Only process if stats exist and display plugin enable...
180		if not self.stats or self.is_disable():
181		return ret
182
183		# Max size for the interface name
184		name_max_width = max_width - 13
185
186		# Header
187		msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
188		ret.append(self.curse_add_line(msg, "TITLE"))
189		if args.diskio_iops:
190		msg = '{:>7}'.format('IOR/s')
191		ret.append(self.curse_add_line(msg))
192		msg = '{:>7}'.format('IOW/s')
193		ret.append(self.curse_add_line(msg))
194		else:
195		msg = '{:>7}'.format('R/s')
196		ret.append(self.curse_add_line(msg))
197		msg = '{:>7}'.format('W/s')
198		ret.append(self.curse_add_line(msg))
199		# Disk list (sorted by name)
200		for i in self.sorted_stats():
201		# Hide stats if never be different from 0 (issue #1787)
202		if all([self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields]):
203		continue
204		# Is there an alias for the disk name ?
205		disk_real_name = i['disk_name']
206		disk_name = self.has_alias(i['disk_name'])
207		if disk_name is None:
208		disk_name = disk_real_name
209		# New line
210		ret.append(self.curse_new_line())
211		if len(disk_name) > name_max_width:
212		# Cut disk name if it is too long
213		disk_name = '_' + disk_name[-name_max_width+1:]
214		msg = '{:{width}}'.format(nativestr(disk_name),
215		width=name_max_width+1)
216		ret.append(self.curse_add_line(msg))
217		if args.diskio_iops:
218		# count
219		txps = self.auto_unit(
220		int(i['read_count'] // i['time_since_update']))
221		rxps = self.auto_unit(
222		int(i['write_count'] // i['time_since_update']))
223		msg = '{:>7}'.format(txps)
224		ret.append(self.curse_add_line(msg,
225		self.get_views(item=i[self.get_key()],
226		key='read_count',
227		option='decoration')))
228		msg = '{:>7}'.format(rxps)
229		ret.append(self.curse_add_line(msg,
230		self.get_views(item=i[self.get_key()],
231		key='write_count',
232		option='decoration')))
233		else:
234		# Bitrate
235		txps = self.auto_unit(
236		int(i['read_bytes'] // i['time_since_update']))
237		rxps = self.auto_unit(
238		int(i['write_bytes'] // i['time_since_update']))
239		msg = '{:>7}'.format(txps)
240		ret.append(self.curse_add_line(msg,
241		self.get_views(item=i[self.get_key()],
242		key='read_bytes',
243		option='decoration')))
244		msg = '{:>7}'.format(rxps)
245		ret.append(self.curse_add_line(msg,
246		self.get_views(item=i[self.get_key()],
247		key='write_bytes',
248		option='decoration')))
249
250		return ret
251

nicolargo / glances

Push — master ( 69b639...e7fa0a )

glances.plugins.glances_diskio.Plugin.msg_curse() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like