glances.outputs.glances_curses._GlancesCurses._handle_diskio_latency() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

_GlancesCurses._handle_diskio_latency() A
last analyzed 2026-01-10 15:26 UTC

↳ Parent: glances.outputs.glances_curses
Complexity

Conditions
Size

Total Lines	5
Code Lines	4
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
cc	2
eloc	4
nop	1
dl	0
loc	5
rs	10
c	0
b	0
f	0
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Curses interface class."""

import functools
import getpass
import sys

from glances.events_list import glances_events
from glances.globals import MACOS, WINDOWS, disable, enable, nativestr, u
from glances.logger import logger
from glances.outputs.glances_colors import GlancesColors
from glances.outputs.glances_unicode import unicode_message
from glances.processes import glances_processes, sort_processes_stats_list
from glances.timer import Timer

# Import curses library for "normal" operating system
try:
    import curses
    import curses.panel
    from curses.textpad import Textbox
except ImportError:
    logger.critical("Curses module not found. Glances cannot start in standalone mode.")
    if WINDOWS:
        logger.critical("For Windows you can try installing windows-curses with pip install.")
    sys.exit(1)


class _GlancesCurses:
    """This class manages the curses display (and key pressed).

    Note: It is a private class, use GlancesCursesClient or GlancesCursesBrowser.
    """

    _hotkeys = {
        '\n': {'handler': '_handle_enter'},
        '0': {'switch': 'disable_irix'},
        '1': {'switch': 'percpu'},
        '2': {'switch': 'disable_left_sidebar'},
        '3': {'switch': 'disable_quicklook'},
        '4': {'handler': '_handle_quicklook'},
        '5': {'handler': '_handle_top_menu'},
        '6': {'switch': 'meangpu'},
        '/': {'switch': 'process_short_name'},
        'a': {'sort_key': 'auto'},
        'A': {'switch': 'disable_amps'},
        'b': {'switch': 'byte'},
        'B': {'handler': '_handle_diskio_iops'},
        'c': {'sort_key': 'cpu_percent'},
        'C': {'switch': 'disable_cloud'},
        'd': {'switch': 'disable_diskio'},
        'D': {'switch': 'disable_containers'},
        # 'e' > Enable/Disable process extended
        'E': {'handler': '_handle_erase_filter'},
        'f': {'handler': '_handle_fs_stats'},
        'F': {'switch': 'fs_free_space'},
        'g': {'switch': 'generate_graph'},
        'G': {'switch': 'disable_gpu'},
        'h': {'switch': 'help_tag'},
        'i': {'sort_key': 'io_counters'},
        'I': {'switch': 'disable_ip'},
        'j': {'switch': 'programs'},
        # 'k' > Kill selected process
        'K': {'switch': 'disable_connections'},
        'l': {'switch': 'disable_alert'},
        'L': {'handler': '_handle_diskio_latency'},
        'm': {'sort_key': 'memory_percent'},
        'M': {'switch': 'reset_minmax_tag'},
        'n': {'switch': 'disable_network'},
        'N': {'switch': 'disable_now'},
        'p': {'sort_key': 'name'},
        'P': {'switch': 'disable_ports'},
        # 'q' or ESCAPE > Quit
        'Q': {'switch': 'enable_irq'},
        'r': {'switch': 'disable_smart'},
        'R': {'switch': 'disable_raid'},
        's': {'switch': 'disable_sensors'},
        'S': {'switch': 'sparkline'},
        't': {'sort_key': 'cpu_times'},
        'T': {'switch': 'network_sum'},
        'u': {'sort_key': 'username'},
        'U': {'switch': 'network_cumul'},
        'V': {'switch': 'disable_vms'},
        'w': {'handler': '_handle_clean_logs'},
        'W': {'switch': 'disable_wifi'},
        'x': {'handler': '_handle_clean_critical_logs'},
        'z': {'handler': '_handle_disable_process'},
        '+': {'handler': '_handle_increase_nice'},
        '-': {'handler': '_handle_decrease_nice'},
        # "<" (shift + left arrow) navigation through process sort
        # ">" (shift + right arrow) navigation through process sort
        # "<" (left arrow) scroll through process name
        # ">" (right arrow) scroll through process name
        # 'UP' > Up in the server list
        # 'DOWN' > Down in the server list
    }

    _sort_loop = sort_processes_stats_list

    # Define top menu
    _top = ['quicklook', 'cpu', 'percpu', 'gpu', 'mem', 'memswap', 'load']
    _quicklook_max_width = 58

    # Define left sidebar
    # This variable is used in the make webui task in order to generate the
    # glances/outputs/static/js/uiconfig.json file for the web interface
    # This list can also be overwritten by the configuration file ([outputs] left_menu option)
    _left_sidebar = [
        'network',
        'ports',
        'wifi',
        'connections',
        'diskio',
        'fs',
        'irq',
        'folders',
        'raid',
        'smart',
        'sensors',
        'now',
    ]
    _left_sidebar_min_width = 23
    _left_sidebar_max_width = 34

    # Define right sidebar in a method because it depends of self.args.programs
    # See def _right_sidebar method

    def __init__(self, config=None, args=None):
        # Init
        self.config = config
        self.args = args

        # Init windows positions
        self.term_w = 80
        self.term_h = 24

        # Space between stats
        self.space_between_column = 3
        self.space_between_line = 2

        # Init the curses screen
        try:
            self.screen = curses.initscr()
            if not self.screen:
                logger.critical("Cannot init the curses library.\n")
                sys.exit(1)
            else:
                logger.debug(f"Curses library initialized with term: {curses.longname()}")
        except Exception as e:
            if args.export:
                logger.info("Cannot init the curses library, quiet mode on and export.")
                args.quiet = True
                return

            logger.critical(f"Cannot init the curses library ({e})")
            sys.exit(1)

        # Load configuration file
        self.load_config(config)

        # Init Curses cursor
        self._init_curses_cursor()

        # Init the colors
        self.colors_list = GlancesColors(args).get()

        # Init main window
        self.term_window = self.screen.subwin(0, 0)

        # Init edit filter tag
        self.edit_filter = False

        # Init nice increase/decrease tag
        self.increase_nice_process = False
        self.decrease_nice_process = False

        # Init kill process tag
        self.kill_process = False

        # Init the process min/max reset
        self.args.reset_minmax_tag = False

        # Init Glances cursor
        self.args.cursor_position = 0
        self.args.cursor_process_name_position = 0

        # For the moment cursor only available in standalone mode
        self.args.disable_cursor = not self.args.is_standalone

        # Catch key pressed with non blocking mode
        self.term_window.keypad(1)
        self.term_window.nodelay(1)
        self.pressedkey = -1

        # Is this the end ?
        self.is_end = False

        # History tag
        self._init_history()

    def load_config(self, config):
        """Load the outputs section of the configuration file."""
        if config is not None and config.has_section('outputs'):
            logger.debug('Read the outputs section in the configuration file')
            # Separator
            self.args.enable_separator = config.get_bool_value(
                'outputs', 'separator', default=self.args.enable_separator
            )
            # Set the left sidebar list
            self._left_sidebar = config.get_list_value('outputs', 'left_menu', default=self._left_sidebar)
            # Background color
            self.args.disable_bg = config.get_bool_value('outputs', 'disable_bg', default=self.args.disable_bg)

    def _right_sidebar(self):
        return [
            'vms',
            'containers',
            'processcount',
            'amps',
            'programlist' if self.args.programs else 'processlist',
            'alert',
        ]

    def _init_history(self):
        """Init the history option."""

        self.reset_history_tag = False

    def _init_curses_cursor(self):
        """Init cursors."""

        if hasattr(curses, 'noecho'):
            curses.noecho()
        if hasattr(curses, 'cbreak'):
            curses.cbreak()
        self.set_cursor(0)

    def set_cursor(self, value):
        """Configure the curse cursor appearance.

        0: invisible
        1: visible
        2: very visible
        """
        if hasattr(curses, 'curs_set'):
            try:
                curses.curs_set(value)
            except Exception:
                pass

    def get_key(self, window):
        return window.getch()

    def catch_actions_from_hotkey(self, hotkey):
        if self.pressedkey == ord(hotkey) and 'switch' in self._hotkeys[hotkey]:
            self._handle_switch(hotkey)
        elif self.pressedkey == ord(hotkey) and 'sort_key' in self._hotkeys[hotkey]:
            self._handle_sort_key(hotkey)
        if self.pressedkey == ord(hotkey) and 'handler' in self._hotkeys[hotkey]:
            action = getattr(self, self._hotkeys[hotkey]['handler'])
            action()

    def catch_other_actions_maybe_return_to_browser(self, return_to_browser):
        {
            self.pressedkey in {ord('e')} and not self.args.programs: self._handle_process_extended,
            self.pressedkey in {ord('k')} and not self.args.disable_cursor: self._handle_kill_process,
            self.pressedkey
            in {curses.KEY_LEFT if self.args.arrow_keys_sort else curses.KEY_SLEFT}: self._handle_sort_left,
            self.pressedkey
            in {curses.KEY_RIGHT if self.args.arrow_keys_sort else curses.KEY_SRIGHT}: self._handle_sort_right,
            self.pressedkey
            in {curses.KEY_SLEFT if self.args.arrow_keys_sort else curses.KEY_LEFT}: self._handle_process_name_left,
            self.pressedkey
            in {curses.KEY_SRIGHT if self.args.arrow_keys_sort else curses.KEY_RIGHT}: self._handle_process_name_right,
            self.pressedkey in {curses.KEY_UP, 65} and not self.args.disable_cursor: self._handle_cursor_up,
            self.pressedkey in {curses.KEY_DOWN, 66} and not self.args.disable_cursor: self._handle_cursor_down,
            self.pressedkey in {curses.KEY_F5, 18}: self._handle_refresh,
            self.pressedkey in {ord('\x1b'), ord('q')}: functools.partial(self._handle_quit, return_to_browser),
        }.get(True, lambda: None)()

    def __catch_key(self, return_to_browser=False):
        # Catch the pressed key
        self.pressedkey = self.get_key(self.term_window)
        if self.pressedkey == -1:
            return self.pressedkey

        # Actions (available in the global hotkey dict)...
        logger.debug(f"Keypressed (code: {self.pressedkey})")
        [self.catch_actions_from_hotkey(hotkey) for hotkey in self._hotkeys]

        # Other actions with key > 255 (ord will not work) and/or additional test...
        self.catch_other_actions_maybe_return_to_browser(return_to_browser)

        # Return the key code
        return self.pressedkey

    def _handle_switch(self, hotkey):
        option = '_'.join(self._hotkeys[hotkey]['switch'].split('_')[1:])
        if self._hotkeys[hotkey]['switch'].startswith('disable_'):
            if getattr(self.args, self._hotkeys[hotkey]['switch']):
                enable(self.args, option)
            else:
                disable(self.args, option)
        elif self._hotkeys[hotkey]['switch'].startswith('enable_'):
            if getattr(self.args, self._hotkeys[hotkey]['switch']):
                disable(self.args, option)
            else:
                enable(self.args, option)
        else:
            setattr(
                self.args,
                self._hotkeys[hotkey]['switch'],
                not getattr(self.args, self._hotkeys[hotkey]['switch']),
            )

    def _handle_sort_key(self, hotkey):
        glances_processes.set_sort_key(self._hotkeys[hotkey]['sort_key'], self._hotkeys[hotkey]['sort_key'] == 'auto')

    def _handle_enter(self):
        self.edit_filter = not self.edit_filter

    def _handle_quicklook(self):
        self.args.full_quicklook = not self.args.full_quicklook
        if self.args.full_quicklook:
            self.enable_fullquicklook()
        else:
            self.disable_fullquicklook()

    def _handle_top_menu(self):
        self.args.disable_top = not self.args.disable_top
        if self.args.disable_top:
            self.disable_top()
        else:
            self.enable_top()

    def _handle_process_extended(self):
        self.args.enable_process_extended = not self.args.enable_process_extended
        if not self.args.enable_process_extended:
            glances_processes.disable_extended()
        else:
            glances_processes.enable_extended()
        self.args.disable_cursor = self.args.enable_process_extended and self.args.is_standalone

    def _handle_erase_filter(self):
        glances_processes.process_filter = None

    def _handle_fs_stats(self):
        self.args.disable_fs = not self.args.disable_fs
        self.args.disable_folders = not self.args.disable_folders

    def _handle_increase_nice(self):
        self.increase_nice_process = not self.increase_nice_process

    def _handle_decrease_nice(self):
        self.decrease_nice_process = not self.decrease_nice_process

    def _handle_kill_process(self):
        self.kill_process = not self.kill_process

    def _handle_process_name_left(self):
        if self.args.cursor_process_name_position > 0:
            self.args.cursor_process_name_position -= 1

    def _handle_process_name_right(self):
        self.args.cursor_process_name_position += 1

    def _handle_clean_logs(self):
        glances_events.clean()

    def _handle_clean_critical_logs(self):
        glances_events.clean(critical=True)

    def _handle_disable_process(self):
        self.args.disable_process = not self.args.disable_process
        if self.args.disable_process:
            glances_processes.disable()
        else:
            glances_processes.enable()

    def _handle_diskio_iops(self):
        """Switch between bytes/s and IOPS for Disk IO."""
        self.args.diskio_iops = not self.args.diskio_iops
        if self.args.diskio_iops:
            self.args.diskio_latency = False

    def _handle_diskio_latency(self):
        """Switch between bytes/s and latency for Disk IO."""
        self.args.diskio_latency = not self.args.diskio_latency
        if self.args.diskio_latency:
            self.args.diskio_iops = False

    def _handle_sort_left(self):
        next_sort = (self.loop_position() - 1) % len(self._sort_loop)
        glances_processes.set_sort_key(self._sort_loop[next_sort], False)

    def _handle_sort_right(self):
        next_sort = (self.loop_position() + 1) % len(self._sort_loop)
        glances_processes.set_sort_key(self._sort_loop[next_sort], False)

    def _handle_cursor_up(self):
        if self.args.cursor_position > 0:
            self.args.cursor_position -= 1

    def _handle_cursor_down(self):
        if self.args.cursor_position < glances_processes.processes_count:
            self.args.cursor_position += 1

    def _handle_quit(self, return_to_browser):
        if return_to_browser:
            logger.info("Stop Glances client and return to the browser")
        else:
            logger.info(f"Stop Glances (keypressed: {self.pressedkey})")
            # End the curses window
            self.end()
            # Exit the program
            sys.exit(0)

    def _handle_refresh(self):
        glances_processes.reset_internal_cache()

    def loop_position(self):
        """Return the current sort in the loop"""
        for i, v in enumerate(self._sort_loop):
            if v == glances_processes.sort_key:
                return i
        return 0

    def disable_top(self):
        """Disable the top panel"""
        for p in ['quicklook', 'cpu', 'gpu', 'mem', 'memswap', 'load']:
            setattr(self.args, 'disable_' + p, True)

    def enable_top(self):
        """Enable the top panel"""
        for p in ['quicklook', 'cpu', 'gpu', 'mem', 'memswap', 'load']:
            setattr(self.args, 'disable_' + p, False)

    def disable_fullquicklook(self):
        """Disable the full quicklook mode"""
        for p in ['quicklook', 'cpu', 'gpu', 'mem', 'memswap']:
            setattr(self.args, 'disable_' + p, False)

    def enable_fullquicklook(self):
        """Disable the full quicklook mode"""
        self.args.disable_quicklook = False
        for p in ['cpu', 'gpu', 'mem', 'memswap']:
            setattr(self.args, 'disable_' + p, True)

    def end(self):
        """Shutdown the curses window."""
        if hasattr(curses, 'echo'):
            curses.echo()
        if hasattr(curses, 'nocbreak'):
            curses.nocbreak()
        try:
            curses.curs_set(1)
        except Exception:
            pass
        try:
            curses.endwin()
        except Exception:
            pass
        self.is_end = True

    def init_line_column(self):
        """Init the line and column position for the curses interface."""
        self.init_line()
        self.init_column()

    def init_line(self):
        """Init the line position for the curses interface."""
        self.line = 0
        self.next_line = 0

    def init_column(self):
        """Init the column position for the curses interface."""
        self.column = 0
        self.next_column = 0

    def new_line(self, separator=False):
        """New line in the curses interface."""
        self.line = self.next_line

    def new_column(self):
        """New column in the curses interface."""
        self.column = self.next_column

    def separator_line(self, color='SEPARATOR'):
        """Add a separator line in the curses interface."""
        if not self.args.enable_separator:
            return
        self.new_line()
        self.line -= 1
        line_width = self.term_window.getmaxyx()[1] - self.column
        if self.line >= 0 and self.line < self.term_window.getmaxyx()[0]:
            position = [self.line, self.column]
            line_color = self.colors_list[color]
            line_type = curses.ACS_HLINE if not self.args.disable_unicode else unicode_message('MEDIUM_LINE', self.args)
            self.term_window.hline(
                *position,
                line_type,
                line_width,
                line_color,
            )

    def __get_stat_display(self, stats, layer):
        """Return a dict of dict with all the stats display.
        # TODO: Drop extra parameter

        :param stats: Global stats dict
        :param layer: ~ cs_status
            "None": standalone or server mode
            "Connected": Client is connected to a Glances server
            "SNMP": Client is connected to a SNMP server
            "Disconnected": Client is disconnected from the server

        :returns: dict of dict
            * key: plugin name
            * value: dict returned by the get_stats_display Plugin method
        """
        ret = {}

        for p in stats.getPluginsList(enable=False):
            # Ignore Quicklook because it is compute later in __display_top
            if p == 'quicklook':
                continue

            # Compute the plugin max size for the left sidebar
            plugin_max_width = None
            if p in self._left_sidebar:
                plugin_max_width = min(
                    self._left_sidebar_max_width,
                    max(self._left_sidebar_min_width, self.term_window.getmaxyx()[1] - 105),
                )

            # Get the view
            ret[p] = stats.get_plugin(p).get_stats_display(args=self.args, max_width=plugin_max_width)

        return ret

    def display(self, stats, cs_status=None):
        """Display stats on the screen.

        :param stats: Stats database to display
        :param cs_status:
            "None": standalone or server mode
            "Connected": Client is connected to a Glances server
            "SNMP": Client is connected to a SNMP server
            "Disconnected": Client is disconnected from the server

        :return: True if the stats have been displayed else False if the help have been displayed
        """
        # Init the internal line/column for Glances Curses
        self.init_line_column()

        # Update the stats messages
        ###########################

        # Get all the plugins view
        self.args.cs_status = cs_status
        __stat_display = self.__get_stat_display(stats, layer=cs_status)

        # Display the stats on the curses interface
        ###########################################

        # Help screen (on top of the other stats)
        if self.args.help_tag:
            # Display the stats...
            self.display_plugin(stats.get_plugin('help').get_stats_display(args=self.args))
            # ... and exit
            return False

        # =======================================
        # Display first line (system+ip+uptime)
        # Optionally: Cloud is on the second line
        # =======================================
        self.__display_header(__stat_display)
        self.separator_line()

        # ==============================================================
        # Display second line (<SUMMARY>+CPU|PERCPU+<GPU>+LOAD+MEM+SWAP)
        # ==============================================================
        self.__display_top(__stat_display, stats)
        self.init_column()
        self.separator_line()

        # ==================================================================
        # Display left sidebar (NETWORK+PORTS+DISKIO+FS+SENSORS+Current time)
        # ==================================================================
        self.__display_left(__stat_display)

        # ====================================
        # Display right stats (process and co)
        # ====================================
        self.__display_right(__stat_display)

        # =====================
        # Others popup messages
        # =====================

        # Display edit filter popup
        # Only in standalone mode (cs_status is None)
        if self.edit_filter and cs_status is None:
            new_filter = self.display_popup(
                'Process filter pattern: \n\n'
                + 'Examples:\n'
                + '- .*python.*\n'
                + '- /usr/lib.*\n'
                + '- name:.*nautilus.*\n'
                + '- cmdline:.*glances.*\n'
                + '- username:nicolargo\n'
                + '- username:^root        ',
                popup_type='input',
                input_value=glances_processes.process_filter_input,
            )
            glances_processes.process_filter = new_filter
        elif self.edit_filter and cs_status is not None:
            self.display_popup('Process filter only available in standalone mode')
        self.edit_filter = False

        # Manage increase/decrease nice level of the selected process
        # Only in standalone mode (cs_status is None)
        if self.increase_nice_process and cs_status is None:
            self.nice_increase(stats.get_plugin('processlist').get_raw()[self.args.cursor_position])
        self.increase_nice_process = False
        if self.decrease_nice_process and cs_status is None:
            self.nice_decrease(stats.get_plugin('processlist').get_raw()[self.args.cursor_position])
        self.decrease_nice_process = False

        # Display kill process confirmation popup
        # Only in standalone mode (cs_status is None)
        if self.kill_process and cs_status is None:
            self.kill(stats.get_plugin('processlist').get_raw()[self.args.cursor_position])
        elif self.kill_process and cs_status is not None:
            self.display_popup('Kill process only available for local processes')
        self.kill_process = False

        # Display graph generation popup
        if self.args.generate_graph:
            if 'graph' in stats.getExportsList():
                self.display_popup(f'Generate graph in {self.args.export_graph_path}')
            else:
                logger.warning('Graph export module is disable. Run Glances with --export graph to enable it.')
                self.args.generate_graph = False

        return True

    def nice_increase(self, process):
        glances_processes.nice_increase(process['pid'])

    def nice_decrease(self, process):
        glances_processes.nice_decrease(process['pid'])

    def kill(self, process):
        """Kill a process, or a list of process if the process has a childrens field.

        :param process
        :return: None
        """
        logger.debug(f"Selected process to kill: {process}")

        if 'childrens' in process:
            pid_to_kill = process['childrens']
        else:
            pid_to_kill = [process['pid']]

        confirm = self.display_popup(
            'Kill process: {} (pid: {}) ?\n\nConfirm ([y]es/[n]o): '.format(
                process['name'],
                ', '.join(map(str, pid_to_kill)),
            ),
            popup_type='yesno',
        )

        if confirm.lower().startswith('y'):
            for pid in pid_to_kill:
                try:
                    ret_kill = glances_processes.kill(pid)
                except Exception as e:
                    logger.error(f'Can not kill process {pid} ({e})')
                else:
                    logger.info(f'Kill signal has been sent to process {pid} (return code: {ret_kill})')

    def __display_header(self, stat_display):
        """Display the firsts lines (header) in the Curses interface.

        system + ip + uptime
        (cloud)
        """
        # First line
        self.new_line()
        self.space_between_column = 0
        l_uptime = 1
        for i in ['system', 'ip', 'uptime']:
            if i in stat_display:
                l_uptime += self.get_stats_display_width(stat_display[i])
        self.display_plugin(stat_display["system"], display_optional=(self.term_window.getmaxyx()[1] >= l_uptime))
        self.space_between_column = 3
        if 'ip' in stat_display:
            self.new_column()
            self.display_plugin(stat_display["ip"], display_optional=(self.term_window.getmaxyx()[1] >= 100))
        self.new_column()
        cloud_width = self.get_stats_display_width(stat_display.get("cloud", 0))
        self.display_plugin(stat_display["uptime"], add_space=-(cloud_width != 0))
        self.init_column()
        if cloud_width != 0:
            # Second line (optional)
            self.new_line()
            self.display_plugin(stat_display["cloud"])

    def __display_top(self, stat_display, stats):
        """Display the second line in the Curses interface.

        <QUICKLOOK> + CPU|PERCPU + <GPU> + MEM + SWAP + LOAD
        """
        self.init_column()
        self.new_line()

        # Init quicklook
        stat_display['quicklook'] = {'msgdict': []}

        # Dict for plugins width
        plugin_widths = {}
        for p in self._top:
            plugin_widths[p] = (
                self.get_stats_display_width(stat_display.get(p, 0)) if hasattr(self.args, 'disable_' + p) else 0
            )

        # Width of all plugins
        stats_width = sum(plugin_widths.values())

        # Number of plugin but quicklook
        stats_number = sum(
            [int(stat_display[p]['msgdict'] != []) for p in self._top if not getattr(self.args, 'disable_' + p)]
        )

        if not self.args.disable_quicklook:
            # Quick look is in the place !
            if self.args.full_quicklook:
                quicklook_width = self.term_window.getmaxyx()[1] - (
                    stats_width + 8 + stats_number * self.space_between_column
                )
            else:
                quicklook_width = min(
                    self.term_window.getmaxyx()[1] - (stats_width + 8 + stats_number * self.space_between_column),
                    self._quicklook_max_width - 5,
                )
            try:
                stat_display["quicklook"] = stats.get_plugin('quicklook').get_stats_display(
                    max_width=quicklook_width, args=self.args
                )
            except AttributeError as e:
                logger.debug(f"Quicklook plugin not available ({e})")
            else:
                plugin_widths['quicklook'] = self.get_stats_display_width(stat_display["quicklook"])
                stats_width = sum(plugin_widths.values()) + 1
            self.space_between_column = 1
            self.display_plugin(stat_display["quicklook"])
            self.new_column()

        # Compute spaces between plugins
        # Note: Only one space between Quicklook and others
        plugin_display_optional = {}
        for p in self._top:
            plugin_display_optional[p] = True
        if stats_number > 1:
            self.space_between_column = max(1, int((self.term_window.getmaxyx()[1] - stats_width) / (stats_number - 1)))
            for p in ['mem', 'cpu']:
                # No space ? Remove optional stats
                if self.space_between_column < 3:
                    plugin_display_optional[p] = False
                    plugin_widths[p] = (
                        self.get_stats_display_width(stat_display[p], without_option=True)
                        if hasattr(self.args, 'disable_' + p)
                        else 0
                    )
                    stats_width = sum(plugin_widths.values()) + 1
                    self.space_between_column = max(
                        1, int((self.term_window.getmaxyx()[1] - stats_width) / (stats_number - 1))
                    )
        else:
            self.space_between_column = 0

        # Display CPU, MEM, SWAP and LOAD
        for p in self._top:
            if p == 'quicklook':
                continue
            if p in stat_display:
                self.display_plugin(stat_display[p], display_optional=plugin_display_optional[p])
            if p != 'load':
                # Skip last column
                self.new_column()

        # Space between column
        self.space_between_column = 3

        # Backup line position
        self.saved_line = self.next_line

    def __display_left(self, stat_display):
        """Display the left sidebar in the Curses interface."""
        self.init_column()

        if self.args.disable_left_sidebar:
            return

        for p in self._left_sidebar:
            if (hasattr(self.args, 'enable_' + p) or hasattr(self.args, 'disable_' + p)) and p in stat_display:
                self.new_line()
                if p == 'sensors':
                    self.display_plugin(
                        stat_display['sensors'],
                        max_y=(self.term_window.getmaxyx()[0] - self.get_stats_display_height(stat_display['now']) - 2),
                    )
                else:
                    self.display_plugin(stat_display[p])

    def __display_right(self, stat_display):
        """Display the right sidebar in the Curses interface.

        docker + processcount + amps + processlist + alert
        """
        # Do not display anything if space is not available...
        if self.term_window.getmaxyx()[1] < self._left_sidebar_min_width:
            return

        # Restore line position
        self.next_line = self.saved_line

        # Display right sidebar
        self.new_column()
        for p in self._right_sidebar():
            if (hasattr(self.args, 'enable_' + p) or hasattr(self.args, 'disable_' + p)) and p in stat_display:
                self.new_line()
                if p in ['processlist', 'programlist']:
                    p_index = self._right_sidebar().index(p) + 1
                    self.display_plugin(
                        stat_display[p],
                        display_optional=(self.term_window.getmaxyx()[1] > 102),
                        display_additional=(not MACOS),
                        max_y=(
                            self.term_window.getmaxyx()[0]
                            - sum(
                                [
                                    self.get_stats_display_height(stat_display[i])
                                    for i in self._right_sidebar()[p_index:]
                                ]
                            )
                            - 2
                        ),
                    )
                else:
                    self.display_plugin(stat_display[p])

    def display_popup(
        self,
        message,
        size_x=None,
        size_y=None,
        duration=3,
        popup_type='info',
        input_size=30,
        input_value=None,
        is_password=False,
    ):
        """
        Display a centered popup.

        popup_type: ='info'
         Just an information popup, no user interaction
         Display a centered popup with the given message during duration seconds
         If size_x and size_y: set the popup size
         else set it automatically
         Return True if the popup could be displayed

        popup_type='input'
         Display a centered popup with the given message and a input field
         If size_x and size_y: set the popup size
         else set it automatically
         Return the input string or None if the field is empty

        popup_type='yesno'
         Display a centered popup with the given message
         If size_x and size_y: set the popup size
         else set it automatically
         Return True (yes) or False (no)
        """
        # Center the popup
        sentence_list = message.split('\n')
        if size_x is None:
            size_x = len(max(sentence_list, key=len)) + 4
            # Add space for the input field
            if popup_type == 'input':
                size_x += input_size
        if size_y is None:
            size_y = len(sentence_list) + 4
        screen_x = self.term_window.getmaxyx()[1]
        screen_y = self.term_window.getmaxyx()[0]
        if size_x > screen_x or size_y > screen_y:
            # No size to display the popup => abord
            return False
        pos_x = int((screen_x - size_x) / 2)
        pos_y = int((screen_y - size_y) / 2)

        # Create the popup
        popup = curses.newwin(size_y, size_x, pos_y, pos_x)

        # Fill the popup
        popup.border()

        # Add the message
        for y, m in enumerate(sentence_list):
            if m:
                popup.addnstr(2 + y, 2, m, len(m))

        if popup_type == 'info':
            # Display the popup
            popup.refresh()
            self.wait(duration * 1000)
            return True

        if popup_type == 'input':
            logger.info(popup_type)
            logger.info(is_password)
            # Create a sub-window for the text field
            sub_pop = popup.derwin(1, input_size, 2, 2 + len(m))

            sub_pop.attron(self.colors_list['FILTER'])
            # Init the field with the current value
            if input_value is not None:
                sub_pop.addnstr(0, 0, input_value, len(input_value))
            # Display the popup
            popup.refresh()
            sub_pop.refresh()
            # Create the textbox inside the sub-windows
            self.set_cursor(2)
            self.term_window.keypad(1)
            if is_password:
                textbox = getpass.getpass('')
                self.set_cursor(0)
                if textbox != '':
                    return textbox
                return None

            # No password
            textbox = GlancesTextbox(sub_pop, insert_mode=True)
            textbox.edit()
            self.set_cursor(0)
            if textbox.gather() != '':
                return textbox.gather()[:-1]
            return None

        if popup_type == 'yesno':
            # Create a sub-window for the text field
            sub_pop = popup.derwin(1, 2, len(sentence_list) + 1, len(m) + 2)
            sub_pop.attron(self.colors_list['FILTER'])
            # Init the field with the current value
            try:
                sub_pop.addnstr(0, 0, '', 0)
            except curses.error:
                pass
            # Display the popup
            popup.refresh()
            sub_pop.refresh()
            # Create the textbox inside the sub-windows
            self.set_cursor(2)
            self.term_window.keypad(1)
            textbox = GlancesTextboxYesNo(sub_pop, insert_mode=False)
            textbox.edit()
            self.set_cursor(0)
            # self.term_window.keypad(0)
            return textbox.gather()

        return None

    def setup_upper_left_pos(self, plugin_stats):
        screen_y, screen_x = self.term_window.getmaxyx()

        if plugin_stats['align'] == 'right':
            # Right align (last column)
            display_x = screen_x - self.get_stats_display_width(plugin_stats)
        else:
            display_x = self.column

        if plugin_stats['align'] == 'bottom':
            # Bottom (last line)
            display_y = screen_y - self.get_stats_display_height(plugin_stats)
        else:
            display_y = self.line

        return display_y, display_x

    def get_next_x_and_x_max(self, m, x, x_max):
        # New column
        # Python 2: we need to decode to get real screen size because
        # UTF-8 special tree chars occupy several bytes.
        # Python 3: strings are strings and bytes are bytes, all is
        # good.
        try:
            x += len(u(m['msg']))
        except UnicodeDecodeError:
            # Quick and dirty hack for issue #745
            pass
        if x > x_max:
            x_max = x

        return x, x_max

    def display_stats_with_current_size(self, m, y, x):
        screen_x = self.term_window.getmaxyx()[1]
        self.term_window.addnstr(
            y,
            x,
            m['msg'],
            # Do not display outside the screen
            screen_x - x,
            self.colors_list[m['decoration']],
        )

    def display_stats(self, plugin_stats, init, helper):
        y, x, x_max = init
        for m in plugin_stats['msgdict']:
            # New line
            try:
                if m['msg'].startswith('\n'):
                    y, x = helper['goto next, add first col'](y, x)
                    continue
            except Exception:
                # Avoid exception (see issue #1692)
                pass
            # Do not display outside the screen
            if x < 0:
                continue
            if helper['x overbound?'](m, x):
                continue
            if helper['y overbound?'](y):
                break
            # If display_optional = False do not display optional stats
            if helper['display optional?'](m):
                continue
            # If display_additional = False do not display additional stats
            if helper['display additional?'](m):
                continue
            # Is it possible to display the stat with the current screen size
            # !!! Crash if not try/except... Why ???
            try:
                self.display_stats_with_current_size(m, y, x)
            except Exception:
                pass
            else:
                x, x_max = self.get_next_x_and_x_max(m, x, x_max)

        return y, x, x_max

    def display_plugin(self, plugin_stats, display_optional=True, display_additional=True, max_y=65535, add_space=0):
        """Display the plugin_stats on the screen.

        :param plugin_stats:
        :param display_optional: display the optional stats if True
        :param display_additional: display additional stats if True
        :param max_y: do not display line > max_y
        :param add_space: add x space (line) after the plugin
        """
        # Exit if:
        # - the plugin_stats message is empty
        # - the display tag = False
        if plugin_stats is None or not plugin_stats['msgdict'] or not plugin_stats['display']:
            # Exit
            return 0

        # Get the screen size
        screen_y, screen_x = self.term_window.getmaxyx()

        # Set the upper/left position of the message
        display_y, display_x = self.setup_upper_left_pos(plugin_stats)

        helper = {
            'goto next, add first col': lambda y, x: (y + 1, display_x),

            'x overbound?': lambda m, x: not m['splittable'] and (x + len(m['msg']) > screen_x),

            'y overbound?': lambda y: y < 0 or (y + 1 > screen_y) or (y > max_y),

            'display optional?': lambda m: not display_optional and m['optional'],
            'display additional?': lambda m: not display_additional and m['additional'],
        }

        # Display
        init = display_y, display_x, display_x
        y, x, x_max = self.display_stats(plugin_stats, init, helper)

        # Compute the next Glances column/line position
        self.next_column = max(self.next_column, x_max + self.space_between_column)
        self.next_line = max(self.next_line, y + self.space_between_line)

        # Have empty lines after the plugins
        self.next_line += add_space
        return None

    def clear(self):
        """Erase the content of the screen.
        The difference is that clear() also calls clearok(). clearok()
        basically tells ncurses to forget whatever it knows about the current
        terminal contents, so that when refresh() is called, it will actually
        begin by clearing the entire terminal screen before redrawing any of it."""
        self.term_window.clear()

    def erase(self):
        """Erase the content of the screen.
        erase() on the other hand, just clears the screen (the internal
        object, not the terminal screen). When refresh() is later called,
        ncurses will still compute the minimum number of characters to send to
        update the terminal."""
        self.term_window.erase()

    def refresh(self):
        """Refresh the windows"""
        self.term_window.refresh()

    def flush(self, stats, cs_status=None):
        """Erase and update the screen.

        :param stats: Stats database to display
        :param cs_status:
            "None": standalone or server mode
            "Connected": Client is connected to the server
            "Disconnected": Client is disconnected from the server
        """
        # See https://stackoverflow.com/a/43486979/1919431
        self.erase()
        self.display(stats, cs_status=cs_status)
        self.refresh()

    def update(self, stats, duration=3, cs_status=None, return_to_browser=False):
        """Update the screen.

        :param stats: Stats database to display
        :param duration: duration of the loop
        :param cs_status:
            "None": standalone or server mode
            "Connected": Client is connected to the server
            "Disconnected": Client is disconnected from the server
        :param return_to_browser:
            True: Do not exist, return to the browser list
            False: Exit and return to the shell

        :return: True if exit key has been pressed else False
        """
        # Flush display
        self.flush(stats, cs_status=cs_status)

        # If the duration is < 0 (update + export time > refresh_time)
        # Then display the interface and log a message
        if duration <= 0:
            logger.warning('Update and export time higher than refresh_time.')
            duration = 0.1

        # Wait duration (in s) time
        isexitkey = False
        countdown = Timer(duration)
        # Set the default timeout (in ms) between two getch
        self.term_window.timeout(100)
        while not countdown.finished() and not isexitkey:
            # Getkey
            pressedkey = self.__catch_key(return_to_browser=return_to_browser)

            if pressedkey == -1:
                self.wait()
                continue

            isexitkey = pressedkey == ord('\x1b') or pressedkey == ord('q')

            if pressedkey == curses.KEY_F5 or self.pressedkey == 18:
                # Were asked to refresh (F5 or Ctrl-R)
                self.clear()
                return isexitkey

            if pressedkey in (curses.KEY_UP, 65, curses.KEY_DOWN, 66, curses.KEY_LEFT, 68, curses.KEY_RIGHT, 67):
                # Up of won key pressed, reset the countdown
                # Better for user experience
                countdown.reset()

            if isexitkey and self.args.help_tag:
                # Quit from help should return to main screen, not exit #1874
                self.args.help_tag = not self.args.help_tag
                return False

            if not isexitkey and pressedkey > -1:
                # Redraw display
                self.flush(stats, cs_status=cs_status)
                # Overwrite the timeout with the countdown
                self.wait(delay=int(countdown.get() * 1000))

        return isexitkey

    def wait(self, delay=100):
        """Wait delay in ms"""
        curses.napms(delay)

    def get_stats_display_width(self, curse_msg, without_option=False):
        """Return the width of the formatted curses message."""
        try:
            if without_option:
                # Size without options
                c = len(
                    max(
                        ''.join(
                            [
                                (u(u(nativestr(i['msg'])).encode('ascii', 'replace')) if not i['optional'] else "")
                                for i in curse_msg['msgdict']
                            ]
                        ).split('\n'),
                        key=len,
                    )
                )
            else:
                # Size with all options
                c = len(
                    max(
                        ''.join(
                            [u(u(nativestr(i['msg'])).encode('ascii', 'replace')) for i in curse_msg['msgdict']]
                        ).split('\n'),
                        key=len,
                    )
                )
        except Exception as e:
            logger.debug(f'ERROR: Can not compute plugin width ({e})')
            return 0
        else:
            return c

    def get_stats_display_height(self, curse_msg):
        """Return the height of the formatted curses message.

        The height is defined by the number of '\n' (new line).
        """
        try:
            c = [i['msg'] for i in curse_msg['msgdict']].count('\n')
        except Exception as e:
            logger.debug(f'ERROR: Can not compute plugin height ({e})')
            return 0
        else:
            return c + 1


class GlancesCursesStandalone(_GlancesCurses):
    """Class for the Glances curse standalone."""

    # Default number of processes to displayed is set to 50
    glances_processes.max_processes = 50


class GlancesCursesClient(_GlancesCurses):
    """Class for the Glances curse client."""

    # Default number of processes to displayed is set to 50
    # For the moment, cursor in client/server mode is not supported see #3221
    glances_processes.max_processes = 50


class GlancesTextbox(Textbox):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def do_command(self, ch):
        if ch == 10:  # Enter
            return 0
        if ch == 127:  # Back
            return 8
        return super().do_command(ch)


class GlancesTextboxYesNo(Textbox):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def do_command(self, ch):
        return super().do_command(ch)

nicolargo / glances

_GlancesCurses._handle_diskio_latency() A last analyzed 2026-01-10 15:26 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

_GlancesCurses._handle_diskio_latency() A
last analyzed 2026-01-10 15:26 UTC
