nicolargo / glances

glances/amps/glances_amp.py

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <nicolas@nicolargo.com>
#
# 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/>.

"""
I am your father...

...for all Glances Application Monitoring Processes (AMP).

AMP (Application Monitoring Process)
A Glances AMP is a Python script called (every *refresh* seconds) if:
- the AMP is *enabled* in the Glances configuration file
- a process is running (match the *regex* define in the configuration file)
The script should define a Amp (GlancesAmp) class with, at least, an update method.

This line is too long as per the coding-style (83/80).

The update method should call the set_result method to set the AMP return string.

This line is too long as per the coding-style (81/80).

The return string is a string with one or more line (\n between lines).
If the *one_line* var is true then the AMP will be displayed in one line.
"""

from glances.compat import u, b, n, nativestr

import missing from __future__ import absolute_import

Unused nativestr imported from glances.compat

Unused b imported from glances.compat

Unused n imported from glances.compat

from glances.timer import Timer

import missing from __future__ import absolute_import

from glances.logger import logger

import missing from __future__ import absolute_import

class GlancesAmp(object):
    """Main class for Glances AMP."""

    NAME = '?'
    VERSION = '?'
    DESCRIPTION = '?'
    AUTHOR = '?'
    EMAIL = '?'

    def __init__(self, name=None, args=None):
        """Init AMP classe."""
        logger.debug("AMP - Init {} version {}".format(self.NAME, self.VERSION))

Use formatting in logging functions and pass the parameters as arguments

        # AMP name (= module name without glances_)
        if name is None:
            self.amp_name = self.__class__.__module__[len('glances_'):]
        else:
            self.amp_name = name

        # Init the args
        self.args = args

        # Init the configs
        self.configs = {}

        # A timer is needed to only update every refresh seconds
        # Init to 0 in order to update the AMP on startup
        self.timer = Timer(0)

    def load_config(self, config):
        """Load AMP parameters from the configuration file."""

        # Read AMP confifuration.
        # For ex, the AMP foo should have the following section:
        #
        # [foo]
        # enable=true
        # regex=\/usr\/bin\/nginx
        # refresh=60
        #
        # and optionnaly:
        #
        # one_line=false
        # option1=opt1
        # ...
        #
        amp_section = 'amp_' + self.amp_name
        if (hasattr(config, 'has_section') and
                config.has_section(amp_section)):

Wrong continued indentation (remove 4 spaces).

            logger.debug("AMP - {}: Load configuration".format(self.NAME))

Use formatting in logging functions and pass the parameters as arguments

            for param, _ in config.items(amp_section):
                try:
                    self.configs[param] = config.get_float_value(amp_section, param)

This line is too long as per the coding-style (84/80).

                except ValueError:
                    self.configs[param] = config.get_value(amp_section, param).split(',')

This line is too long as per the coding-style (89/80).

                    if len(self.configs[param]) == 1:
                        self.configs[param] = self.configs[param][0]
                logger.debug("AMP - {}: Load parameter: {} = {}".format(self.NAME, param, self.configs[param]))

This line is too long as per the coding-style (111/80).

Use formatting in logging functions and pass the parameters as arguments

        else:
            logger.debug("AMP - {}: Can not find section {} in the configuration file".format(self.NAME, self.amp_name))

This line is too long as per the coding-style (120/80).

Use formatting in logging functions and pass the parameters as arguments

            return False

        if self.enable():
            # Refresh option is mandatory
            for k in ['refresh']:
                if k not in self.configs:
                    logger.warning("AMP - {}: Can not find configuration key {} in section {} (the AMP will be disabled)".format(self.NAME, k, self.amp_name))

This line is too long as per the coding-style (158/80).

Use formatting in logging functions and pass the parameters as arguments

                    self.configs['enable'] = 'false'
        else:
            logger.debug("AMP - {} is disabled".format(self.NAME))

Use formatting in logging functions and pass the parameters as arguments

        # Init the count to 0
        self.configs['count'] = 0

        return self.enable()

    def get(self, key):
        """Generic method to get the item in the AMP configuration"""
        if key in self.configs:

Unnecessary "else" after "return"

            return self.configs[key]
        else:
            return None

    def enable(self):
        """Return True|False if the AMP is enabled in the configuration file (enable=true|false)."""

This line is too long as per the coding-style (100/80).

        ret = self.get('enable')
        if ret is None:

Unnecessary "else" after "return"

            return False
        else:
            return ret.lower().startswith('true')

    def regex(self):
        """Return regular expression used to identified the current application."""

This line is too long as per the coding-style (83/80).

        return self.get('regex')

    def refresh(self):
        """Return refresh time in seconds for the current application monitoring process."""

This line is too long as per the coding-style (92/80).

        return self.get('refresh')

    def one_line(self):
        """Return True|False if the AMP shoukd be displayed in oneline (one_lineline=true|false)."""

This line is too long as per the coding-style (100/80).

        ret = self.get('one_line')
        if ret is None:

Unnecessary "else" after "return"

            return False
        else:
            return ret.lower().startswith('true')

    def time_until_refresh(self):
        """Return time in seconds until refresh."""
        return self.timer.get()

    def should_update(self):
        """Return True is the AMP should be updated:
        - AMP is enable
        - only update every 'refresh' seconds
        """
        if self.timer.finished():
            self.timer.set(self.refresh())
            self.timer.reset()
            return self.enable()
        return False

    def set_count(self, count):
        """Set the number of processes matching the regex"""
        self.configs['count'] = count

    def count(self):
        """Get the number of processes matching the regex"""
        return self.get('count')

    def count_min(self):
        """Get the minimum number of processes"""
        return self.get('countmin')

    def count_max(self):
        """Get the maximum number of processes"""
        return self.get('countmax')

    def set_result(self, result, separator=''):
        """Store the result (string) into the result key of the AMP
        if one_line is true then replace \n by separator
        """
        if self.one_line():
            self.configs['result'] = u(result).replace('\n', separator)
        else:
            self.configs['result'] = u(result)

    def result(self):
        """ Return the result of the AMP (as a string)"""
        ret = self.get('result')
        if ret is not None:
            ret = u(ret)
        return ret

    def update_wrapper(self, process_list):
        """Wrapper for the children update"""
        # Set the number of running process
        self.set_count(len(process_list))
        # Call the children update method
        if self.should_update():

Unnecessary "else" after "return"

            return self.update(process_list)

The Instance of GlancesAmp does not seem to have a member named update.

        else:
            return self.result()