glances.standalone.GlancesStandalone.__init__() - 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.standalone.GlancesStandalone.init() F

↳ Parent: glances.standalone

Complexity

Conditions

Size

Total Lines	86
Code Lines	51

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	15
eloc	51
nop	3
dl	0
loc	86
rs	2.9998
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
#

"""Manage the Glances standalone session."""

import sys
import time

from glances.logger import logger

from glances.globals import WINDOWS
from glances.processes import glances_processes
from glances.stats import GlancesStats
from glances.outputs.glances_curses import GlancesCursesStandalone
from glances.outputs.glances_stdout import GlancesStdout
from glances.outputs.glances_stdout_json import GlancesStdoutJson
from glances.outputs.glances_stdout_csv import GlancesStdoutCsv
from glances.outputs.glances_stdout_issue import GlancesStdoutIssue
from glances.outputs.glances_stdout_apidoc import GlancesStdoutApiDoc
from glances.outdated import Outdated
from glances.timer import Counter


class GlancesStandalone(object):
    """This class creates and manages the Glances standalone session."""

    def __init__(self, config=None, args=None):
        self.config = config
        self.args = args

        # Quiet mode
        self._quiet = args.quiet
        self.refresh_time = args.time

        # Init stats
        start_duration = Counter()
        start_duration.reset()
        self.stats = GlancesStats(config=config, args=args)
        logger.debug("Plugins initialisation duration: {} seconds".format(start_duration.get()))

        # Modules (plugins and exporters) are loaded at this point
        # Glances can display the list if asked...
        if args.modules_list:
            self.display_modules_list()
            sys.exit(0)

        # The args is needed to get the selected process in the process list (Curses mode)
        glances_processes.set_args(args)

        # If process extended stats is disabled by user
        if not args.enable_process_extended:
            logger.debug("Extended stats for top process are disabled")
            glances_processes.disable_extended()
        else:
            logger.debug("Extended stats for top process are enabled")
            glances_processes.enable_extended()

        # Manage optional process filter
        if args.process_filter is not None:
            logger.info("Process filter is set to: {}".format(args.process_filter))
            glances_processes.process_filter = args.process_filter

        if (args.export or args.stdout) and args.export_process_filter is not None:
            logger.info("Export process filter is set to: {}".format(args.export_process_filter))
            glances_processes.export_process_filter = args.export_process_filter

        if (not WINDOWS) and args.no_kernel_threads:
            # Ignore kernel threads in process list
            glances_processes.disable_kernel_threads()

        # Initial system information update
        start_duration.reset()
        self.stats.update()
        logger.debug("First stats update duration: {} seconds".format(start_duration.get()))

        if self.quiet:
            logger.info("Quiet mode is ON, nothing will be displayed")
            # In quiet mode, nothing is displayed
            glances_processes.max_processes = 0
        elif args.stdout_issue:
            logger.info("Issue mode is ON")
            # Init screen
            self.screen = GlancesStdoutIssue(config=config, args=args)
        elif args.stdout_apidoc:
            logger.info("Fields descriptions mode is ON")
            # Init screen
            self.screen = GlancesStdoutApiDoc(config=config, args=args)
        elif args.stdout:
            logger.info("Stdout mode is ON, following stats will be displayed: {}".format(args.stdout))
            # Init screen
            self.screen = GlancesStdout(config=config, args=args)
        elif args.stdout_json:
            logger.info("Stdout JSON mode is ON, following stats will be displayed: {}".format(args.stdout_json))
            # Init screen
            self.screen = GlancesStdoutJson(config=config, args=args)
        elif args.stdout_csv:
            logger.info("Stdout CSV mode is ON, following stats will be displayed: {}".format(args.stdout_csv))
            # Init screen
            self.screen = GlancesStdoutCsv(config=config, args=args)
        else:
            # Default number of processes to displayed is set to 50
            glances_processes.max_processes = 50

            # Init screen
            self.screen = GlancesCursesStandalone(config=config, args=args)

            # If an error occur during the screen init, continue if export option is set
            # It is done in the screen.init function
            self._quiet = args.quiet

        # Check the latest Glances version
        self.outdated = Outdated(config=config, args=args)

    @property
    def quiet(self):
        return self._quiet

    def display_modules_list(self):
        """Display modules list"""
        print("Plugins list: {}".format(', '.join(sorted(self.stats.getPluginsList(enable=False)))))
        print("Exporters list: {}".format(', '.join(sorted(self.stats.getExportsList(enable=False)))))

    def serve_issue(self):
        """Special mode for the --issue option

        Update is done in the screen.update function
        """
        ret = not self.screen.update(self.stats)
        self.end()
        return ret

    def __serve_once(self):
        """Main loop for the CLI.

        :return: True if we should continue (no exit key has been pressed)
        """
        # Update stats
        # Start a counter used to compute the time needed
        counter = Counter()
        self.stats.update()
        logger.debug('Stats updated duration: {} seconds'.format(counter.get()))

        # Patch for issue1326 to avoid < 0 refresh
        adapted_refresh = (self.refresh_time - counter.get()) if (self.refresh_time - counter.get()) > 0 else 0

        # Display stats
        # and wait refresh_time - counter
        if not self.quiet:
            # The update function return True if an exit key 'q' or 'ESC'
            # has been pressed.
            counter_display = Counter()
            ret = not self.screen.update(self.stats, duration=adapted_refresh)
            logger.debug('Stats display duration: {} seconds'.format(counter_display.get() - adapted_refresh))
        else:
            # Nothing is displayed
            # Break should be done via a signal (CTRL-C)
            time.sleep(adapted_refresh)
            ret = True

        # Export stats
        # Start a counter used to compute the time needed
        counter_export = Counter()
        self.stats.export(self.stats)
        logger.debug('Stats exported duration: {} seconds'.format(counter_export.get()))

        return ret

    def serve_n(self, n=1):
        """Serve n time."""
        for _ in range(n):
            if not self.__serve_once():
                break
        # self.end()

    def serve_forever(self):
        """Wrapper to the serve_forever function."""
        if self.args.stop_after:
            self.serve_n(self.args.stop_after)
        else:
            while self.__serve_once():
                pass
        # self.end()

    def end(self):
        """End of the standalone CLI."""
        if not self.quiet:
            self.screen.end()

        # Exit from export modules
        self.stats.end()

        # Check Glances version versus PyPI one
        if self.outdated.is_outdated():
            print(
                "You are using Glances version {}, however version {} is available.".format(
                    self.outdated.installed_version(), self.outdated.latest_version()
                )
            )
            print("You should consider upgrading using: pip install --upgrade glances")
            print("Disable this warning temporarily using: glances --disable-check-update")
            print(
                "To disable it permanently, refer config reference at "
                "https://glances.readthedocs.io/en/latest/config.html#syntax"
            )


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			"""Manage the Glances standalone session."""
11
12			import sys
13			import time
14
15			from glances.logger import logger
16
17			from glances.globals import WINDOWS
18			from glances.processes import glances_processes
19			from glances.stats import GlancesStats
20			from glances.outputs.glances_curses import GlancesCursesStandalone
21			from glances.outputs.glances_stdout import GlancesStdout
22			from glances.outputs.glances_stdout_json import GlancesStdoutJson
23			from glances.outputs.glances_stdout_csv import GlancesStdoutCsv
24			from glances.outputs.glances_stdout_issue import GlancesStdoutIssue
25			from glances.outputs.glances_stdout_apidoc import GlancesStdoutApiDoc
26			from glances.outdated import Outdated
27			from glances.timer import Counter
28
29
30			class GlancesStandalone(object):
31			"""This class creates and manages the Glances standalone session."""
32
33			def __init__(self, config=None, args=None):
34			self.config = config
35			self.args = args
36
37			# Quiet mode
38			self._quiet = args.quiet
39			self.refresh_time = args.time
40
41			# Init stats
42			start_duration = Counter()
43			start_duration.reset()
44			self.stats = GlancesStats(config=config, args=args)
45			logger.debug("Plugins initialisation duration: {} seconds".format(start_duration.get()))
46
47			# Modules (plugins and exporters) are loaded at this point
48			# Glances can display the list if asked...
49			if args.modules_list:
50			self.display_modules_list()
51			sys.exit(0)
52
53			# The args is needed to get the selected process in the process list (Curses mode)
54			glances_processes.set_args(args)
55
56			# If process extended stats is disabled by user
57			if not args.enable_process_extended:
58			logger.debug("Extended stats for top process are disabled")
59			glances_processes.disable_extended()
60			else:
61			logger.debug("Extended stats for top process are enabled")
62			glances_processes.enable_extended()
63
64			# Manage optional process filter
65			if args.process_filter is not None:
66			logger.info("Process filter is set to: {}".format(args.process_filter))
67			glances_processes.process_filter = args.process_filter
68
69			if (args.export or args.stdout) and args.export_process_filter is not None:
70			logger.info("Export process filter is set to: {}".format(args.export_process_filter))
71			glances_processes.export_process_filter = args.export_process_filter
72
73			if (not WINDOWS) and args.no_kernel_threads:
74			# Ignore kernel threads in process list
75			glances_processes.disable_kernel_threads()
76
77			# Initial system information update
78			start_duration.reset()
79			self.stats.update()
80			logger.debug("First stats update duration: {} seconds".format(start_duration.get()))
81
82			if self.quiet:
83			logger.info("Quiet mode is ON, nothing will be displayed")
84			# In quiet mode, nothing is displayed
85			glances_processes.max_processes = 0
86			elif args.stdout_issue:
87			logger.info("Issue mode is ON")
88			# Init screen
89			self.screen = GlancesStdoutIssue(config=config, args=args)
90			elif args.stdout_apidoc:
91			logger.info("Fields descriptions mode is ON")
92			# Init screen
93			self.screen = GlancesStdoutApiDoc(config=config, args=args)
94			elif args.stdout:
95			logger.info("Stdout mode is ON, following stats will be displayed: {}".format(args.stdout))
96			# Init screen
97			self.screen = GlancesStdout(config=config, args=args)
98			elif args.stdout_json:
99			logger.info("Stdout JSON mode is ON, following stats will be displayed: {}".format(args.stdout_json))
100			# Init screen
101			self.screen = GlancesStdoutJson(config=config, args=args)
102			elif args.stdout_csv:
103			logger.info("Stdout CSV mode is ON, following stats will be displayed: {}".format(args.stdout_csv))
104			# Init screen
105			self.screen = GlancesStdoutCsv(config=config, args=args)
106			else:
107			# Default number of processes to displayed is set to 50
108			glances_processes.max_processes = 50
109
110			# Init screen
111			self.screen = GlancesCursesStandalone(config=config, args=args)
112
113			# If an error occur during the screen init, continue if export option is set
114			# It is done in the screen.init function
115			self._quiet = args.quiet
116
117			# Check the latest Glances version
118			self.outdated = Outdated(config=config, args=args)
119
120			@property
121			def quiet(self):
122			return self._quiet
123
124			def display_modules_list(self):
125			"""Display modules list"""
126			print("Plugins list: {}".format(', '.join(sorted(self.stats.getPluginsList(enable=False)))))
127			print("Exporters list: {}".format(', '.join(sorted(self.stats.getExportsList(enable=False)))))
128
129			def serve_issue(self):
130			"""Special mode for the --issue option
131
132			Update is done in the screen.update function
133			"""
134			ret = not self.screen.update(self.stats)
135			self.end()
136			return ret
137
138			def __serve_once(self):
139			"""Main loop for the CLI.
140
141			:return: True if we should continue (no exit key has been pressed)
142			"""
143			# Update stats
144			# Start a counter used to compute the time needed
145			counter = Counter()
146			self.stats.update()
147			logger.debug('Stats updated duration: {} seconds'.format(counter.get()))
148
149			# Patch for issue1326 to avoid < 0 refresh
150			adapted_refresh = (self.refresh_time - counter.get()) if (self.refresh_time - counter.get()) > 0 else 0
151
152			# Display stats
153			# and wait refresh_time - counter
154			if not self.quiet:
155			# The update function return True if an exit key 'q' or 'ESC'
156			# has been pressed.
157			counter_display = Counter()
158			ret = not self.screen.update(self.stats, duration=adapted_refresh)
159			logger.debug('Stats display duration: {} seconds'.format(counter_display.get() - adapted_refresh))
160			else:
161			# Nothing is displayed
162			# Break should be done via a signal (CTRL-C)
163			time.sleep(adapted_refresh)
164			ret = True
165
166			# Export stats
167			# Start a counter used to compute the time needed
168			counter_export = Counter()
169			self.stats.export(self.stats)
170			logger.debug('Stats exported duration: {} seconds'.format(counter_export.get()))
171
172			return ret
173
174			def serve_n(self, n=1):
175			"""Serve n time."""
176			for _ in range(n):
177			if not self.__serve_once():
178			break
179			# self.end()
180
181			def serve_forever(self):
182			"""Wrapper to the serve_forever function."""
183			if self.args.stop_after:
184			self.serve_n(self.args.stop_after)
185			else:
186			while self.__serve_once():
187			pass
188			# self.end()
189
190			def end(self):
191			"""End of the standalone CLI."""
192			if not self.quiet:
193			self.screen.end()
194
195			# Exit from export modules
196			self.stats.end()
197
198			# Check Glances version versus PyPI one
199			if self.outdated.is_outdated():
200			print(
201			"You are using Glances version {}, however version {} is available.".format(
202			self.outdated.installed_version(), self.outdated.latest_version()
203			)
204			)
205			print("You should consider upgrading using: pip install --upgrade glances")
206			print("Disable this warning temporarily using: glances --disable-check-update")
207			print(
208			"To disable it permanently, refer config reference at "
209			"https://glances.readthedocs.io/en/latest/config.html#syntax"
210			)
211

nicolargo / glances

Push — master ( ee826a...d9056e )

glances.standalone.GlancesStandalone.__init__() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

glances.standalone.GlancesStandalone.init() F