glances.plugins.wifi.ThreadHotspot.stopped() - Code Metrics - Inspection of "Refactor the Wifi plugin" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 6b0de5...610ee0 )

by Nicolas

created 2024-04-28 09:16 UTC

glances.plugins.wifi.ThreadHotspot.stopped() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	3
rs	10
c	0
b	0
f	0

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

"""Wifi plugin.

Stats are retreived from the /proc/net/wireless file (Linux only):

# cat /proc/net/wireless
Inter-| sta-|   Quality        |   Discarded packets               | Missed | WE
 face | tus | link level noise |  nwid  crypt   frag  retry   misc | beacon | 22
wlp2s0: 0000   51.  -59.  -256        0      0      0      0   5881        0
"""

import operator

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

# Backup solution is to use the /proc/net/wireless file
# but it only give signal information about the current hotspot
WIRELESS_FILE = '/proc/net/wireless'
wireless_file_exists = file_exists(WIRELESS_FILE)

if not wireless_file_exists:
    logger.debug("Wifi plugin is disabled (no %s file found)" % (WIRELESS_FILE))

# 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 = {
    'ssid': {
        'description': 'Wi-Fi network name.'
    },
    'quality_link': {
        'description': 'Signal quality level.',
        'unit': 'dBm',
    },
    'quality_level': {
        'description': 'Signal strong level.',
        'unit': 'dBm',
    },
}


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

    Get stats of the current Wifi hotspots.
    """

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

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

        # Global Thread running all the scans
        self._thread = None

    def exit(self):
        """Overwrite the exit method to close threads."""
        if self._thread is not None:
            self._thread.stop()
        # Call the father class
        super(PluginModel, self).exit()

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

        :returns: string -- SSID is the dict key
        """
        return 'ssid'

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

        Stats is a list of dict (one dict per hotspot)

        :returns: list -- Stats is a list of dict (hotspot)
        """
        # Init new stats
        stats = self.get_init_value()

        # Exist if we can not grab the stats
        if not wireless_file_exists:
            return stats

        if self.input_method == 'local' and wireless_file_exists:
            # As a backup solution, use the /proc/net/wireless file
            with open(WIRELESS_FILE, 'r') as f:
                # The first two lines are header
                f.readline()
                f.readline()
                # Others lines are Wifi stats
                wifi_stats = f.readline()
                while wifi_stats != '':
                    # Extract the stats
                    wifi_stats = wifi_stats.split()
                    # Add the Wifi link to the list
                    stats.append(
                        {
                            'key': self.get_key(),
                            'ssid': wifi_stats[0][:-1],
                            'quality_link': float(wifi_stats[2]),
                            'quality_level': float(wifi_stats[3]),
                        }
                    )
                    # Next line
                    wifi_stats = f.readline()

        elif self.input_method == 'snmp':
            # Update stats using SNMP

            # Not implemented yet
            pass

        # Update the stats
        self.stats = stats

        return self.stats

    def get_alert(self, value):
        """Overwrite the default get_alert method.

        Alert is on signal quality where lower is better...

        :returns: string -- Signal alert
        """
        ret = 'OK'
        try:
            if value <= self.get_limit('critical', stat_name=self.plugin_name):
                ret = 'CRITICAL'
            elif value <= self.get_limit('warning', stat_name=self.plugin_name):
                ret = 'WARNING'
            elif value <= self.get_limit('careful', stat_name=self.plugin_name):
                ret = 'CAREFUL'
        except (TypeError, KeyError):
            # Catch TypeError for issue1373
            ret = 'DEFAULT'

        return ret

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

        # Add specifics information
        # Alert on quality_level thresholds
        for i in self.stats:
            self.views[i[self.get_key()]]['quality_level']['decoration'] = self.get_alert(i['quality_level'])

    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 not wireless_file_exists or self.is_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            if_name_max_width = max_width - 5
        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

        # Build the string message
        # Header
        msg = '{:{width}}'.format('WIFI', width=if_name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = '{:>7}'.format('dBm')
        ret.append(self.curse_add_line(msg))

        # Hotspot list (sorted by name)
        for i in sorted(self.stats, key=operator.itemgetter(self.get_key())):
            # Do not display hotspot with no name (/ssid)...
            # of ssid/signal None... See issue #1151 and #issue1973
            if i['ssid'] == '' or i['ssid'] is None or i['quality_level'] is None:
                continue
            ret.append(self.curse_new_line())
            # New hotspot
            hotspot_name = i['ssid']
            msg = '{:{width}}'.format(
                nativestr(hotspot_name),
                width=if_name_max_width
            )
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format(
                i['quality_level'],
            )
            ret.append(
                self.curse_add_line(msg, self.get_views(item=i[self.get_key()],
                                                        key='quality_level',
                                                        option='decoration'))
            )

        return ret


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# SPDX-FileCopyrightText: 2023 Nicolas Hennion <[email protected]>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8			#
9
10			"""Wifi plugin.
11
12			Stats are retreived from the /proc/net/wireless file (Linux only):
13
14			# cat /proc/net/wireless
15			Inter-\| sta-\| Quality \| Discarded packets \| Missed \| WE
16			face \| tus \| link level noise \| nwid crypt frag retry misc \| beacon \| 22
17			wlp2s0: 0000 51. -59. -256 0 0 0 0 5881 0
18			"""
19
20			import operator
21
22			from glances.globals import nativestr, file_exists
23			from glances.plugins.plugin.model import GlancesPluginModel
24			from glances.logger import logger
25
26			# Backup solution is to use the /proc/net/wireless file
27			# but it only give signal information about the current hotspot
28			WIRELESS_FILE = '/proc/net/wireless'
29			wireless_file_exists = file_exists(WIRELESS_FILE)
30
31			if not wireless_file_exists:
32			logger.debug("Wifi plugin is disabled (no %s file found)" % (WIRELESS_FILE))
33
34			# Fields description
35			# description: human readable description
36			# short_name: shortname to use un UI
37			# unit: unit type
38			# rate: if True then compute and add _gauge and _rate_per_is fields
39			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
40			fields_description = {
41			'ssid': {
42			'description': 'Wi-Fi network name.'
43			},
44			'quality_link': {
45			'description': 'Signal quality level.',
46			'unit': 'dBm',
47			},
48			'quality_level': {
49			'description': 'Signal strong level.',
50			'unit': 'dBm',
51			},
52			}
53
54
55			class PluginModel(GlancesPluginModel):
56			"""Glances Wifi plugin.
57
58			Get stats of the current Wifi hotspots.
59			"""
60
61			def __init__(self, args=None, config=None):
62			"""Init the plugin."""
63			super(PluginModel, self).__init__(args=args, config=config, stats_init_value=[])
64
65			# We want to display the stat in the curse interface
66			self.display_curse = True
67
68			# Global Thread running all the scans
69			self._thread = None
70
71			def exit(self):
72			"""Overwrite the exit method to close threads."""
73			if self._thread is not None:
74			self._thread.stop()
75			# Call the father class
76			super(PluginModel, self).exit()
77
78			def get_key(self):
79			"""Return the key of the list.
80
81			:returns: string -- SSID is the dict key
82			"""
83			return 'ssid'
84
85			@GlancesPluginModel._check_decorator
86			@GlancesPluginModel._log_result_decorator
87			def update(self):
88			"""Update Wifi stats using the input method.
89
90			Stats is a list of dict (one dict per hotspot)
91
92			:returns: list -- Stats is a list of dict (hotspot)
93			"""
94			# Init new stats
95			stats = self.get_init_value()
96
97			# Exist if we can not grab the stats
98			if not wireless_file_exists:
99			return stats
100
101			if self.input_method == 'local' and wireless_file_exists:
102			# As a backup solution, use the /proc/net/wireless file
103			with open(WIRELESS_FILE, 'r') as f:
104			# The first two lines are header
105			f.readline()
106			f.readline()
107			# Others lines are Wifi stats
108			wifi_stats = f.readline()
109			while wifi_stats != '':
110			# Extract the stats
111			wifi_stats = wifi_stats.split()
112			# Add the Wifi link to the list
113			stats.append(
114			{
115			'key': self.get_key(),
116			'ssid': wifi_stats[0][:-1],
117			'quality_link': float(wifi_stats[2]),
118			'quality_level': float(wifi_stats[3]),
119			}
120			)
121			# Next line
122			wifi_stats = f.readline()
123
124			elif self.input_method == 'snmp':
125			# Update stats using SNMP
126
127			# Not implemented yet
128			pass
129
130			# Update the stats
131			self.stats = stats
132
133			return self.stats
134
135			def get_alert(self, value):
136			"""Overwrite the default get_alert method.
137
138			Alert is on signal quality where lower is better...
139
140			:returns: string -- Signal alert
141			"""
142			ret = 'OK'
143			try:
144			if value <= self.get_limit('critical', stat_name=self.plugin_name):
145			ret = 'CRITICAL'
146			elif value <= self.get_limit('warning', stat_name=self.plugin_name):
147			ret = 'WARNING'
148			elif value <= self.get_limit('careful', stat_name=self.plugin_name):
149			ret = 'CAREFUL'
150			except (TypeError, KeyError):
151			# Catch TypeError for issue1373
152			ret = 'DEFAULT'
153
154			return ret
155
156			def update_views(self):
157			"""Update stats views."""
158			# Call the father's method
159			super(PluginModel, self).update_views()
160
161			# Add specifics information
162			# Alert on quality_level thresholds
163			for i in self.stats:
164			self.views[i[self.get_key()]]['quality_level']['decoration'] = self.get_alert(i['quality_level'])
165
166			def msg_curse(self, args=None, max_width=None):
167			"""Return the dict to display in the curse interface."""
168			# Init the return message
169			ret = []
170
171			# Only process if stats exist and display plugin enable...
172			if not self.stats or not wireless_file_exists or self.is_disabled():
173			return ret
174
175			# Max size for the interface name
176			if max_width:
177			if_name_max_width = max_width - 5
178			else:
179			# No max_width defined, return an emptu curse message
180			logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
181			return ret
182
183			# Build the string message
184			# Header
185			msg = '{:{width}}'.format('WIFI', width=if_name_max_width)
186			ret.append(self.curse_add_line(msg, "TITLE"))
187			msg = '{:>7}'.format('dBm')
188			ret.append(self.curse_add_line(msg))
189
190			# Hotspot list (sorted by name)
191			for i in sorted(self.stats, key=operator.itemgetter(self.get_key())):
192			# Do not display hotspot with no name (/ssid)...
193			# of ssid/signal None... See issue #1151 and #issue1973
194			if i['ssid'] == '' or i['ssid'] is None or i['quality_level'] is None:
195			continue
196			ret.append(self.curse_new_line())
197			# New hotspot
198			hotspot_name = i['ssid']
199			msg = '{:{width}}'.format(
200			nativestr(hotspot_name),
201			width=if_name_max_width
202			)
203			ret.append(self.curse_add_line(msg))
204			msg = '{:>7}'.format(
205			i['quality_level'],
206			)
207			ret.append(
208			self.curse_add_line(msg, self.get_views(item=i[self.get_key()],
209			key='quality_level',
210			option='decoration'))
211			)
212
213			return ret
214

nicolargo / glances

Push — develop ( 6b0de5...610ee0 )

glances.plugins.wifi.ThreadHotspot.stopped() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like