Test Failed
Push — master ( 183265...afa1da )
by Nicolas
03:15 queued 16s
created

PluginModel._get_wireless_stats()   A

Complexity

Conditions 3

Size

Total Lines 24
Code Lines 16

Duplication

Lines 0
Ratio 0 %

Importance

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