glances.standalone.GlancesStandalone.serve_n() - Code Metrics - Inspection of "Update release note for Glances 3.2.5" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 7e7379...128504 )

by Nicolas

created 2022-04-10 14:14 UTC

glances.standalone.GlancesStandalone.serve_n() A

↳ Parent: glances.standalone

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	4
nop	2
dl	0
loc	5
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

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

import sys
import time

from glances.globals import WINDOWS
from glances.logger import logger
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_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)

        # 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:
            glances_processes.process_filter = args.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_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)

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

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

        # Patch for issue1326 to avoid < 0 refresh
        adapted_refresh = self.refresh_time - counter.get()
        adapted_refresh = adapted_refresh if adapted_refresh > 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.
            ret = not self.screen.update(self.stats, duration=adapted_refresh)
        else:
            # Nothing is displayed
            # Break should be done via a signal (CTRL-C)
            time.sleep(adapted_refresh)
            ret = True

        return ret

    def serve_forever(self):
        """Wrapper to the serve_forever function."""
        if self.args.stop_after:
            for _ in range(self.args.stop_after):
                if not self.__serve_once():
                    break
        else:
            while self.__serve_once():
                pass
        # self.end()

    def serve_n(self, n=1):
        """Serve n time."""
        for _ in range(n):
            if not self.__serve_once():
                break
        # 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")


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2019 Nicolargo <[email protected]>
6			#
7			# Glances is free software; you can redistribute it and/or modify
8			# it under the terms of the GNU Lesser General Public License as published by
9			# the Free Software Foundation, either version 3 of the License, or
10			# (at your option) any later version.
11			#
12			# Glances is distributed in the hope that it will be useful,
13			# but WITHOUT ANY WARRANTY; without even the implied warranty of
14			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15			# GNU Lesser General Public License for more details.
16			#
17			# You should have received a copy of the GNU Lesser General Public License
18			# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20			"""Manage the Glances standalone session."""
21
22			import sys
23			import time
24
25			from glances.globals import WINDOWS
26			from glances.logger import logger
27			from glances.processes import glances_processes
28			from glances.stats import GlancesStats
29			from glances.outputs.glances_curses import GlancesCursesStandalone
30			from glances.outputs.glances_stdout import GlancesStdout
31			from glances.outputs.glances_stdout_csv import GlancesStdoutCsv
32			from glances.outputs.glances_stdout_issue import GlancesStdoutIssue
33			from glances.outputs.glances_stdout_apidoc import GlancesStdoutApiDoc
34			from glances.outdated import Outdated
35			from glances.timer import Counter
36
37
38			class GlancesStandalone(object):
39
40			"""This class creates and manages the Glances standalone session."""
41
42			def __init__(self, config=None, args=None):
43			self.config = config
44			self.args = args
45
46			# Quiet mode
47			self._quiet = args.quiet
48			self.refresh_time = args.time
49
50			# Init stats
51			start_duration = Counter()
52			start_duration.reset()
53			self.stats = GlancesStats(config=config, args=args)
54			logger.debug("Plugins initialisation duration: {} seconds".format(start_duration.get()))
55
56			# Modules (plugins and exporters) are loaded at this point
57			# Glances can display the list if asked...
58			if args.modules_list:
59			self.display_modules_list()
60			sys.exit(0)
61
62			# If process extended stats is disabled by user
63			if not args.enable_process_extended:
64			logger.debug("Extended stats for top process are disabled")
65			glances_processes.disable_extended()
66			else:
67			logger.debug("Extended stats for top process are enabled")
68			glances_processes.enable_extended()
69
70			# Manage optional process filter
71			if args.process_filter is not None:
72			glances_processes.process_filter = args.process_filter
73
74			if (not WINDOWS) and args.no_kernel_threads:
75			# Ignore kernel threads in process list
76			glances_processes.disable_kernel_threads()
77
78			# Initial system information update
79			start_duration.reset()
80			self.stats.update()
81			logger.debug("First stats update duration: {} seconds".format(start_duration.get()))
82
83			if self.quiet:
84			logger.info("Quiet mode is ON, nothing will be displayed")
85			# In quiet mode, nothing is displayed
86			glances_processes.max_processes = 0
87			elif args.stdout_issue:
88			logger.info("Issue mode is ON")
89			# Init screen
90			self.screen = GlancesStdoutIssue(config=config, args=args)
91			elif args.stdout_apidoc:
92			logger.info("Fields descriptions mode is ON")
93			# Init screen
94			self.screen = GlancesStdoutApiDoc(config=config, args=args)
95			elif args.stdout:
96			logger.info("Stdout mode is ON, following stats will be displayed: {}".format(args.stdout))
97			# Init screen
98			self.screen = GlancesStdout(config=config, args=args)
99			elif args.stdout_csv:
100			logger.info("Stdout CSV mode is ON, following stats will be displayed: {}".format(args.stdout_csv))
101			# Init screen
102			self.screen = GlancesStdoutCsv(config=config, args=args)
103			else:
104			# Default number of processes to displayed is set to 50
105			glances_processes.max_processes = 50
106
107			# Init screen
108			self.screen = GlancesCursesStandalone(config=config, args=args)
109
110			# Check the latest Glances version
111			self.outdated = Outdated(config=config, args=args)
112
113			@property
114			def quiet(self):
115			return self._quiet
116
117			def display_modules_list(self):
118			"""Display modules list"""
119			print("Plugins list: {}".format(', '.join(sorted(self.stats.getPluginsList(enable=False)))))
120			print("Exporters list: {}".format(', '.join(sorted(self.stats.getExportsList(enable=False)))))
121
122			def serve_issue(self):
123			"""Special mode for the --issue option
124
125			Update is done in the screen.update function
126			"""
127			ret = not self.screen.update(self.stats)
128			self.end()
129			return ret
130
131			def __serve_once(self):
132			"""Main loop for the CLI.
133
134			:return: True if we should continue (no exit key has been pressed)
135			"""
136			# Update stats
137			# Start a counter used to compute the time needed
138			counter = Counter()
139			self.stats.update()
140			logger.debug('Stats updated duration: {} seconds'.format(counter.get()))
141
142			# Export stats
143			# Start a counter used to compute the time needed
144			counter_export = Counter()
145			self.stats.export(self.stats)
146			logger.debug('Stats exported duration: {} seconds'.format(counter_export.get()))
147
148			# Patch for issue1326 to avoid < 0 refresh
149			adapted_refresh = self.refresh_time - counter.get()
150			adapted_refresh = adapted_refresh if adapted_refresh > 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			ret = not self.screen.update(self.stats, duration=adapted_refresh)
158			else:
159			# Nothing is displayed
160			# Break should be done via a signal (CTRL-C)
161			time.sleep(adapted_refresh)
162			ret = True
163
164			return ret
165
166			def serve_forever(self):
167			"""Wrapper to the serve_forever function."""
168			if self.args.stop_after:
169			for _ in range(self.args.stop_after):
170			if not self.__serve_once():
171			break
172			else:
173			while self.__serve_once():
174			pass
175			# self.end()
176
177			def serve_n(self, n=1):
178			"""Serve n time."""
179			for _ in range(n):
180			if not self.__serve_once():
181			break
182			# self.end()
183
184			def end(self):
185			"""End of the standalone CLI."""
186			if not self.quiet:
187			self.screen.end()
188
189			# Exit from export modules
190			self.stats.end()
191
192			# Check Glances version versus PyPI one
193			if self.outdated.is_outdated():
194			print(
195			"You are using Glances version {}, however version {} is available.".format(
196			self.outdated.installed_version(), self.outdated.latest_version()
197			)
198			)
199			print("You should consider upgrading using: pip install --upgrade glances")
200

nicolargo / glances

Push — master ( 7e7379...128504 )

glances.standalone.GlancesStandalone.serve_n() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like