glances.plugins.system.PluginModel.update() - 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.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') if config else None

    @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			"""Glances' host/system plugin.
114
115			stats is a dict
116			"""
117
118			def __init__(self, args=None, config=None):
119			"""Init the plugin."""
120			super(PluginModel, self).__init__(args=args, config=config, fields_description=fields_description)
121
122			# We want to display the stat in the curse interface
123			self.display_curse = True
124
125			# Set default rate to 60 seconds
126			if self.get_refresh():
127			self.set_refresh(60)
128
129			# Get the default message (if defined)
130			self.system_info_msg = config.get_value('system', 'system_info_msg') if config else None
131
132			@GlancesPluginModel._check_decorator
133			@GlancesPluginModel._log_result_decorator
134			def update(self):
135			"""Update the host/system info using the input method.
136
137			:return: the stats dict
138			"""
139			# Init new stats
140			stats = self.get_init_value()
141
142			if self.input_method == 'local':
143			# Update stats using the standard system lib
144			stats['os_name'] = platform.system()
145			stats['hostname'] = platform.node()
146			stats['platform'] = platform.architecture()[0]
147			if stats['os_name'] == "Linux":
148			try:
149			linux_distro = platform.linux_distribution()
150			except AttributeError:
151			stats['linux_distro'] = _linux_os_release()
152			else:
153			if linux_distro[0] == '':
154			stats['linux_distro'] = _linux_os_release()
155			else:
156			stats['linux_distro'] = ' '.join(linux_distro[:2])
157			stats['os_version'] = platform.release()
158			elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
159			stats['os_version'] = platform.release()
160			elif stats['os_name'] == "Darwin":
161			stats['os_version'] = platform.mac_ver()[0]
162			elif stats['os_name'] == "Windows":
163			os_version = platform.win32_ver()
164			stats['os_version'] = ' '.join(os_version[::2])
165			# if the python version is 32 bit perhaps the windows operating
166			# system is 64bit
167			if stats['platform'] == '32bit' and 'PROCESSOR_ARCHITEW6432' in os.environ:
168			stats['platform'] = '64bit'
169			else:
170			stats['os_version'] = ""
171
172			# Add human readable name
173			if self.system_info_msg:
174			try:
175			stats['hr_name'] = self.system_info_msg.format(**stats)
176			except KeyError as e:
177			logger.debug(f'Error in system_info_msg ({e})')
178			stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)
179			elif stats['os_name'] == "Linux":
180			stats['hr_name'] = '{linux_distro} {platform} / {os_name} {os_version}'.format(**stats)
181			else:
182			stats['hr_name'] = '{os_name} {os_version} {platform}'.format(**stats)
183
184			elif self.input_method == 'snmp':
185			# Update stats using SNMP
186			try:
187			stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
188			except KeyError:
189			stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
190			# Default behavior: display all the information
191			stats['os_name'] = stats['system_name']
192			# Windows OS tips
193			if self.short_system_name == 'windows':
194			for r, v in iteritems(snmp_to_human['windows']):
195			if re.search(r, stats['system_name']):
196			stats['os_name'] = v
197			break
198			# Add human readable name
199			stats['hr_name'] = stats['os_name']
200
201			# Update the stats
202			self.stats = stats
203
204			return self.stats
205
206			def msg_curse(self, args=None, max_width=None):
207			"""Return the string to display in the curse interface."""
208			# Init the return message
209			ret = []
210
211			# Only process if stats exist and plugin not disabled
212			if not self.stats or self.is_disabled():
213			return ret
214
215			# Build the string message
216			if args.client:
217			# Client mode
218			if args.cs_status.lower() == "connected":
219			msg = 'Connected to '
220			ret.append(self.curse_add_line(msg, 'OK'))
221			elif args.cs_status.lower() == "snmp":
222			msg = 'SNMP from '
223			ret.append(self.curse_add_line(msg, 'OK'))
224			elif args.cs_status.lower() == "disconnected":
225			msg = 'Disconnected from '
226			ret.append(self.curse_add_line(msg, 'CRITICAL'))
227
228			# Hostname is mandatory
229			msg = self.stats['hostname']
230			ret.append(self.curse_add_line(msg, "TITLE"))
231
232			# System info
233			msg = ' ' + self.stats['hr_name']
234			ret.append(self.curse_add_line(msg, optional=True))
235
236			# Return the message with decoration
237			return ret
238

nicolargo / glances

Push — master ( ee826a...d9056e )

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