glances.plugins.system.PluginModel.update() - Code Metrics - Inspection of "[WEBUI] Custom system header information #2695" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 43b355...49977d )

by Nicolas

created 2024-03-10 10:09 UTC

glances.plugins.system.PluginModel.update() F

↳ Parent: glances.plugins.system

Complexity

Conditions

Size

Total Lines	73
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	20
eloc	50
nop	1
dl	0
loc	73
rs	0
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
#

"""System plugin."""

import os
import platform
import re
from io import open

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

# {
#   "os_name": "Linux",
#   "hostname": "XPS13-9333",
#   "platform": "64bit",
#   "linux_distro": "Ubuntu 22.04",
#   "os_version": "5.15.0-88-generic",
#   "hr_name": "Ubuntu 22.04 64bit"
# }
# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'os_name': {
        'description': 'Operating system name',
    },
    'hostname': {
        'description': 'Hostname',
    },
    'platform': {
        'description': 'Platform (32 or 64 bits)',
    },
    'linux_distro': {
        'description': 'Linux distribution',
    },
    'os_version': {
        'description': 'Operating system version',
    },
    'hr_name': {
        'description': 'Human readable operating sytem name',
    }
}

# SNMP OID
snmp_oid = {
    'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
    'netapp': {
        'hostname': '1.3.6.1.2.1.1.5.0',
        'system_name': '1.3.6.1.2.1.1.1.0',
        'platform': '1.3.6.1.4.1.789.1.1.5.0',
    },
}

# SNMP to human read
# Dict (key: OS short name) of dict (reg exp OID to human)
# Windows:
# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
snmp_to_human = {
    'windows': {
        'Windows Version 10.0': 'Windows 10|11 or Server 2016|2019|2022',
        'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
        'Windows Version 6.2': 'Windows 8 or Server 2012',
        'Windows Version 6.1': 'Windows 7 or Server 2008R2',
        'Windows Version 6.0': 'Windows Vista or Server 2008',
        'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
        'Windows Version 5.1': 'Windows XP',
        'Windows Version 5.0': 'Windows 2000',
    }
}


def _linux_os_release():
    """Try to determine the name of a Linux distribution.

    This function checks for the /etc/os-release file.
    It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
    An empty string is returned if the above values cannot be determined.
    """
    pretty_name = ''
    ashtray = {}
    keys = ['NAME', 'VERSION_ID']
    try:
        with open(os.path.join('/etc', 'os-release')) as f:
            for line in f:
                for key in keys:
                    if line.startswith(key):
                        ashtray[key] = re.sub(r'^"|"$', '', line.strip().split('=')[1])
    except (OSError, IOError):
        return pretty_name

    if ashtray:
        if 'NAME' in ashtray:
            pretty_name = ashtray['NAME']
        if 'VERSION_ID' in ashtray:
            pretty_name += ' {}'.format(ashtray['VERSION_ID'])

    return pretty_name


class PluginModel(GlancesPluginModel):

    """Glances' host/system plugin.

    stats is a dict
    """

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

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

        # Set default rate to 60 seconds
        if self.get_refresh():
            self.set_refresh(60)

        # Get the default message (if defined)
        self.system_info_msg = config.get_value('system', 'system_info_msg')

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the host/system info using the input method.

        :return: the stats dict
        """
        # Init new stats
        stats = self.get_init_value()

        if self.input_method == 'local':
            # Update stats using the standard system lib
            stats['os_name'] = platform.system()
            stats['hostname'] = platform.node()
            stats['platform'] = platform.architecture()[0]
            if stats['os_name'] == "Linux":
                try:
                    linux_distro = platform.linux_distribution()
                except AttributeError:
                    stats['linux_distro'] = _linux_os_release()
                else:
                    if linux_distro[0] == '':
                        stats['linux_distro'] = _linux_os_release()
                    else:
                        stats['linux_distro'] = ' '.join(linux_distro[:2])
                stats['os_version'] = platform.release()
            elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
                stats['os_version'] = platform.release()
            elif stats['os_name'] == "Darwin":
                stats['os_version'] = platform.mac_ver()[0]
            elif stats['os_name'] == "Windows":
                os_version = platform.win32_ver()
                stats['os_version'] = ' '.join(os_version[::2])
                # if the python version is 32 bit perhaps the windows operating
                # system is 64bit
                if stats['platform'] == '32bit' and 'PROCESSOR_ARCHITEW6432' in os.environ:
                    stats['platform'] = '64bit'
            else:
                stats['os_version'] = ""

            # Add human readable name
            if self.system_info_msg:
                try:
                    stats['hr_name'] = self.system_info_msg.format(**stats)
                except KeyError as e:
                    logger.debug(f'Error in system_info_msg ({e})')
                    stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)
            elif stats['os_name'] == "Linux":
                stats['hr_name'] = '{linux_distro} {platform} / {os_name} {os_version}'.format(**stats)
            else:
                stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)

        elif self.input_method == 'snmp':
            # Update stats using SNMP
            try:
                stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
            except KeyError:
                stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
            # Default behavior: display all the information
            stats['os_name'] = stats['system_name']
            # Windows OS tips
            if self.short_system_name == 'windows':
                for r, v in iteritems(snmp_to_human['windows']):
                    if re.search(r, stats['system_name']):
                        stats['os_name'] = v
                        break
            # Add human readable name
            stats['hr_name'] = stats['os_name']

        # Update the stats
        self.stats = stats

        return self.stats

    def msg_curse(self, args=None, max_width=None):
        """Return the string to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist and plugin not disabled
        if not self.stats or self.is_disabled():
            return ret

        # Build the string message
        if args.client:
            # Client mode
            if args.cs_status.lower() == "connected":
                msg = 'Connected to '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "snmp":
                msg = 'SNMP from '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "disconnected":
                msg = 'Disconnected from '
                ret.append(self.curse_add_line(msg, 'CRITICAL'))

        # Hostname is mandatory
        msg = self.stats['hostname']
        ret.append(self.curse_add_line(msg, "TITLE"))

        # System info
        msg = ' ' + self.stats['hr_name']
        ret.append(self.curse_add_line(msg, optional=True))

        # Return the message with 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			"""System plugin."""
11
12			import os
13			import platform
14			import re
15			from io import open
16
17			from glances.logger import logger
18			from glances.globals import iteritems
19			from glances.plugins.plugin.model import GlancesPluginModel
20
21			# {
22			# "os_name": "Linux",
23			# "hostname": "XPS13-9333",
24			# "platform": "64bit",
25			# "linux_distro": "Ubuntu 22.04",
26			# "os_version": "5.15.0-88-generic",
27			# "hr_name": "Ubuntu 22.04 64bit"
28			# }
29			# Fields description
30			# description: human readable description
31			# short_name: shortname to use un UI
32			# unit: unit type
33			# rate: is it a rate ? If yes, // by time_since_update when displayed,
34			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
35			fields_description = {
36			'os_name': {
37			'description': 'Operating system name',
38			},
39			'hostname': {
40			'description': 'Hostname',
41			},
42			'platform': {
43			'description': 'Platform (32 or 64 bits)',
44			},
45			'linux_distro': {
46			'description': 'Linux distribution',
47			},
48			'os_version': {
49			'description': 'Operating system version',
50			},
51			'hr_name': {
52			'description': 'Human readable operating sytem name',
53			}
54			}
55
56			# SNMP OID
57			snmp_oid = {
58			'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
59			'netapp': {
60			'hostname': '1.3.6.1.2.1.1.5.0',
61			'system_name': '1.3.6.1.2.1.1.1.0',
62			'platform': '1.3.6.1.4.1.789.1.1.5.0',
63			},
64			}
65
66			# SNMP to human read
67			# Dict (key: OS short name) of dict (reg exp OID to human)
68			# Windows:
69			# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
70			snmp_to_human = {
71			'windows': {
72			'Windows Version 10.0': 'Windows 10\|11 or Server 2016\|2019\|2022',
73			'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
74			'Windows Version 6.2': 'Windows 8 or Server 2012',
75			'Windows Version 6.1': 'Windows 7 or Server 2008R2',
76			'Windows Version 6.0': 'Windows Vista or Server 2008',
77			'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
78			'Windows Version 5.1': 'Windows XP',
79			'Windows Version 5.0': 'Windows 2000',
80			}
81			}
82
83
84			def _linux_os_release():
85			"""Try to determine the name of a Linux distribution.
86
87			This function checks for the /etc/os-release file.
88			It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
89			An empty string is returned if the above values cannot be determined.
90			"""
91			pretty_name = ''
92			ashtray = {}
93			keys = ['NAME', 'VERSION_ID']
94			try:
95			with open(os.path.join('/etc', 'os-release')) as f:
96			for line in f:
97			for key in keys:
98			if line.startswith(key):
99			ashtray[key] = re.sub(r'^"\|"$', '', line.strip().split('=')[1])
100			except (OSError, IOError):
101			return pretty_name
102
103			if ashtray:
104			if 'NAME' in ashtray:
105			pretty_name = ashtray['NAME']
106			if 'VERSION_ID' in ashtray:
107			pretty_name += ' {}'.format(ashtray['VERSION_ID'])
108
109			return pretty_name
110
111
112			class PluginModel(GlancesPluginModel):
113
114			"""Glances' host/system plugin.
115
116			stats is a dict
117			"""
118
119			def __init__(self, args=None, config=None):
120			"""Init the plugin."""
121			super(PluginModel, self).__init__(
122			args=args,
123			config=config,
124			fields_description=fields_description
125			)
126
127			# We want to display the stat in the curse interface
128			self.display_curse = True
129
130			# Set default rate to 60 seconds
131			if self.get_refresh():
132			self.set_refresh(60)
133
134			# Get the default message (if defined)
135			self.system_info_msg = config.get_value('system', 'system_info_msg')
136
137			@GlancesPluginModel._check_decorator
138			@GlancesPluginModel._log_result_decorator
139			def update(self):
140			"""Update the host/system info using the input method.
141
142			:return: the stats dict
143			"""
144			# Init new stats
145			stats = self.get_init_value()
146
147			if self.input_method == 'local':
148			# Update stats using the standard system lib
149			stats['os_name'] = platform.system()
150			stats['hostname'] = platform.node()
151			stats['platform'] = platform.architecture()[0]
152			if stats['os_name'] == "Linux":
153			try:
154			linux_distro = platform.linux_distribution()
155			except AttributeError:
156			stats['linux_distro'] = _linux_os_release()
157			else:
158			if linux_distro[0] == '':
159			stats['linux_distro'] = _linux_os_release()
160			else:
161			stats['linux_distro'] = ' '.join(linux_distro[:2])
162			stats['os_version'] = platform.release()
163			elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
164			stats['os_version'] = platform.release()
165			elif stats['os_name'] == "Darwin":
166			stats['os_version'] = platform.mac_ver()[0]
167			elif stats['os_name'] == "Windows":
168			os_version = platform.win32_ver()
169			stats['os_version'] = ' '.join(os_version[::2])
170			# if the python version is 32 bit perhaps the windows operating
171			# system is 64bit
172			if stats['platform'] == '32bit' and 'PROCESSOR_ARCHITEW6432' in os.environ:
173			stats['platform'] = '64bit'
174			else:
175			stats['os_version'] = ""
176
177			# Add human readable name
178			if self.system_info_msg:
179			try:
180			stats['hr_name'] = self.system_info_msg.format(**stats)
181			except KeyError as e:
182			logger.debug(f'Error in system_info_msg ({e})')
183			stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)
184			elif stats['os_name'] == "Linux":
185			stats['hr_name'] = '{linux_distro} {platform} / {os_name} {os_version}'.format(**stats)
186			else:
187			stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)
188
189			elif self.input_method == 'snmp':
190			# Update stats using SNMP
191			try:
192			stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
193			except KeyError:
194			stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
195			# Default behavior: display all the information
196			stats['os_name'] = stats['system_name']
197			# Windows OS tips
198			if self.short_system_name == 'windows':
199			for r, v in iteritems(snmp_to_human['windows']):
200			if re.search(r, stats['system_name']):
201			stats['os_name'] = v
202			break
203			# Add human readable name
204			stats['hr_name'] = stats['os_name']
205
206			# Update the stats
207			self.stats = stats
208
209			return self.stats
210
211			def msg_curse(self, args=None, max_width=None):
212			"""Return the string to display in the curse interface."""
213			# Init the return message
214			ret = []
215
216			# Only process if stats exist and plugin not disabled
217			if not self.stats or self.is_disabled():
218			return ret
219
220			# Build the string message
221			if args.client:
222			# Client mode
223			if args.cs_status.lower() == "connected":
224			msg = 'Connected to '
225			ret.append(self.curse_add_line(msg, 'OK'))
226			elif args.cs_status.lower() == "snmp":
227			msg = 'SNMP from '
228			ret.append(self.curse_add_line(msg, 'OK'))
229			elif args.cs_status.lower() == "disconnected":
230			msg = 'Disconnected from '
231			ret.append(self.curse_add_line(msg, 'CRITICAL'))
232
233			# Hostname is mandatory
234			msg = self.stats['hostname']
235			ret.append(self.curse_add_line(msg, "TITLE"))
236
237			# System info
238			msg = ' ' + self.stats['hr_name']
239			ret.append(self.curse_add_line(msg, optional=True))
240
241			# Return the message with decoration
242			return ret
243

nicolargo / glances

Push — develop ( 43b355...49977d )

glances.plugins.system.PluginModel.update() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like