glances.plugins.network.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.network.PluginModel.msg_curse() F

↳ Parent: glances.plugins.network

Complexity

Conditions

Size

Total Lines	105
Code Lines	64

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	18
eloc	64
nop	3
dl	0
loc	105
rs	1.2
c	0
b	0
f	0

How to fix Long Method Complexity

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

"""Network plugin."""
from __future__ import unicode_literals

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

import psutil

# 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 = {
    'interface_name': {'description': 'Interface name.'},
    'alias': {'description': 'Interface alias name (optional).'},
    'bytes_recv': {
        'description': 'Number of bytes received.',
        'rate': True,
        'unit': 'byte',
    },
    'bytes_sent': {
        'description': 'Number of bytes sent.',
        'rate': True,
        'unit': 'byte',
    },
    'bytes_all': {
        'description': 'Number of bytes received and sent.',
        'rate': True,
        'unit': 'byte',
    },
    'speed': {
        'description': 'Maximum interface speed (in bit per second). Can return 0 on some operating-system.',
        'unit': 'bitpersecond',
    },
    'is_up': {'description': 'Is the interface up ?', 'unit': 'bool'},
}

# SNMP OID
# http://www.net-snmp.org/docs/mibs/interfaces.html
# Dict key = interface_name
snmp_oid = {
    'default': {
        'interface_name': '1.3.6.1.2.1.2.2.1.2',
        'bytes_recv': '1.3.6.1.2.1.2.2.1.10',
        'bytes_sent': '1.3.6.1.2.1.2.2.1.16',
    }
}

# Define the history items list
items_history_list = [
    {'name': 'bytes_recv_rate_per_sec', 'description': 'Download rate per second', 'y_unit': 'B/s'},
    {'name': 'bytes_sent_rate_per_sec', 'description': 'Upload rate per second', 'y_unit': 'B/s'},
]


class PluginModel(GlancesPluginModel):
    """Glances network plugin.

    stats is a list
    """

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

        """Init the plugin."""
        super(PluginModel, self).__init__(
            args=args,
            config=config,
            items_history_list=items_history_list,
            fields_description=fields_description,
            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 = ['bytes_recv', 'bytes_sent']

        # 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 'interface_name'

    # @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update network stats using the input method.

        :return: list of stats dict (one dict per interface)
        """
        if self.input_method == 'local':
            stats = self.update_local()
        else:
            stats = self.get_init_value()

        # Update the stats
        self.stats = stats

        return self.stats

    @GlancesPluginModel._manage_rate
    def update_local(self):
        # Update stats using the standard system lib

        stats = self.get_init_value()

        # Grab network interface stat using the psutil net_io_counter method
        # Example:
        # { 'veth4cbf8f0a': snetio(
        #   bytes_sent=102038421, bytes_recv=1263258,
        #   packets_sent=25046, packets_recv=14114,
        #   errin=0, errout=0, dropin=0, dropout=0), ... }
        try:
            net_io_counters = psutil.net_io_counters(pernic=True)
        except UnicodeDecodeError as e:
            logger.debug('Can not get network interface counters ({})'.format(e))
            return self.stats

        # Grab interface's status (issue #765)
        # Grab interface's speed (issue #718)
        # Example:
        # { 'veth4cbf8f0a': snicstats(
        #   isup=True, duplex=<NicDuplex.NIC_DUPLEX_FULL: 2>,
        #   speed=10000, mtu=1500, flags='up,broadcast,running,multicast'), ... }
        net_status = {}
        try:
            net_status = psutil.net_if_stats()
        except OSError as e:
            # see psutil #797/glances #1106
            logger.debug('Can not get network interface status ({})'.format(e))

        for interface_name, interface_stat in net_io_counters.items():
            # Do not take hidden interface into account
            # or KeyError: 'eth0' when interface is not connected #1348
            if not self.is_display(interface_name) or interface_name not in net_status:
                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(interface_stat)
            stat.update(self.filter_stats(net_status[interface_name]))

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

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

            # Add alias define in the configuration file
            stat['alias'] = self.has_alias(interface_name)

            # Add sent + revc  stats
            stat['bytes_all'] = stat['bytes_sent'] + stat['bytes_recv']

            # Interface speed in Mbps, convert it to bps
            # Can be always 0 on some OSes
            stat['speed'] = stat['speed'] * 1048576

            stats.append(stat)

        return stats

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

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

        # Add specifics information
        # Alert
        for i in self.get_raw():
            # Skip alert if no timespan to measure
            if 'bytes_recv_rate_per_sec' not in i or 'bytes_sent_rate_per_sec' not in i:
                continue

            # Convert rate to bps (to be able to compare to interface speed)
            bps_rx = int(i['bytes_recv_rate_per_sec'] * 8)
            bps_tx = int(i['bytes_sent_rate_per_sec'] * 8)

            # Decorate the bitrate with the configuration file thresholds
            if_real_name = i['interface_name'].split(':')[0]
            alert_rx = self.get_alert(bps_rx, header=if_real_name + '_rx')
            alert_tx = self.get_alert(bps_tx, header=if_real_name + '_tx')

            # If nothing is define in the configuration file...
            # ... then use the interface speed (not available on all systems)
            if alert_rx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
                alert_rx = self.get_alert(current=bps_rx, maximum=i['speed'], header='rx')
            if alert_tx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
                alert_tx = self.get_alert(current=bps_tx, maximum=i['speed'], header='tx')

            # then decorates
            self.views[i[self.get_key()]]['bytes_recv']['decoration'] = alert_rx
            self.views[i[self.get_key()]]['bytes_sent']['decoration'] = alert_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 - 12
        else:
            # No max_width defined, return an emptu curse message
            logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
            return ret

        # Header
        msg = '{:{width}}'.format('NETWORK', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.network_cumul:
            # Cumulative stats
            if args.network_sum:
                # Sum stats
                msg = '{:>14}'.format('Rx+Tx')
                ret.append(self.curse_add_line(msg))
            else:
                # Rx/Tx stats
                msg = '{:>7}'.format('Rx')
                ret.append(self.curse_add_line(msg))
                msg = '{:>7}'.format('Tx')
                ret.append(self.curse_add_line(msg))
        else:
            # Bitrate stats
            if args.network_sum:
                # Sum stats
                msg = '{:>14}'.format('Rx+Tx/s')
                ret.append(self.curse_add_line(msg))
            else:
                msg = '{:>7}'.format('Rx/s')
                ret.append(self.curse_add_line(msg))
                msg = '{:>7}'.format('Tx/s')
                ret.append(self.curse_add_line(msg))
        # Interface list (sorted by name)
        for i in self.sorted_stats():
            # Do not display interface in down state (issue #765)
            if ('is_up' in i) and (i['is_up'] is False):
                continue
            # 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
            # Format stats
            # Is there an alias for the interface name ?
            if i['alias'] is None:
                if_name = i['interface_name'].split(':')[0]
            else:
                if_name = i['alias']
            if len(if_name) > name_max_width:
                # Cut interface name if it is too long
                if_name = '_' + if_name[-name_max_width + 1 :]

            if args.byte:
                # Bytes per second (for dummy)
                to_bit = 1
                unit = ''
            else:
                # Bits per second (for real network administrator | Default)
                to_bit = 8
                unit = 'b'

            if args.network_cumul and 'bytes_recv' in i:
                rx = self.auto_unit(int(i['bytes_recv'] * to_bit)) + unit
                tx = self.auto_unit(int(i['bytes_sent'] * to_bit)) + unit
                ax = self.auto_unit(int(i['bytes_all'] * to_bit)) + unit
            elif 'bytes_recv_rate_per_sec' in i:
                rx = self.auto_unit(int(i['bytes_recv_rate_per_sec'] * to_bit)) + unit
                tx = self.auto_unit(int(i['bytes_sent_rate_per_sec'] * to_bit)) + unit
                ax = self.auto_unit(int(i['bytes_all_rate_per_sec'] * to_bit)) + unit
            else:
                # Avoid issue when a new interface is created on the fly
                # Example: start Glances, then start a new container
                continue

            # New line
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(if_name, width=name_max_width)
            ret.append(self.curse_add_line(msg))
            if args.network_sum:
                msg = '{:>14}'.format(ax)
                ret.append(self.curse_add_line(msg))
            else:
                msg = '{:>7}'.format(rx)
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='bytes_recv', option='decoration')
                    )
                )
                msg = '{:>7}'.format(tx)
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='bytes_sent', option='decoration')
                    )
                )

        return ret


1		# -- coding: utf-8 --
2		#
3		# This file is part of Glances.
4		#
5		# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
6		#
7		# SPDX-License-Identifier: LGPL-3.0-only
8		#
9
10		"""Network plugin."""
11		from __future__ import unicode_literals
12
13		from glances.plugins.plugin.model import GlancesPluginModel
14		from glances.logger import logger
15
16		import psutil
17
18		# Fields description
19		# description: human readable description
20		# short_name: shortname to use un UI
21		# unit: unit type
22		# rate: if True then compute and add _gauge and _rate_per_is fields
23		# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
24		fields_description = {
25		'interface_name': {'description': 'Interface name.'},
26		'alias': {'description': 'Interface alias name (optional).'},
27		'bytes_recv': {
28		'description': 'Number of bytes received.',
29		'rate': True,
30		'unit': 'byte',
31		},
32		'bytes_sent': {
33		'description': 'Number of bytes sent.',
34		'rate': True,
35		'unit': 'byte',
36		},
37		'bytes_all': {
38		'description': 'Number of bytes received and sent.',
39		'rate': True,
40		'unit': 'byte',
41		},
42		'speed': {
43		'description': 'Maximum interface speed (in bit per second). Can return 0 on some operating-system.',
44		'unit': 'bitpersecond',
45		},
46		'is_up': {'description': 'Is the interface up ?', 'unit': 'bool'},
47		}
48
49		# SNMP OID
50		# http://www.net-snmp.org/docs/mibs/interfaces.html
51		# Dict key = interface_name
52		snmp_oid = {
53		'default': {
54		'interface_name': '1.3.6.1.2.1.2.2.1.2',
55		'bytes_recv': '1.3.6.1.2.1.2.2.1.10',
56		'bytes_sent': '1.3.6.1.2.1.2.2.1.16',
57		}
58		}
59
60		# Define the history items list
61		items_history_list = [
62		{'name': 'bytes_recv_rate_per_sec', 'description': 'Download rate per second', 'y_unit': 'B/s'},
63		{'name': 'bytes_sent_rate_per_sec', 'description': 'Upload rate per second', 'y_unit': 'B/s'},
64		]
65
66
67		class PluginModel(GlancesPluginModel):
68		"""Glances network plugin.
69
70		stats is a list
71		"""
72
73	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...
74		"""Init the plugin."""
75		super(PluginModel, self).__init__(
76		args=args,
77		config=config,
78		items_history_list=items_history_list,
79		fields_description=fields_description,
80		stats_init_value=[],
81		)
82
83		# We want to display the stat in the curse interface
84		self.display_curse = True
85
86		# Hide stats if it has never been != 0
87		if config is not None:
88		self.hide_zero = config.get_bool_value(self.plugin_name, 'hide_zero', default=False)
89		else:
90		self.hide_zero = False
91		self.hide_zero_fields = ['bytes_recv', 'bytes_sent']
92
93		# Force a first update because we need two updates to have the first stat
94		self.update()
95		self.refresh_timer.set(0)
96
97		def get_key(self):
98		"""Return the key of the list."""
99		return 'interface_name'
100
101		# @GlancesPluginModel._check_decorator
102		@GlancesPluginModel._log_result_decorator
103		def update(self):
104		"""Update network stats using the input method.
105
106		:return: list of stats dict (one dict per interface)
107		"""
108		if self.input_method == 'local':
109		stats = self.update_local()
110		else:
111		stats = self.get_init_value()
112
113		# Update the stats
114		self.stats = stats
115
116		return self.stats
117
118		@GlancesPluginModel._manage_rate
119		def update_local(self):
120		# Update stats using the standard system lib
121
122		stats = self.get_init_value()
123
124		# Grab network interface stat using the psutil net_io_counter method
125		# Example:
126		# { 'veth4cbf8f0a': snetio(
127		# bytes_sent=102038421, bytes_recv=1263258,
128		# packets_sent=25046, packets_recv=14114,
129		# errin=0, errout=0, dropin=0, dropout=0), ... }
130		try:
131		net_io_counters = psutil.net_io_counters(pernic=True)
132		except UnicodeDecodeError as e:
133		logger.debug('Can not get network interface counters ({})'.format(e))
134		return self.stats
135
136		# Grab interface's status (issue #765)
137		# Grab interface's speed (issue #718)
138		# Example:
139		# { 'veth4cbf8f0a': snicstats(
140		# isup=True, duplex=<NicDuplex.NIC_DUPLEX_FULL: 2>,
141		# speed=10000, mtu=1500, flags='up,broadcast,running,multicast'), ... }
142		net_status = {}
143		try:
144		net_status = psutil.net_if_stats()
145		except OSError as e:
146		# see psutil #797/glances #1106
147		logger.debug('Can not get network interface status ({})'.format(e))
148
149		for interface_name, interface_stat in net_io_counters.items():
150		# Do not take hidden interface into account
151		# or KeyError: 'eth0' when interface is not connected #1348
152		if not self.is_display(interface_name) or interface_name not in net_status:
153		continue
154
155		# Filter stats to keep only the fields we want (define in fields_description)
156		# It will also convert psutil objects to a standard Python dict
157		stat = self.filter_stats(interface_stat)
158		stat.update(self.filter_stats(net_status[interface_name]))
159
160		# Add the key
161		stat['key'] = self.get_key()
162
163		# Add disk name
164		stat['interface_name'] = interface_name
165
166		# Add alias define in the configuration file
167		stat['alias'] = self.has_alias(interface_name)
168
169		# Add sent + revc stats
170		stat['bytes_all'] = stat['bytes_sent'] + stat['bytes_recv']
171
172		# Interface speed in Mbps, convert it to bps
173		# Can be always 0 on some OSes
174		stat['speed'] = stat['speed'] * 1048576
175
176		stats.append(stat)
177
178		return stats
179
180		def update_views(self):
181		"""Update stats views."""
182		# Call the father's method
183		super(PluginModel, self).update_views()
184
185		# Check if the stats should be hidden
186		self.update_views_hidden()
187
188		# Add specifics information
189		# Alert
190		for i in self.get_raw():
191		# Skip alert if no timespan to measure
192		if 'bytes_recv_rate_per_sec' not in i or 'bytes_sent_rate_per_sec' not in i:
193		continue
194
195		# Convert rate to bps (to be able to compare to interface speed)
196		bps_rx = int(i['bytes_recv_rate_per_sec'] * 8)
197		bps_tx = int(i['bytes_sent_rate_per_sec'] * 8)
198
199		# Decorate the bitrate with the configuration file thresholds
200		if_real_name = i['interface_name'].split(':')[0]
201		alert_rx = self.get_alert(bps_rx, header=if_real_name + '_rx')
202		alert_tx = self.get_alert(bps_tx, header=if_real_name + '_tx')
203
204		# If nothing is define in the configuration file...
205		# ... then use the interface speed (not available on all systems)
206		if alert_rx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
207		alert_rx = self.get_alert(current=bps_rx, maximum=i['speed'], header='rx')
208		if alert_tx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
209		alert_tx = self.get_alert(current=bps_tx, maximum=i['speed'], header='tx')
210
211		# then decorates
212		self.views[i[self.get_key()]]['bytes_recv']['decoration'] = alert_rx
213		self.views[i[self.get_key()]]['bytes_sent']['decoration'] = alert_tx
214
215		def msg_curse(self, args=None, max_width=None):
216		"""Return the dict to display in the curse interface."""
217		# Init the return message
218		ret = []
219
220		# Only process if stats exist and display plugin enable...
221		if not self.stats or self.is_disabled():
222		return ret
223
224		# Max size for the interface name
225		if max_width:
226		name_max_width = max_width - 12
227		else:
228		# No max_width defined, return an emptu curse message
229		logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
230		return ret
231
232		# Header
233		msg = '{:{width}}'.format('NETWORK', width=name_max_width)
234		ret.append(self.curse_add_line(msg, "TITLE"))
235		if args.network_cumul:
236		# Cumulative stats
237		if args.network_sum:
238		# Sum stats
239		msg = '{:>14}'.format('Rx+Tx')
240		ret.append(self.curse_add_line(msg))
241		else:
242		# Rx/Tx stats
243		msg = '{:>7}'.format('Rx')
244		ret.append(self.curse_add_line(msg))
245		msg = '{:>7}'.format('Tx')
246		ret.append(self.curse_add_line(msg))
247		else:
248		# Bitrate stats
249		if args.network_sum:
250		# Sum stats
251		msg = '{:>14}'.format('Rx+Tx/s')
252		ret.append(self.curse_add_line(msg))
253		else:
254		msg = '{:>7}'.format('Rx/s')
255		ret.append(self.curse_add_line(msg))
256		msg = '{:>7}'.format('Tx/s')
257		ret.append(self.curse_add_line(msg))
258		# Interface list (sorted by name)
259		for i in self.sorted_stats():
260		# Do not display interface in down state (issue #765)
261		if ('is_up' in i) and (i['is_up'] is False):
262		continue
263		# Hide stats if never be different from 0 (issue #1787)
264		if all([self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields]):
265		continue
266		# Format stats
267		# Is there an alias for the interface name ?
268		if i['alias'] is None:
269		if_name = i['interface_name'].split(':')[0]
270		else:
271		if_name = i['alias']
272		if len(if_name) > name_max_width:
273		# Cut interface name if it is too long
274		if_name = '_' + if_name[-name_max_width + 1 :]
275
276		if args.byte:
277		# Bytes per second (for dummy)
278		to_bit = 1
279		unit = ''
280		else:
281		# Bits per second (for real network administrator \| Default)
282		to_bit = 8
283		unit = 'b'
284
285		if args.network_cumul and 'bytes_recv' in i:
286		rx = self.auto_unit(int(i['bytes_recv'] * to_bit)) + unit
287		tx = self.auto_unit(int(i['bytes_sent'] * to_bit)) + unit
288		ax = self.auto_unit(int(i['bytes_all'] * to_bit)) + unit
289		elif 'bytes_recv_rate_per_sec' in i:
290		rx = self.auto_unit(int(i['bytes_recv_rate_per_sec'] * to_bit)) + unit
291		tx = self.auto_unit(int(i['bytes_sent_rate_per_sec'] * to_bit)) + unit
292		ax = self.auto_unit(int(i['bytes_all_rate_per_sec'] * to_bit)) + unit
293		else:
294		# Avoid issue when a new interface is created on the fly
295		# Example: start Glances, then start a new container
296		continue
297
298		# New line
299		ret.append(self.curse_new_line())
300		msg = '{:{width}}'.format(if_name, width=name_max_width)
301		ret.append(self.curse_add_line(msg))
302		if args.network_sum:
303		msg = '{:>14}'.format(ax)
304		ret.append(self.curse_add_line(msg))
305		else:
306		msg = '{:>7}'.format(rx)
307		ret.append(
308		self.curse_add_line(
309		msg, self.get_views(item=i[self.get_key()], key='bytes_recv', option='decoration')
310		)
311		)
312		msg = '{:>7}'.format(tx)
313		ret.append(
314		self.curse_add_line(
315		msg, self.get_views(item=i[self.get_key()], key='bytes_sent', option='decoration')
316		)
317		)
318
319		return ret
320

nicolargo / glances

Push — master ( ee826a...d9056e )

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

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like