GlancesStats.getAllPlugins() - Code Metrics - Inspection of "Generate documentation for Glances 2.8" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2b80fa...6ea077 )

by Nicolas

created 2017-01-21 13:48 UTC

GlancesStats.getAllPlugins() B

↳ Parent: GlancesStats

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	7
rs	8.5454

# -*- 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 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 getAllAsDict(self):
        """Return all the stats (dict)."""
        # Python > 2.6
        # {p: self._plugins[p].get_raw() for p in self._plugins}
        ret = {}
        for p in self._plugins:
            ret[p] = self._plugins[p].get_raw()
        return ret

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

    def getAllLimitsAsDict(self):
        """Return all the stats limits (dict)."""
        ret = {}
        for p in self._plugins:
            ret[p] = self._plugins[p].limits
        return ret

    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)."""
        ret = {}
        for p in self._plugins:
            ret[p] = self._plugins[p].get_views()
        return ret

    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 getAllExports(self):
190			"""
191			Return all the stats to be exported (list).
192			Default behavor is to export all the stat
193			"""
194			return [self._plugins[p].get_export() for p in self._plugins]
195
196			def getAllAsDict(self):
197			"""Return all the stats (dict)."""
198			# Python > 2.6
199			# {p: self._plugins[p].get_raw() for p in self._plugins}
200			ret = {}
201			for p in self._plugins:
202			ret[p] = self._plugins[p].get_raw()
203			return ret
204
205			def getAllLimits(self):
206			"""Return the plugins limits list."""
207			return [self._plugins[p].limits for p in self._plugins]
208
209			def getAllLimitsAsDict(self):
210			"""Return all the stats limits (dict)."""
211			ret = {}
212			for p in self._plugins:
213			ret[p] = self._plugins[p].limits
214			return ret
215
216			def getAllViews(self):
217			"""Return the plugins views."""
218			return [self._plugins[p].get_views() for p in self._plugins]
219
220			def getAllViewsAsDict(self):
221			"""Return all the stats views (dict)."""
222			ret = {}
223			for p in self._plugins:
224			ret[p] = self._plugins[p].get_views()
225			return ret
226
227			def get_plugin_list(self):
228			"""Return the plugin list."""
229			return self._plugins
230
231			def get_plugin(self, plugin_name):
232			"""Return the plugin name."""
233			if plugin_name in self._plugins:
234			return self._plugins[plugin_name]
235			else:
236			return None
237
238			def end(self):
239			"""End of the Glances stats."""
240			# Close export modules
241			for e in self._exports:
242			self._exports[e].exit()
243			# Close plugins
244			for p in self._plugins:
245			self._plugins[p].exit()
246

nicolargo / glances

Push — master ( 2b80fa...6ea077 )

GlancesStats.getAllPlugins() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like