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

Completed

Push — master ( 3dcd25...20576f )

by Nicolas

created 2016-09-10 09:14 UTC

GlancesStats.load_plugins_and_exports() A

↳ Parent: GlancesStats.load_modules()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	14
rs	9.4285
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2016 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(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):
        """Return the plugins list."""
        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:
            # logger.debug("Update %s stats" % p)
            self._plugins[p].update()

    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) 2016 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(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			# Load the plugins
76			self.load_plugins(args=args)
77
78			# Init the export modules dict
79			self._exports = collections.defaultdict(dict)
80			# Load the export modules
81			self.load_exports(args=args)
82
83			# Restoring system path
84			sys.path = sys_path
85
86			def load_plugins(self, args=None):
87			"""Load all plugins in the 'plugins' folder."""
88			header = "glances_"
89			for item in os.listdir(plugins_path):
90			if (item.startswith(header) and
91			item.endswith(".py") and
92			item != (header + "plugin.py")):
93			# Import the plugin
94			plugin = __import__(os.path.basename(item)[:-3])
95			# Add the plugin to the dictionary
96			# The key is the plugin name
97			# for example, the file glances_xxx.py
98			# generate self._plugins_list["xxx"] = ...
99			plugin_name = os.path.basename(item)[len(header):-3].lower()
100			if plugin_name in ('help', 'amps', 'ports'):
101			self._plugins[plugin_name] = plugin.Plugin(args=args, config=self.config)
102			else:
103			self._plugins[plugin_name] = plugin.Plugin(args=args)
104			# Log plugins list
105			logger.debug("Available plugins list: {}".format(self.getAllPlugins()))
106
107			def load_exports(self, args=None):
108			"""Load all export modules in the 'exports' folder."""
109			if args is None:
110			return False
111			header = "glances_"
112			# Transform the arguments list into a dict
113			# The aim is to chec if the export module should be loaded
114			args_var = vars(locals()['args'])
115			for item in os.listdir(exports_path):
116			export_name = os.path.basename(item)[len(header):-3].lower()
117			if (item.startswith(header) and
118			item.endswith(".py") and
119			item != (header + "export.py") and
120			item != (header + "history.py") and
121			args_var['export_' + export_name] is not None and
122			args_var['export_' + export_name] is not False):
123			# Import the export module
124			export_module = __import__(os.path.basename(item)[:-3])
125			# Add the export to the dictionary
126			# The key is the module name
127			# for example, the file glances_xxx.py
128			# generate self._exports_list["xxx"] = ...
129			self._exports[export_name] = export_module.Export(args=args, config=self.config)
130			# Log plugins list
131			logger.debug("Available exports modules list: {}".format(self.getExportList()))
132			return True
133
134			def getAllPlugins(self):
135			"""Return the plugins list."""
136			return [p for p in self._plugins]
137
138			def getExportList(self):
139			"""Return the exports modules list."""
140			return [e for e in self._exports]
141
142			def load_limits(self, config=None):
143			"""Load the stats limits (except the one in the exclude list)."""
144			# For each plugins, call the load_limits method
145			for p in self._plugins:
146			self._plugins[p].load_limits(config)
147
148			def update(self):
149			"""Wrapper method to update the stats."""
150			# For standalone and server modes
151			# For each plugins, call the update method
152			for p in self._plugins:
153			# logger.debug("Update %s stats" % p)
154			self._plugins[p].update()
155
156			def export(self, input_stats=None):
157			"""Export all the stats.
158
159			Each export module is ran in a dedicated thread.
160			"""
161			# threads = []
162			input_stats = input_stats or {}
163
164			for e in self._exports:
165			logger.debug("Export stats using the %s module" % e)
166			thread = threading.Thread(target=self._exports[e].update,
167			args=(input_stats,))
168			# threads.append(thread)
169			thread.start()
170
171			def getAll(self):
172			"""Return all the stats (list)."""
173			return [self._plugins[p].get_raw() for p in self._plugins]
174
175			def getAllExports(self):
176			"""
177			Return all the stats to be exported (list).
178			Default behavor is to export all the stat
179			"""
180			return [self._plugins[p].get_export() for p in self._plugins]
181
182			def getAllAsDict(self):
183			"""Return all the stats (dict)."""
184			# Python > 2.6
185			# {p: self._plugins[p].get_raw() for p in self._plugins}
186			ret = {}
187			for p in self._plugins:
188			ret[p] = self._plugins[p].get_raw()
189			return ret
190
191			def getAllLimits(self):
192			"""Return the plugins limits list."""
193			return [self._plugins[p].limits for p in self._plugins]
194
195			def getAllLimitsAsDict(self):
196			"""Return all the stats limits (dict)."""
197			ret = {}
198			for p in self._plugins:
199			ret[p] = self._plugins[p].limits
200			return ret
201
202			def getAllViews(self):
203			"""Return the plugins views."""
204			return [self._plugins[p].get_views() for p in self._plugins]
205
206			def getAllViewsAsDict(self):
207			"""Return all the stats views (dict)."""
208			ret = {}
209			for p in self._plugins:
210			ret[p] = self._plugins[p].get_views()
211			return ret
212
213			def get_plugin_list(self):
214			"""Return the plugin list."""
215			return self._plugins
216
217			def get_plugin(self, plugin_name):
218			"""Return the plugin name."""
219			if plugin_name in self._plugins:
220			return self._plugins[plugin_name]
221			else:
222			return None
223
224			def end(self):
225			"""End of the Glances stats."""
226			# Close export modules
227			for e in self._exports:
228			self._exports[e].exit()
229			# Close plugins
230			for p in self._plugins:
231			self._plugins[p].exit()
232

nicolargo / glances

Push — master ( 3dcd25...20576f )

GlancesStats.load_plugins_and_exports() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like