glances.config.Config.__init__() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

glances.config.Config.init() A
last analyzed 2024-05-18 09:29 UTC

↳ Parent: glances.config

Complexity

Conditions

Size

Total Lines	14
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	10
nop	2
dl	0
loc	14
rs	9.9
c	0
b	0
f	0

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Manage the configuration file."""

import builtins
import multiprocessing
import os
import re
import sys

from glances.globals import BSD, LINUX, MACOS, SUNOS, WINDOWS, ConfigParser, NoOptionError, NoSectionError, system_exec
from glances.logger import logger


def user_config_dir():
    r"""Return a list of per-user config dir (full path).

    - Linux, *BSD, SunOS: ~/.config/glances
    - macOS: ~/Library/Application Support/glances
    - Windows: %APPDATA%\glances
    """
    paths = []
    if WINDOWS:
        paths.append(os.environ.get('APPDATA'))
    elif MACOS:
        paths.append(os.environ.get('XDG_CONFIG_HOME') or os.path.expanduser('~/.config'))
        paths.append(os.path.expanduser('~/Library/Application Support'))
    else:
        paths.append(os.environ.get('XDG_CONFIG_HOME') or os.path.expanduser('~/.config'))

    return [os.path.join(path, 'glances') if path is not None else '' for path in paths]


def user_cache_dir():
    r"""Return a list of per-user cache dir (full path).

    - Linux, *BSD, SunOS: ~/.cache/glances
    - macOS: ~/Library/Caches/glances
    - Windows: {%LOCALAPPDATA%,%APPDATA%}\glances\cache
    """
    if WINDOWS:
        path = os.path.join(os.environ.get('LOCALAPPDATA') or os.environ.get('APPDATA'), 'glances', 'cache')
    elif MACOS:
        path = os.path.expanduser('~/Library/Caches/glances')
    else:
        path = os.path.join(os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'), 'glances')

    return [path]


def system_config_dir():
    r"""Return a list of system-wide config dir (full path).

    - Linux, SunOS: /etc/glances
    - *BSD, macOS: /usr/local/etc/glances
    - Windows: %APPDATA%\glances
    """
    if LINUX or SUNOS:
        path = '/etc'
    elif BSD or MACOS:
        path = '/usr/local/etc'
    else:
        path = os.environ.get('APPDATA')
    if path is None:
        path = ''
    else:
        path = os.path.join(path, 'glances')

    return [path]


def default_config_dir():
    r"""Return a list of system-wide config dir (full path).

    - Linux, SunOS, *BSD, macOS: /usr/share/doc (as defined in the setup.py files)
    - Windows: %APPDATA%\glances
    """
    if LINUX or SUNOS or BSD or MACOS:
        path = '/usr/share/doc'
    else:
        path = os.environ.get('APPDATA')
    if path is None:
        path = ''
    else:
        path = os.path.join(path, 'glances')

    return [path]


class Config:
    """This class is used to access/read config file, if it exists.

    :param config_dir: the path to search for config file
    :type config_dir: str or None
    """

    def __init__(self, config_dir=None):
        self.config_dir = config_dir
        self.config_filename = 'glances.conf'
        self._loaded_config_file = None

        # Re pattern for optimize research of `foo`
        self.re_pattern = re.compile(r'(\`.+?\`)')

        try:
            self.parser = ConfigParser(interpolation=None)
        except TypeError:
            self.parser = ConfigParser()

        self.read()

    def config_file_paths(self):
        r"""Get a list of config file paths.

        The list is built taking into account of the OS, priority and location.

        * custom path: /path/to/glances
        * Linux, SunOS: ~/.config/glances, /etc/glances
        * *BSD: ~/.config/glances, /usr/local/etc/glances
        * macOS: ~/.config/glances, ~/Library/Application Support/glances, /usr/local/etc/glances
        * Windows: %APPDATA%\glances

        The config file will be searched in the following order of priority:
            * /path/to/file (via -C flag)
            * user's home directory (per-user settings)
            * system-wide directory (system-wide settings)
            * default pip directory (as defined in the setup.py file)
        """
        paths = []

        # self.config_dir is the path to the config file (via -C flag)
        if self.config_dir:
            paths.append(self.config_dir)

        # user_config_dir() returns a list of paths
        paths.extend([os.path.join(path, self.config_filename) for path in user_config_dir()])

        # system_config_dir() returns a list of paths
        paths.extend([os.path.join(path, self.config_filename) for path in system_config_dir()])

        # default_config_dir() returns a list of paths
        paths.extend([os.path.join(path, self.config_filename) for path in default_config_dir()])

        return paths

    def read(self):
        """Read the config file, if it exists. Using defaults otherwise."""
        for config_file in self.config_file_paths():
            logger.debug(f'Search glances.conf file in {config_file}')
            if os.path.exists(config_file):
                try:
                    with builtins.open(config_file, encoding='utf-8') as f:
                        self.parser.read_file(f)
                        self.parser.read(f)
                    logger.info(f"Read configuration file '{config_file}'")
                except UnicodeDecodeError as err:
                    logger.error(f"Can not read configuration file '{config_file}': {err}")
                    sys.exit(1)
                # Save the loaded configuration file path (issue #374)
                self._loaded_config_file = config_file
                break

        # Set the default values for section not configured
        self.sections_set_default()

    def sections_set_default(self):
        # Globals
        if not self.parser.has_section('global'):
            self.parser.add_section('global')
        self.set_default('global', 'strftime_format', '')
        self.set_default('global', 'check_update', 'true')

        # Quicklook
        if not self.parser.has_section('quicklook'):
            self.parser.add_section('quicklook')
        self.set_default_cwc('quicklook', 'cpu')
        self.set_default_cwc('quicklook', 'mem')
        self.set_default_cwc('quicklook', 'swap')

        # CPU
        if not self.parser.has_section('cpu'):
            self.parser.add_section('cpu')
        self.set_default_cwc('cpu', 'user')
        self.set_default_cwc('cpu', 'system')
        self.set_default_cwc('cpu', 'steal')
        # By default I/O wait should be lower than 1/number of CPU cores
        iowait_bottleneck = (1.0 / multiprocessing.cpu_count()) * 100.0
        self.set_default_cwc(
            'cpu',
            'iowait',
            [
                str(iowait_bottleneck - (iowait_bottleneck * 0.20)),
                str(iowait_bottleneck - (iowait_bottleneck * 0.10)),
                str(iowait_bottleneck),
            ],
        )
        # Context switches bottleneck identification #1212
        ctx_switches_bottleneck = (500000 * 0.10) * multiprocessing.cpu_count()
        self.set_default_cwc(
            'cpu',
            'ctx_switches',
            [
                str(ctx_switches_bottleneck - (ctx_switches_bottleneck * 0.20)),
                str(ctx_switches_bottleneck - (ctx_switches_bottleneck * 0.10)),
                str(ctx_switches_bottleneck),
            ],
        )

        # Per-CPU
        if not self.parser.has_section('percpu'):
            self.parser.add_section('percpu')
        self.set_default_cwc('percpu', 'user')
        self.set_default_cwc('percpu', 'system')

        # Load
        if not self.parser.has_section('load'):
            self.parser.add_section('load')
        self.set_default_cwc('load', cwc=['0.7', '1.0', '5.0'])

        # Mem
        if not self.parser.has_section('mem'):
            self.parser.add_section('mem')
        self.set_default_cwc('mem')

        # Swap
        if not self.parser.has_section('memswap'):
            self.parser.add_section('memswap')
        self.set_default_cwc('memswap')

        # NETWORK
        if not self.parser.has_section('network'):
            self.parser.add_section('network')
        self.set_default_cwc('network', 'rx')
        self.set_default_cwc('network', 'tx')

        # FS
        if not self.parser.has_section('fs'):
            self.parser.add_section('fs')
        self.set_default_cwc('fs')

        # Sensors
        if not self.parser.has_section('sensors'):
            self.parser.add_section('sensors')
        self.set_default_cwc('sensors', 'temperature_core', cwc=['60', '70', '80'])
        self.set_default_cwc('sensors', 'temperature_hdd', cwc=['45', '52', '60'])
        self.set_default_cwc('sensors', 'battery', cwc=['80', '90', '95'])

        # Process list
        if not self.parser.has_section('processlist'):
            self.parser.add_section('processlist')
        self.set_default_cwc('processlist', 'cpu')
        self.set_default_cwc('processlist', 'mem')

    @property
    def loaded_config_file(self):
        """Return the loaded configuration file."""
        return self._loaded_config_file

    def as_dict(self):
        """Return the configuration as a dict"""
        dictionary = {}
        for section in self.parser.sections():
            dictionary[section] = {}
            for option in self.parser.options(section):
                dictionary[section][option] = self.parser.get(section, option)
        return dictionary

    def sections(self):
        """Return a list of all sections."""
        return self.parser.sections()

    def items(self, section):
        """Return the items list of a section."""
        return self.parser.items(section)

    def has_section(self, section):
        """Return info about the existence of a section."""
        return self.parser.has_section(section)

    def set_default_cwc(self, section, option_header=None, cwc=['50', '70', '90']):
        """Set default values for careful, warning and critical."""
        if option_header is None:
            header = ''
        else:
            header = option_header + '_'
        self.set_default(section, header + 'careful', cwc[0])
        self.set_default(section, header + 'warning', cwc[1])
        self.set_default(section, header + 'critical', cwc[2])

    def set_default(self, section, option, default):
        """If the option did not exist, create a default value."""
        if not self.parser.has_option(section, option):
            self.parser.set(section, option, default)

    def get_value(self, section, option, default=None):
        """Get the value of an option, if it exists.

        If it did not exist, then return the default value.

        It allows user to define dynamic configuration key (see issue#1204)
        Dynamic value should starts and end with the ` char
        Example: prefix=`hostname`
        """
        ret = default
        try:
            ret = self.parser.get(section, option)
        except (NoOptionError, NoSectionError):
            pass

        # Search a substring `foo` and replace it by the result of its exec
        if ret is not None:
            try:
                match = self.re_pattern.findall(ret)
                for m in match:
                    ret = ret.replace(m, system_exec(m[1:-1]))
            except TypeError:
                pass
        return ret

    def get_list_value(self, section, option, default=None, separator=','):
        """Get the list value of an option, if it exists."""
        try:
            return self.parser.get(section, option).split(separator)
        except (NoOptionError, NoSectionError):
            return default

    def get_int_value(self, section, option, default=0):
        """Get the int value of an option, if it exists."""
        try:
            return self.parser.getint(section, option)
        except (NoOptionError, NoSectionError):
            return int(default)

    def get_float_value(self, section, option, default=0.0):
        """Get the float value of an option, if it exists."""
        try:
            return self.parser.getfloat(section, option)
        except (NoOptionError, NoSectionError):
            return float(default)

    def get_bool_value(self, section, option, default=True):
        """Get the bool value of an option, if it exists."""
        try:
            return self.parser.getboolean(section, option)
        except (NoOptionError, NoSectionError):
            return bool(default)


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""Manage the configuration file."""
10
11			import builtins
12			import multiprocessing
13			import os
14			import re
15			import sys
16
17			from glances.globals import BSD, LINUX, MACOS, SUNOS, WINDOWS, ConfigParser, NoOptionError, NoSectionError, system_exec
18			from glances.logger import logger
19
20
21			def user_config_dir():
22			r"""Return a list of per-user config dir (full path).
23
24			- Linux, *BSD, SunOS: ~/.config/glances
25			- macOS: ~/Library/Application Support/glances
26			- Windows: %APPDATA%\glances
27			"""
28			paths = []
29			if WINDOWS:
30			paths.append(os.environ.get('APPDATA'))
31			elif MACOS:
32			paths.append(os.environ.get('XDG_CONFIG_HOME') or os.path.expanduser('~/.config'))
33			paths.append(os.path.expanduser('~/Library/Application Support'))
34			else:
35			paths.append(os.environ.get('XDG_CONFIG_HOME') or os.path.expanduser('~/.config'))
36
37			return [os.path.join(path, 'glances') if path is not None else '' for path in paths]
38
39
40			def user_cache_dir():
41			r"""Return a list of per-user cache dir (full path).
42
43			- Linux, *BSD, SunOS: ~/.cache/glances
44			- macOS: ~/Library/Caches/glances
45			- Windows: {%LOCALAPPDATA%,%APPDATA%}\glances\cache
46			"""
47			if WINDOWS:
48			path = os.path.join(os.environ.get('LOCALAPPDATA') or os.environ.get('APPDATA'), 'glances', 'cache')
49			elif MACOS:
50			path = os.path.expanduser('~/Library/Caches/glances')
51			else:
52			path = os.path.join(os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'), 'glances')
53
54			return [path]
55
56
57			def system_config_dir():
58			r"""Return a list of system-wide config dir (full path).
59
60			- Linux, SunOS: /etc/glances
61			- *BSD, macOS: /usr/local/etc/glances
62			- Windows: %APPDATA%\glances
63			"""
64			if LINUX or SUNOS:
65			path = '/etc'
66			elif BSD or MACOS:
67			path = '/usr/local/etc'
68			else:
69			path = os.environ.get('APPDATA')
70			if path is None:
71			path = ''
72			else:
73			path = os.path.join(path, 'glances')
74
75			return [path]
76
77
78			def default_config_dir():
79			r"""Return a list of system-wide config dir (full path).
80
81			- Linux, SunOS, *BSD, macOS: /usr/share/doc (as defined in the setup.py files)
82			- Windows: %APPDATA%\glances
83			"""
84			if LINUX or SUNOS or BSD or MACOS:
85			path = '/usr/share/doc'
86			else:
87			path = os.environ.get('APPDATA')
88			if path is None:
89			path = ''
90			else:
91			path = os.path.join(path, 'glances')
92
93			return [path]
94
95
96			class Config:
97			"""This class is used to access/read config file, if it exists.
98
99			:param config_dir: the path to search for config file
100			:type config_dir: str or None
101			"""
102
103			def __init__(self, config_dir=None):
104			self.config_dir = config_dir
105			self.config_filename = 'glances.conf'
106			self._loaded_config_file = None
107
108			# Re pattern for optimize research of `foo`
109			self.re_pattern = re.compile(r'(\`.+?\`)')
110
111			try:
112			self.parser = ConfigParser(interpolation=None)
113			except TypeError:
114			self.parser = ConfigParser()
115
116			self.read()
117
118			def config_file_paths(self):
119			r"""Get a list of config file paths.
120
121			The list is built taking into account of the OS, priority and location.
122
123			* custom path: /path/to/glances
124			* Linux, SunOS: ~/.config/glances, /etc/glances
125			* *BSD: ~/.config/glances, /usr/local/etc/glances
126			* macOS: ~/.config/glances, ~/Library/Application Support/glances, /usr/local/etc/glances
127			* Windows: %APPDATA%\glances
128
129			The config file will be searched in the following order of priority:
130			* /path/to/file (via -C flag)
131			* user's home directory (per-user settings)
132			* system-wide directory (system-wide settings)
133			* default pip directory (as defined in the setup.py file)
134			"""
135			paths = []
136
137			# self.config_dir is the path to the config file (via -C flag)
138			if self.config_dir:
139			paths.append(self.config_dir)
140
141			# user_config_dir() returns a list of paths
142			paths.extend([os.path.join(path, self.config_filename) for path in user_config_dir()])
143
144			# system_config_dir() returns a list of paths
145			paths.extend([os.path.join(path, self.config_filename) for path in system_config_dir()])
146
147			# default_config_dir() returns a list of paths
148			paths.extend([os.path.join(path, self.config_filename) for path in default_config_dir()])
149
150			return paths
151
152			def read(self):
153			"""Read the config file, if it exists. Using defaults otherwise."""
154			for config_file in self.config_file_paths():
155			logger.debug(f'Search glances.conf file in {config_file}')
156			if os.path.exists(config_file):
157			try:
158			with builtins.open(config_file, encoding='utf-8') as f:
159			self.parser.read_file(f)
160			self.parser.read(f)
161			logger.info(f"Read configuration file '{config_file}'")
162			except UnicodeDecodeError as err:
163			logger.error(f"Can not read configuration file '{config_file}': {err}")
164			sys.exit(1)
165			# Save the loaded configuration file path (issue #374)
166			self._loaded_config_file = config_file
167			break
168
169			# Set the default values for section not configured
170			self.sections_set_default()
171
172			def sections_set_default(self):
173			# Globals
174			if not self.parser.has_section('global'):
175			self.parser.add_section('global')
176			self.set_default('global', 'strftime_format', '')
177			self.set_default('global', 'check_update', 'true')
178
179			# Quicklook
180			if not self.parser.has_section('quicklook'):
181			self.parser.add_section('quicklook')
182			self.set_default_cwc('quicklook', 'cpu')
183			self.set_default_cwc('quicklook', 'mem')
184			self.set_default_cwc('quicklook', 'swap')
185
186			# CPU
187			if not self.parser.has_section('cpu'):
188			self.parser.add_section('cpu')
189			self.set_default_cwc('cpu', 'user')
190			self.set_default_cwc('cpu', 'system')
191			self.set_default_cwc('cpu', 'steal')
192			# By default I/O wait should be lower than 1/number of CPU cores
193			iowait_bottleneck = (1.0 / multiprocessing.cpu_count()) * 100.0
194			self.set_default_cwc(
195			'cpu',
196			'iowait',
197			[
198			str(iowait_bottleneck - (iowait_bottleneck * 0.20)),
199			str(iowait_bottleneck - (iowait_bottleneck * 0.10)),
200			str(iowait_bottleneck),
201			],
202			)
203			# Context switches bottleneck identification #1212
204			ctx_switches_bottleneck = (500000 * 0.10) * multiprocessing.cpu_count()
205			self.set_default_cwc(
206			'cpu',
207			'ctx_switches',
208			[
209			str(ctx_switches_bottleneck - (ctx_switches_bottleneck * 0.20)),
210			str(ctx_switches_bottleneck - (ctx_switches_bottleneck * 0.10)),
211			str(ctx_switches_bottleneck),
212			],
213			)
214
215			# Per-CPU
216			if not self.parser.has_section('percpu'):
217			self.parser.add_section('percpu')
218			self.set_default_cwc('percpu', 'user')
219			self.set_default_cwc('percpu', 'system')
220
221			# Load
222			if not self.parser.has_section('load'):
223			self.parser.add_section('load')
224			self.set_default_cwc('load', cwc=['0.7', '1.0', '5.0'])
225
226			# Mem
227			if not self.parser.has_section('mem'):
228			self.parser.add_section('mem')
229			self.set_default_cwc('mem')
230
231			# Swap
232			if not self.parser.has_section('memswap'):
233			self.parser.add_section('memswap')
234			self.set_default_cwc('memswap')
235
236			# NETWORK
237			if not self.parser.has_section('network'):
238			self.parser.add_section('network')
239			self.set_default_cwc('network', 'rx')
240			self.set_default_cwc('network', 'tx')
241
242			# FS
243			if not self.parser.has_section('fs'):
244			self.parser.add_section('fs')
245			self.set_default_cwc('fs')
246
247			# Sensors
248			if not self.parser.has_section('sensors'):
249			self.parser.add_section('sensors')
250			self.set_default_cwc('sensors', 'temperature_core', cwc=['60', '70', '80'])
251			self.set_default_cwc('sensors', 'temperature_hdd', cwc=['45', '52', '60'])
252			self.set_default_cwc('sensors', 'battery', cwc=['80', '90', '95'])
253
254			# Process list
255			if not self.parser.has_section('processlist'):
256			self.parser.add_section('processlist')
257			self.set_default_cwc('processlist', 'cpu')
258			self.set_default_cwc('processlist', 'mem')
259
260			@property
261			def loaded_config_file(self):
262			"""Return the loaded configuration file."""
263			return self._loaded_config_file
264
265			def as_dict(self):
266			"""Return the configuration as a dict"""
267			dictionary = {}
268			for section in self.parser.sections():
269			dictionary[section] = {}
270			for option in self.parser.options(section):
271			dictionary[section][option] = self.parser.get(section, option)
272			return dictionary
273
274			def sections(self):
275			"""Return a list of all sections."""
276			return self.parser.sections()
277
278			def items(self, section):
279			"""Return the items list of a section."""
280			return self.parser.items(section)
281
282			def has_section(self, section):
283			"""Return info about the existence of a section."""
284			return self.parser.has_section(section)
285
286			def set_default_cwc(self, section, option_header=None, cwc=['50', '70', '90']):
287			"""Set default values for careful, warning and critical."""
288			if option_header is None:
289			header = ''
290			else:
291			header = option_header + '_'
292			self.set_default(section, header + 'careful', cwc[0])
293			self.set_default(section, header + 'warning', cwc[1])
294			self.set_default(section, header + 'critical', cwc[2])
295
296			def set_default(self, section, option, default):
297			"""If the option did not exist, create a default value."""
298			if not self.parser.has_option(section, option):
299			self.parser.set(section, option, default)
300
301			def get_value(self, section, option, default=None):
302			"""Get the value of an option, if it exists.
303
304			If it did not exist, then return the default value.
305
306			It allows user to define dynamic configuration key (see issue#1204)
307			Dynamic value should starts and end with the ` char
308			Example: prefix=`hostname`
309			"""
310			ret = default
311			try:
312			ret = self.parser.get(section, option)
313			except (NoOptionError, NoSectionError):
314			pass
315
316			# Search a substring `foo` and replace it by the result of its exec
317			if ret is not None:
318			try:
319			match = self.re_pattern.findall(ret)
320			for m in match:
321			ret = ret.replace(m, system_exec(m[1:-1]))
322			except TypeError:
323			pass
324			return ret
325
326			def get_list_value(self, section, option, default=None, separator=','):
327			"""Get the list value of an option, if it exists."""
328			try:
329			return self.parser.get(section, option).split(separator)
330			except (NoOptionError, NoSectionError):
331			return default
332
333			def get_int_value(self, section, option, default=0):
334			"""Get the int value of an option, if it exists."""
335			try:
336			return self.parser.getint(section, option)
337			except (NoOptionError, NoSectionError):
338			return int(default)
339
340			def get_float_value(self, section, option, default=0.0):
341			"""Get the float value of an option, if it exists."""
342			try:
343			return self.parser.getfloat(section, option)
344			except (NoOptionError, NoSectionError):
345			return float(default)
346
347			def get_bool_value(self, section, option, default=True):
348			"""Get the bool value of an option, if it exists."""
349			try:
350			return self.parser.getboolean(section, option)
351			except (NoOptionError, NoSectionError):
352			return bool(default)
353

nicolargo / glances

glances.config.Config.__init__() A last analyzed 2024-05-18 09:29 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.config.Config.init() A
last analyzed 2024-05-18 09:29 UTC