GlancesStats.getAllViews() - Code Metrics - Inspection of "version 2.8.1" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e61e99...a0b053 )

by Nicolas

created 2017-02-06 17:13 UTC

GlancesStats.getAllViews() A

↳ Parent: GlancesStats

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	3
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2017 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/>.

"""The stats manager."""

import collections
import os
import sys
import threading

from glances.globals import exports_path, plugins_path, sys_path
from glances.logger import logger


class GlancesStats(object):

    """This class stores, updates and gives stats."""

    def __init__(self, config=None, args=None):
        # Set the config instance
        self.config = config

        # Set the argument instance
        self.args = args

        # Load plugins and exports modules
        self.load_modules(self.args)

        # Load the limits (for plugins)
        self.load_limits(self.config)

    def __getattr__(self, item):
        """Overwrite the getattr method in case of attribute is not found.

        The goal is to dynamically generate the following methods:
        - getPlugname(): return Plugname stat in JSON format
        """
        # Check if the attribute starts with 'get'
        if item.startswith('get'):
            # Get the plugin name
            plugname = item[len('get'):].lower()
            # Get the plugin instance
            plugin = self._plugins[plugname]
            if hasattr(plugin, 'get_stats'):
                # The method get_stats exist, return it
                return getattr(plugin, 'get_stats')
            else:
                # The method get_stats is not found for the plugin
                raise AttributeError(item)
        else:
            # Default behavior
            raise AttributeError(item)

    def load_modules(self, args):
        """Wrapper to load: plugins and export modules."""

        # Init the plugins dict
        self._plugins = collections.defaultdict(dict)

        # Load the plugins
        self.load_plugins(args=args)

        # Init the export modules dict
        self._exports = collections.defaultdict(dict)

        # Load the export modules
        self.load_exports(args=args)

        # Restoring system path
        sys.path = sys_path

    def load_plugins(self, args=None):
        """Load all plugins in the 'plugins' folder."""
        header = "glances_"
        for item in os.listdir(plugins_path):
            if (item.startswith(header) and
                    item.endswith(".py") and
                    item != (header + "plugin.py")):
                # Import the plugin
                plugin = __import__(os.path.basename(item)[:-3])
                # Add the plugin to the dictionary
                # The key is the plugin name
                # for example, the file glances_xxx.py
                # generate self._plugins_list["xxx"] = ...
                plugin_name = os.path.basename(item)[len(header):-3].lower()
                if plugin_name in ('help', 'amps', 'ports'):
                    self._plugins[plugin_name] = plugin.Plugin(args=args, config=self.config)
                else:
                    self._plugins[plugin_name] = plugin.Plugin(args=args)
        # Log plugins list
        logger.debug("Available plugins list: {}".format(self.getAllPlugins()))

    def load_exports(self, args=None):
        """Load all export modules in the 'exports' folder."""
        if args is None:
            return False
        header = "glances_"
        # Transform the arguments list into a dict
        # The aim is to chec if the export module should be loaded
        args_var = vars(locals()['args'])
        for item in os.listdir(exports_path):
            export_name = os.path.basename(item)[len(header):-3].lower()
            if (item.startswith(header) and
                    item.endswith(".py") and
                    item != (header + "export.py") and
                    item != (header + "history.py") and
                    args_var['export_' + export_name] is not None and
                    args_var['export_' + export_name] is not False):
                # Import the export module
                export_module = __import__(os.path.basename(item)[:-3])
                # Add the export to the dictionary
                # The key is the module name
                # for example, the file glances_xxx.py
                # generate self._exports_list["xxx"] = ...
                self._exports[export_name] = export_module.Export(args=args, config=self.config)
        # Log plugins list
        logger.debug("Available exports modules list: {}".format(self.getExportList()))
        return True

    def getAllPlugins(self, enable=True):
        """Return the enable plugins list.
        if enable is False, return the list of all the plugins"""
        if enable:
            return [p for p in self._plugins if self._plugins[p].is_enable()]
        else:
            return [p for p in self._plugins]

    def getExportList(self):
        """Return the exports modules list."""
        return [e for e in self._exports]

    def load_limits(self, config=None):
        """Load the stats limits (except the one in the exclude list)."""
        # For each plugins, call the load_limits method
        for p in self._plugins:
            self._plugins[p].load_limits(config)

    def update(self):
        """Wrapper method to update the stats."""
        # For standalone and server modes
        # For each plugins, call the update method
        for p in self._plugins:
            if self._plugins[p].is_disable():
                # If current plugin is disable
                # then continue to next plugin
                continue
            # Update the stats...
            self._plugins[p].update()
            # ... the history
            self._plugins[p].update_stats_history()
            # ... and the views
            self._plugins[p].update_views()

    def export(self, input_stats=None):
        """Export all the stats.

        Each export module is ran in a dedicated thread.
        """
        # threads = []
        input_stats = input_stats or {}

        for e in self._exports:
            logger.debug("Export stats using the %s module" % e)
            thread = threading.Thread(target=self._exports[e].update,
                                      args=(input_stats,))
            # threads.append(thread)
            thread.start()

    def getAll(self):
        """Return all the stats (list)."""
        return [self._plugins[p].get_raw() for p in self._plugins]

    def getAllAsDict(self):
        """Return all the stats (dict)."""
        return {p: self._plugins[p].get_raw() for p in self._plugins}

    def getAllExports(self):
        """
        Return all the stats to be exported (list).
        Default behavor is to export all the stat
        """
        return [self._plugins[p].get_export() for p in self._plugins]

    def getAllExportsAsDict(self, plugin_list=None):
        """
        Return all the stats to be exported (list).
        Default behavor is to export all the stat
        if plugin_list is provided, only export stats of given plugin (list)
        """
        if plugin_list is None:
            # All plugins should be exported
            plugin_list = self._plugins
        return {p: self._plugins[p].get_export() for p in plugin_list}

    def getAllLimits(self):
        """Return the plugins limits list."""
        return [self._plugins[p].limits for p in self._plugins]

    def getAllLimitsAsDict(self, plugin_list=None):
        """
        Return all the stats limits (dict).
        Default behavor is to export all the limits
        if plugin_list is provided, only export limits of given plugin (list)
        """
        if plugin_list is None:
            # All plugins should be exported
            plugin_list = self._plugins
        return {p: self._plugins[p].limits for p in plugin_list}

    def getAllViews(self):
        """Return the plugins views."""
        return [self._plugins[p].get_views() for p in self._plugins]

    def getAllViewsAsDict(self):
        """Return all the stats views (dict)."""
        return {p: self._plugins[p].get_views() for p in self._plugins}

    def get_plugin_list(self):
        """Return the plugin list."""
        return self._plugins

    def get_plugin(self, plugin_name):
        """Return the plugin name."""
        if plugin_name in self._plugins:
            return self._plugins[plugin_name]
        else:
            return None

    def end(self):
        """End of the Glances stats."""
        # Close export modules
        for e in self._exports:
            self._exports[e].exit()
        # Close plugins
        for p in self._plugins:
            self._plugins[p].exit()


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2017 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			"""The stats manager."""
21
22			import collections
23			import os
24			import sys
25			import threading
26
27			from glances.globals import exports_path, plugins_path, sys_path
28			from glances.logger import logger
29
30
31			class GlancesStats(object):
32
33			"""This class stores, updates and gives stats."""
34
35			def __init__(self, config=None, args=None):
36			# Set the config instance
37			self.config = config
38
39			# Set the argument instance
40			self.args = args
41
42			# Load plugins and exports modules
43			self.load_modules(self.args)
44
45			# Load the limits (for plugins)
46			self.load_limits(self.config)
47
48			def __getattr__(self, item):
49			"""Overwrite the getattr method in case of attribute is not found.
50
51			The goal is to dynamically generate the following methods:
52			- getPlugname(): return Plugname stat in JSON format
53			"""
54			# Check if the attribute starts with 'get'
55			if item.startswith('get'):
56			# Get the plugin name
57			plugname = item[len('get'):].lower()
58			# Get the plugin instance
59			plugin = self._plugins[plugname]
60			if hasattr(plugin, 'get_stats'):
61			# The method get_stats exist, return it
62			return getattr(plugin, 'get_stats')
63			else:
64			# The method get_stats is not found for the plugin
65			raise AttributeError(item)
66			else:
67			# Default behavior
68			raise AttributeError(item)
69
70			def load_modules(self, args):
71			"""Wrapper to load: plugins and export modules."""
72
73			# Init the plugins dict
74			self._plugins = collections.defaultdict(dict)
75
76			# Load the plugins
77			self.load_plugins(args=args)
78
79			# Init the export modules dict
80			self._exports = collections.defaultdict(dict)
81
82			# Load the export modules
83			self.load_exports(args=args)
84
85			# Restoring system path
86			sys.path = sys_path
87
88			def load_plugins(self, args=None):
89			"""Load all plugins in the 'plugins' folder."""
90			header = "glances_"
91			for item in os.listdir(plugins_path):
92			if (item.startswith(header) and
93			item.endswith(".py") and
94			item != (header + "plugin.py")):
95			# Import the plugin
96			plugin = __import__(os.path.basename(item)[:-3])
97			# Add the plugin to the dictionary
98			# The key is the plugin name
99			# for example, the file glances_xxx.py
100			# generate self._plugins_list["xxx"] = ...
101			plugin_name = os.path.basename(item)[len(header):-3].lower()
102			if plugin_name in ('help', 'amps', 'ports'):
103			self._plugins[plugin_name] = plugin.Plugin(args=args, config=self.config)
104			else:
105			self._plugins[plugin_name] = plugin.Plugin(args=args)
106			# Log plugins list
107			logger.debug("Available plugins list: {}".format(self.getAllPlugins()))
108
109			def load_exports(self, args=None):
110			"""Load all export modules in the 'exports' folder."""
111			if args is None:
112			return False
113			header = "glances_"
114			# Transform the arguments list into a dict
115			# The aim is to chec if the export module should be loaded
116			args_var = vars(locals()['args'])
117			for item in os.listdir(exports_path):
118			export_name = os.path.basename(item)[len(header):-3].lower()
119			if (item.startswith(header) and
120			item.endswith(".py") and
121			item != (header + "export.py") and
122			item != (header + "history.py") and
123			args_var['export_' + export_name] is not None and
124			args_var['export_' + export_name] is not False):
125			# Import the export module
126			export_module = __import__(os.path.basename(item)[:-3])
127			# Add the export to the dictionary
128			# The key is the module name
129			# for example, the file glances_xxx.py
130			# generate self._exports_list["xxx"] = ...
131			self._exports[export_name] = export_module.Export(args=args, config=self.config)
132			# Log plugins list
133			logger.debug("Available exports modules list: {}".format(self.getExportList()))
134			return True
135
136			def getAllPlugins(self, enable=True):
137			"""Return the enable plugins list.
138			if enable is False, return the list of all the plugins"""
139			if enable:
140			return [p for p in self._plugins if self._plugins[p].is_enable()]
141			else:
142			return [p for p in self._plugins]
143
144			def getExportList(self):
145			"""Return the exports modules list."""
146			return [e for e in self._exports]
147
148			def load_limits(self, config=None):
149			"""Load the stats limits (except the one in the exclude list)."""
150			# For each plugins, call the load_limits method
151			for p in self._plugins:
152			self._plugins[p].load_limits(config)
153
154			def update(self):
155			"""Wrapper method to update the stats."""
156			# For standalone and server modes
157			# For each plugins, call the update method
158			for p in self._plugins:
159			if self._plugins[p].is_disable():
160			# If current plugin is disable
161			# then continue to next plugin
162			continue
163			# Update the stats...
164			self._plugins[p].update()
165			# ... the history
166			self._plugins[p].update_stats_history()
167			# ... and the views
168			self._plugins[p].update_views()
169
170			def export(self, input_stats=None):
171			"""Export all the stats.
172
173			Each export module is ran in a dedicated thread.
174			"""
175			# threads = []
176			input_stats = input_stats or {}
177
178			for e in self._exports:
179			logger.debug("Export stats using the %s module" % e)
180			thread = threading.Thread(target=self._exports[e].update,
181			args=(input_stats,))
182			# threads.append(thread)
183			thread.start()
184
185			def getAll(self):
186			"""Return all the stats (list)."""
187			return [self._plugins[p].get_raw() for p in self._plugins]
188
189			def getAllAsDict(self):
190			"""Return all the stats (dict)."""
191			return {p: self._plugins[p].get_raw() for p in self._plugins}
192
193			def getAllExports(self):
194			"""
195			Return all the stats to be exported (list).
196			Default behavor is to export all the stat
197			"""
198			return [self._plugins[p].get_export() for p in self._plugins]
199
200			def getAllExportsAsDict(self, plugin_list=None):
201			"""
202			Return all the stats to be exported (list).
203			Default behavor is to export all the stat
204			if plugin_list is provided, only export stats of given plugin (list)
205			"""
206			if plugin_list is None:
207			# All plugins should be exported
208			plugin_list = self._plugins
209			return {p: self._plugins[p].get_export() for p in plugin_list}
210
211			def getAllLimits(self):
212			"""Return the plugins limits list."""
213			return [self._plugins[p].limits for p in self._plugins]
214
215			def getAllLimitsAsDict(self, plugin_list=None):
216			"""
217			Return all the stats limits (dict).
218			Default behavor is to export all the limits
219			if plugin_list is provided, only export limits of given plugin (list)
220			"""
221			if plugin_list is None:
222			# All plugins should be exported
223			plugin_list = self._plugins
224			return {p: self._plugins[p].limits for p in plugin_list}
225
226			def getAllViews(self):
227			"""Return the plugins views."""
228			return [self._plugins[p].get_views() for p in self._plugins]
229
230			def getAllViewsAsDict(self):
231			"""Return all the stats views (dict)."""
232			return {p: self._plugins[p].get_views() for p in self._plugins}
233
234			def get_plugin_list(self):
235			"""Return the plugin list."""
236			return self._plugins
237
238			def get_plugin(self, plugin_name):
239			"""Return the plugin name."""
240			if plugin_name in self._plugins:
241			return self._plugins[plugin_name]
242			else:
243			return None
244
245			def end(self):
246			"""End of the Glances stats."""
247			# Close export modules
248			for e in self._exports:
249			self._exports[e].exit()
250			# Close plugins
251			for p in self._plugins:
252			self._plugins[p].exit()
253

nicolargo / glances

Push — master ( e61e99...a0b053 )

GlancesStats.getAllViews() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like