glances.plugins.system.PluginModel.update() - Code Metrics - Inspection of "Refactor plugin folder in the same way than export..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 30cc9f...48251c )

by Nicolas

created 2023-10-15 16:05 UTC

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

↳ Parent: glances.plugins.system

Complexity

Conditions

Size

Total Lines	67
Code Lines	45

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	18
eloc	45
nop	1
dl	0
loc	67
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
#

"""System plugin."""

import os
import platform
import re
from io import open

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

# 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 or Server 2016',
        '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)

        # 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)

    @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 stats['os_name'] == "Linux":
                stats['hr_name'] = stats['linux_distro']
            else:
                stats['hr_name'] = '{} {}'.format(stats['os_name'], stats['os_version'])
            stats['hr_name'] += ' {}'.format(stats['platform'])

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

nicolargo / glances

Push — develop ( 30cc9f...48251c )

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