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

glances.config.Config.read() A
last analyzed 2024-04-26 07:38 UTC

↳ Parent: glances.config

Complexity

Conditions

Size

Total Lines	19
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	15
nop	1
dl	0
loc	19
rs	9.1832
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# 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 os
import sys
import multiprocessing
from io import open
import re

from glances.globals import ConfigParser, NoOptionError, NoSectionError, system_exec, BSD, LINUX, MACOS, SUNOS, WINDOWS
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(object):

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

nicolargo / glances

glances.config.Config.read() A last analyzed 2024-04-26 07:38 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.config.Config.read() A
last analyzed 2024-04-26 07:38 UTC