GlancesIRQ.get_key() - Code Metrics - Inspection of "Generate documentation for Glances 2.8" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2b80fa...6ea077 )

by Nicolas

created 2017-01-21 13:48 UTC

GlancesIRQ.get_key() A

↳ Parent: GlancesIRQ

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	3
rs	10

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2017 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.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 = []

        # Only available on GNU/Linux
        # Only process if stats exist and display plugin enable...
        if not LINUX or not self.stats or not self.args.enable_irq:
            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'][:15])
            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

        try:
            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
        except (OSError, IOError):
            pass

        return self.stats


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

nicolargo / glances

Push — master ( 2b80fa...6ea077 )

GlancesIRQ.get_key() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like