glances.plugins.irq.GlancesIRQ.__update() - Code Metrics - Inspection of "Merge branch 'issue3171' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 5f3110...c8d70f )

by Nicolas

created 2025-05-02 13:04 UTC

glances.plugins.irq.GlancesIRQ.__update() B

↳ Parent: glances.plugins.irq

Complexity

Conditions

Size

Total Lines	32
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	23
nop	1
dl	0
loc	32
rs	8.3946
c	0
b	0
f	0

#
# This file is part of Glances.
#
# Copyright (C) 2018 Angelo Poerio <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""IRQ plugin."""

import operator
import os

from glances.globals import LINUX
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel
from glances.timer import getTimeSinceLastUpdate

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'irq_line': {
        'description': 'IRQ line name',
    },
    'irq_rate': {
        'description': 'IRQ rate per second',
        'unit': 'numberpersecond',
    },
}


class IrqPlugin(GlancesPluginModel):
    """Glances IRQ plugin.

    stats is a list
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(args=args, config=config, stats_init_value=[], fields_description=fields_description)

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

        # Init the stats
        self.irq = GlancesIRQ()

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

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the IRQ stats."""
        # Init new stats
        stats = self.get_init_value()

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

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

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

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

        # Update the stats
        self.stats = stats

        return self.stats

    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 self.is_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 7
        else:
            # No max_width defined, return an empty curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return ret

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

        for i in self.stats:
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(i['irq_line'][:name_max_width], width=name_max_width)
            ret.append(self.curse_add_line(msg))
            msg = '{:>9}'.format(str(i['irq_rate']))
            ret.append(self.curse_add_line(msg))

        return ret


class GlancesIRQ:
    """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):
        """Build 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):
        """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 += f'_{splitted_line[-1]}'
        return irq_line

    def __sum(self, line):
        """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
        FIQ:   usb_fiq
        """
        splitted_line = line.split()
        try:
            ret = sum(map(int, splitted_line[1 : (self.cpu_number + 1)]))
        except ValueError:
            # Correct issue #1007 on some conf (Raspberry Pi with Raspbian)
            ret = 0
        return ret

    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:
            pass

        return self.stats


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

nicolargo / glances

Push — develop ( 5f3110...c8d70f )

glances.plugins.irq.GlancesIRQ.__update() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like