GlancesIRQ.__update() - Code Metrics - Inspection of "Develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#963)

unknown

created 2016-11-11 15:41 UTC

GlancesIRQ.__update() B

↳ Parent: GlancesIRQ

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	33
rs	8.0894

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2016 Angelo Poerio <[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/>.

"""IRQ plugin."""

import os
import operator

from glances.logger import logger
from glances.globals import LINUX
from glances.timer import getTimeSinceLastUpdate
from glances.plugins.glances_plugin import GlancesPlugin


class Plugin(GlancesPlugin):

    """Glances IRQ plugin.

    stats is a list
    """

    def __init__(self, args=None):
        """Init the plugin."""
        super(Plugin, self).__init__(args=args)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Init the stats
        self.irq = GlancesIRQ()
        self.reset()

    def get_key(self):
        """Return the key of the list."""
        return self.irq.get_key()

    def reset(self):
        """Reset/init the stats."""
        self.stats = []

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update the IRQ stats"""

        # Reset the list
        self.reset()

        # IRQ plugin only available on GNU/Linux
        if not LINUX:
            return self.stats

        if self.input_method == 'local':
            # Grab the stats
            self.stats = self.irq.get()

        elif self.input_method == 'snmp':
            # not available
            pass

        # Get the TOP 5
        self.stats = sorted(self.stats, key=operator.itemgetter(
            'irq_rate'), reverse=True)[:5]  # top 5 IRQ by rate/s

        return self.stats

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super(Plugin, self).update_views()

    def msg_curse(self, args=None, max_width=None):
        """Return the dict to display in the curse interface."""
        # Init the return message
        ret = []

        if not LINUX:  # only available on GNU/Linux
            return ret

        # Only process if stats exist and display plugin enable...
        if not self.stats or self.is_disable():
            return ret

        if max_width is not None and max_width >= 23:
            irq_max_width = max_width - 14
        else:
            irq_max_width = 9

        # Build the string message
        # Header
        msg = '{:{width}}'.format('IRQ', width=irq_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = '{:>14}'.format('Rate/s')
        ret.append(self.curse_add_line(msg))

        for i in self.stats:
            ret.append(self.curse_new_line())
            msg = '{:<15}'.format(i['irq_line'])
            ret.append(self.curse_add_line(msg))
            msg = '{:>8}'.format(str(i['irq_rate']))
            ret.append(self.curse_add_line(msg))

        return ret


class GlancesIRQ(object):
    """
    This class manages the IRQ file
    """

    IRQ_FILE = '/proc/interrupts'

    def __init__(self):
        """
        Init the class
        The stat are stored in a internal list of dict
        """
        self.lasts = {}
        self.reset()

    def reset(self):
        """Reset the stats"""
        self.stats = []
        self.cpu_number = 0

    def get(self):
        """Return the current IRQ stats"""
        return self.__update()

    def get_key(self):
        """Return the key of the dict."""
        return 'irq_line'

    def __header(self, line):
        """The header contain the number of CPU

        CPU0       CPU1       CPU2       CPU3
        0:         21          0          0          0   IO-APIC   2-edge      timer
        """
        self.cpu_number = len(line.split())
        return self.cpu_number

    def __humanname(self, line):
        """Get a line and
        Return the IRQ name, alias or number (choose the best for human)

        IRQ line samples:
        1:      44487        341         44         72   IO-APIC   1-edge      i8042
        LOC:   33549868   22394684   32474570   21855077   Local timer interrupts
        """
        splitted_line = line.split()
        irq_line = splitted_line[0].replace(':', '')
        if irq_line.isdigit():
            # If the first column is a digit, use the alias (last column)
            irq_line += '_{}'.format(splitted_line[-1])
        return irq_line

    def __sum(self, line):
        """Get a line and
        Return the IRQ sum number

        IRQ line samples:
        1:      44487        341         44         72   IO-APIC   1-edge      i8042
        LOC:   33549868   22394684   32474570   21855077   Local timer interrupts
        """
        splitted_line = line.split()
        # sum interrupts on all CPUs
        return sum(map(int, splitted_line[1:(self.cpu_number + 1)]))

    def __update(self):
        """
        Load the IRQ file and update the internal dict
        """

        self.reset()

        if not os.path.exists(self.IRQ_FILE):
            # Correct issue #947: IRQ file do not exist on OpenVZ container
            return self.stats

        with open(self.IRQ_FILE) as irq_proc:
            time_since_update = getTimeSinceLastUpdate('irq')
            # Read the header
            self.__header(irq_proc.readline())
            # Read the rest of the lines (one line per IRQ)
            for line in irq_proc.readlines():
                irq_line = self.__humanname(line)
                current_irqs = self.__sum(line)
                irq_rate = int(
                    current_irqs -
                    self.lasts.get(irq_line) if self.lasts.get(irq_line) else 0 //
                    time_since_update)
                irq_current = {
                    'irq_line': irq_line,
                    'irq_rate': irq_rate,
                    'key': self.get_key(),
                    'time_since_update': time_since_update
                }
                self.stats.append(irq_current)
                self.lasts[irq_line] = current_irqs

        return self.stats


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2016 Angelo Poerio <[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			"""IRQ plugin."""
21
22			import os
23			import operator
24
25			from glances.logger import logger
26			from glances.globals import LINUX
27			from glances.timer import getTimeSinceLastUpdate
28			from glances.plugins.glances_plugin import GlancesPlugin
29
30
31			class Plugin(GlancesPlugin):
32
33			"""Glances IRQ plugin.
34
35			stats is a list
36			"""
37
38			def __init__(self, args=None):
39			"""Init the plugin."""
40			super(Plugin, self).__init__(args=args)
41
42			# We want to display the stat in the curse interface
43			self.display_curse = True
44
45			# Init the stats
46			self.irq = GlancesIRQ()
47			self.reset()
48
49			def get_key(self):
50			"""Return the key of the list."""
51			return self.irq.get_key()
52
53			def reset(self):
54			"""Reset/init the stats."""
55			self.stats = []
56
57			@GlancesPlugin._check_decorator
58			@GlancesPlugin._log_result_decorator
59			def update(self):
60			"""Update the IRQ stats"""
61
62			# Reset the list
63			self.reset()
64
65			# IRQ plugin only available on GNU/Linux
66			if not LINUX:
67			return self.stats
68
69			if self.input_method == 'local':
70			# Grab the stats
71			self.stats = self.irq.get()
72
73			elif self.input_method == 'snmp':
74			# not available
75			pass
76
77			# Get the TOP 5
78			self.stats = sorted(self.stats, key=operator.itemgetter(
79			'irq_rate'), reverse=True)[:5] # top 5 IRQ by rate/s
80
81			return self.stats
82
83			def update_views(self):
84			"""Update stats views."""
85			# Call the father's method
86			super(Plugin, self).update_views()
87
88			def msg_curse(self, args=None, max_width=None):
89			"""Return the dict to display in the curse interface."""
90			# Init the return message
91			ret = []
92
93			if not LINUX: # only available on GNU/Linux
94			return ret
95
96			# Only process if stats exist and display plugin enable...
97			if not self.stats or self.is_disable():
98			return ret
99
100			if max_width is not None and max_width >= 23:
101			irq_max_width = max_width - 14
102			else:
103			irq_max_width = 9
104
105			# Build the string message
106			# Header
107			msg = '{:{width}}'.format('IRQ', width=irq_max_width)
108			ret.append(self.curse_add_line(msg, "TITLE"))
109			msg = '{:>14}'.format('Rate/s')
110			ret.append(self.curse_add_line(msg))
111
112			for i in self.stats:
113			ret.append(self.curse_new_line())
114			msg = '{:<15}'.format(i['irq_line'])
115			ret.append(self.curse_add_line(msg))
116			msg = '{:>8}'.format(str(i['irq_rate']))
117			ret.append(self.curse_add_line(msg))
118
119			return ret
120
121
122			class GlancesIRQ(object):
123			"""
124			This class manages the IRQ file
125			"""
126
127			IRQ_FILE = '/proc/interrupts'
128
129			def __init__(self):
130			"""
131			Init the class
132			The stat are stored in a internal list of dict
133			"""
134			self.lasts = {}
135			self.reset()
136
137			def reset(self):
138			"""Reset the stats"""
139			self.stats = []
140			self.cpu_number = 0
141
142			def get(self):
143			"""Return the current IRQ stats"""
144			return self.__update()
145
146			def get_key(self):
147			"""Return the key of the dict."""
148			return 'irq_line'
149
150			def __header(self, line):
151			"""The header contain the number of CPU
152
153			CPU0 CPU1 CPU2 CPU3
154			0: 21 0 0 0 IO-APIC 2-edge timer
155			"""
156			self.cpu_number = len(line.split())
157			return self.cpu_number
158
159			def __humanname(self, line):
160			"""Get a line and
161			Return the IRQ name, alias or number (choose the best for human)
162
163			IRQ line samples:
164			1: 44487 341 44 72 IO-APIC 1-edge i8042
165			LOC: 33549868 22394684 32474570 21855077 Local timer interrupts
166			"""
167			splitted_line = line.split()
168			irq_line = splitted_line[0].replace(':', '')
169			if irq_line.isdigit():
170			# If the first column is a digit, use the alias (last column)
171			irq_line += '_{}'.format(splitted_line[-1])
172			return irq_line
173
174			def __sum(self, line):
175			"""Get a line and
176			Return the IRQ sum number
177
178			IRQ line samples:
179			1: 44487 341 44 72 IO-APIC 1-edge i8042
180			LOC: 33549868 22394684 32474570 21855077 Local timer interrupts
181			"""
182			splitted_line = line.split()
183			# sum interrupts on all CPUs
184			return sum(map(int, splitted_line[1:(self.cpu_number + 1)]))
185
186			def __update(self):
187			"""
188			Load the IRQ file and update the internal dict
189			"""
190
191			self.reset()
192
193			if not os.path.exists(self.IRQ_FILE):
194			# Correct issue #947: IRQ file do not exist on OpenVZ container
195			return self.stats
196
197			with open(self.IRQ_FILE) as irq_proc:
198			time_since_update = getTimeSinceLastUpdate('irq')
199			# Read the header
200			self.__header(irq_proc.readline())
201			# Read the rest of the lines (one line per IRQ)
202			for line in irq_proc.readlines():
203			irq_line = self.__humanname(line)
204			current_irqs = self.__sum(line)
205			irq_rate = int(
206			current_irqs -
207			self.lasts.get(irq_line) if self.lasts.get(irq_line) else 0 //
208			time_since_update)
209			irq_current = {
210			'irq_line': irq_line,
211			'irq_rate': irq_rate,
212			'key': self.get_key(),
213			'time_since_update': time_since_update
214			}
215			self.stats.append(irq_current)
216			self.lasts[irq_line] = current_irqs
217
218			return self.stats
219

nicolargo / glances

Pull Request — master (#963)

GlancesIRQ.__update() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like