ups-daemon

#!/usr/bin/env python3
""" ups-daemon  -  Utility for interacting with compatible UPSs with network
                   management cards.

    With no options specified, the utility will give the current status of the
    UPS configured with *daemon = true* in the ups-config.json file. With the
    *--daemon* option, *ups-daemon* will continuously check the status of the
    UPS.  When it detects that the UPS is sourcing powering from the battery,
    it will check the amount of time it has been running on battery and run
    the specified suspend script when the specified threshold is exceeded.  It
    will execute the specified resume script when it detects power has resumed.
    When the utility detects a Battery Low event from the UPS or that time
    remaining for battery or the battery charge is below specified thresholds,
    then the shutdown script will be executed. If *ups-daemon* detects a return
    to line power has occurred before the shutdown has completed, it will
    execute the cancel shutdown script.  With the *--verbose* option set,
    event update messages will be output, otherwise, only events are output.
    The *--no_markup* option will cause the output to be in plain text, with
    no color markup codes. The *--logfile filename* option is used to specify
    a logfile, but is not implemented at this time.  The threshold and script
    definitions must be made in the *ups-utils.ini* file using
    *ups-utils.ini.template* as a template.  The logger is enabled with the
    *--debug* option.  The *--ltz* option will result in the use of the local
    time zone in the monitor window and logs.  This will be the local time of
    where the app is running, not the location of the UPS.  The default is UTC.

    Copyright (C) 2019  RicksLab

    This program is free software: you can redistribute it and/or modify it
    under the terms of the GNU General Public License as published by the Free
    Software Foundation, either version 3 of the License, or (at your option)
    any later version.

    This program 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 General Public License for
    more details.

    You should have received a copy of the GNU General Public License along with
    this program.  If not, see <https://www.gnu.org/licenses/>.
"""
__author__ = "RicksLab"
__copyright__ = "Copyright (C) 2019 RicksLab"
__license__ = "GNU General Public License"
__program_name__ = "ups-daemon"
__maintainer__ = "RicksLab"
__docformat__ = 'reStructuredText'
# pylint: disable=multiple-statements
# pylint: disable=line-too-long
# pylint: disable=consider-using-f-string

import argparse
import sys
import os
import inspect
from time import sleep
import signal
import logging
from typing import Any, Dict
from UPSmodules import UPSmodule as UPS
from UPSmodules.env import UT_CONST
from UPSmodules import __version__, __status__, __credits__
from UPSmodules.UPSKeys import MibGroup, MarkUpCodes, MiB


LOGGER = logging.getLogger('ups-utils')


def ctrl_c_handler(target_signal: Any, _frame: Any) -> None:
    """ Signal catcher for ctrl-c to exit monitor loop.

    :param target_signal: Target signal name
    :param _frame: Ignored
    """
    LOGGER.debug('ctrl_c_handler (ID: %s) has been caught. Setting quit flag...', target_signal)
    print('Setting quit flag...')
    UT_CONST.quit = True


def ctrl_u_handler(target_signal: Any, _frame: Any) -> None:
    """ Signal catcher for ctrl-c to exit monitor loop.

    :param target_signal: Target signal name
    :param _frame: Ignored
    """
    LOGGER.debug('ctrl_u_handler (ID: %s) has been caught. Setting refresh daemon flag...', target_signal)
    print('Setting refresh daemon flag...')
    UT_CONST.refresh_daemon = True


def daemon_sleep(sleep_time: int = 30) -> None:
    """ Interruptable sleep routine for daemon loop.

    :param sleep_time:  Sleep time in seconds
    :return: None
    """
    for _sleep_index in range(0, sleep_time):
        sleep(1)
        if UT_CONST.quit:
            print('[{}]: Received Quit Signal'.format(UT_CONST.now(ltz=UT_CONST.use_ltz, as_string=True)))
            sys.exit(0)


def main() -> None:
    """ Main function.
    """
    parser = argparse.ArgumentParser()
    parser.add_argument('--about', help='README', action='store_true', default=False)

    parser.add_argument('--daemon', help='Run in daemon mode', action='store_true', default=False)

    # Verbosity, logging, and debug options
    parser.add_argument('--ltz', help='Use local time zone instead of UTC', action='store_true', default=False)
    parser.add_argument('--no_markup', help='Output plane text',
                        action='store_true', default=False)
    parser.add_argument('--logfile', help='Specify logfile', type=str, default='')
    parser.add_argument('--verbose', help='Output execution exception notices', action='store_true', default=False)
    parser.add_argument('-d', '--debug', help='Debug output', action='store_true', default=False)
    args = parser.parse_args()

    signal.signal(signal.SIGINT, ctrl_c_handler)

    # About me
    if args.about:
        print(__doc__)
        print("Author: ", __author__)
        print("Copyright: ", __copyright__)
        print('Credits: ', *['\n      {}'.format(item) for item in __credits__])
        print("License: ", __license__)
        print("Version: ", __version__)
        print('Install Type: ', UT_CONST.install_type)
        print("Maintainer: ", __maintainer__)
        print("Status: ", __status__)
        sys.exit(0)

    UT_CONST.set_env_args(args, __program_name__)
    UT_CONST.cmd_path = os.path.dirname(inspect.getfile(inspect.currentframe()))
    LOGGER.debug('########## %s %s [%s]', __program_name__, __version__, UT_CONST.cmd_path)
    if args.verbose:
        print('Install type: {}'.format(UT_CONST.install_type))

    reset_code: str = UT_CONST.mark_up_codes[MarkUpCodes.reset]
    color_code: str = '{}{}'.format(UT_CONST.mark_up_codes[MarkUpCodes.red],
                                    UT_CONST.mark_up_codes[MarkUpCodes.bold])
    if not UT_CONST.check_env():
        UT_CONST.process_message('Error: {}Invalid environment. Exiting...{}'.format(
            color_code, reset_code), verbose=True)
        sys.exit(-1)

    ups_list = UPS.UpsList()
    if args.verbose:
        print('{}\n'.format(ups_list))

    daemon_ups = ups_list.get_daemon_ups()
    num_ups = ups_list.num_upss()
    if not daemon_ups:
        UT_CONST.process_message('Error: {}No Daemon UPS defined{}'.format(color_code, reset_code), verbose=True)
        UT_CONST.fatal = True
    elif num_ups['daemon'] > 1:
        UT_CONST.process_message('Error: {}Multiple Daemon UPS defined{}'.format(color_code, reset_code), verbose=True)
        UT_CONST.fatal = True
    if not daemon_ups.is_responsive():
        UT_CONST.process_message('Error: {}Daemon UPS [{}] is unresponsive.{}'.format(
            color_code, daemon_ups['display_name'], reset_code), verbose=True)
        UT_CONST.fatal = True

    if UT_CONST.fatal:
        UT_CONST.process_message('Fatal: {}Exiting...{}'.format(color_code, reset_code), verbose=True)
        UT_CONST.process_message('    Make sure configuration files specify a daemon UPS.\n'
                                 '    Execute: \'ups-ls --about\' for more information of ups-utils\n'
                                 '              configuration file location and status.', verbose=True)
        sys.exit(-1)

    # Display current status of target UPS
    daemon_ups.read_ups_list_items(MibGroup.all, display=False)
    daemon_ups.print()
    ups_list.daemon.print_daemon_parameters()

    if args.daemon:
        signal.signal(signal.SIGUSR1, ctrl_u_handler)
        # Daemon flags
        overload_fault = False
        suspend_state = False
        shutting_down = False

        # Control thresholds
        crit_bat_level = daemon_ups.daemon.daemon_params['threshold_battery_capacity']['crit']
        warn_bat_level = daemon_ups.daemon.daemon_params['threshold_battery_capacity']['warn']
        crit_load_level = daemon_ups.daemon.daemon_params['threshold_battery_load']['crit']
        warn_load_level = daemon_ups.daemon.daemon_params['threshold_battery_load']['warn']
        crit_runtime_rem = daemon_ups.daemon.daemon_params['threshold_battery_time_rem']['crit']
        warn_runtime_rem = daemon_ups.daemon.daemon_params['threshold_battery_time_rem']['warn']
        normal_sleep = daemon_ups.daemon.daemon_params['read_interval']['daemon']
        fault_sleep = daemon_ups.daemon.daemon_param_defaults['read_interval']['limit']

        active_sleep = normal_sleep
        if UT_CONST.no_markup:
            norm_style = warn_style = crit_style = ok_style = err_style = reset_style = ''
        else:
            norm_style = UT_CONST.mark_up_codes[MarkUpCodes.cyan]
            warn_style = UT_CONST.mark_up_codes[MarkUpCodes.warn]
            crit_style = UT_CONST.mark_up_codes[MarkUpCodes.crit]
            ok_style = UT_CONST.mark_up_codes[MarkUpCodes.ok]
            err_style = UT_CONST.mark_up_codes[MarkUpCodes.error]
            reset_style = UT_CONST.mark_up_codes[MarkUpCodes.reset]
        ups_states: Dict[str, str] = {
            'good':     '{} READY:   {}'.format(ok_style, reset_style),
            'ready':    '{} READY:   {}'.format(ok_style, reset_style),
            'fault':    '{} FAULT:   {}'.format(warn_style, reset_style),
            'warning':  '{} WARNING: {}'.format(warn_style, reset_style),
            'error':    '{} ERROR:   {}'.format(err_style, reset_style),
            'critical': '{} CRITICAL:{}'.format(crit_style, reset_style),
        }
        ready_status = True
        while True:
            time_str = UT_CONST.now(ltz=UT_CONST.use_ltz, as_string=True)

            # Check status of UPS
            bat_status = daemon_ups.send_snmp_command(MiB.battery_status, display=False)
            if not bat_status:
                print('[{}] {} UPS [{}] is unresponsive'.format(
                    time_str, ups_states['error'], daemon_ups['display_name']))
                daemon_sleep(active_sleep)
                continue

            # Check for request to refresh Daemon parameters
            if UT_CONST.refresh_daemon:
                ups_list.read_set_daemon()
                # Control thresholds
                crit_bat_level = daemon_ups.daemon.daemon_params['threshold_battery_capacity']['crit']
                warn_bat_level = daemon_ups.daemon.daemon_params['threshold_battery_capacity']['warn']
                crit_load_level = daemon_ups.daemon.daemon_params['threshold_battery_load']['crit']
                warn_load_level = daemon_ups.daemon.daemon_params['threshold_battery_load']['warn']
                crit_runtime_rem = daemon_ups.daemon.daemon_params['threshold_battery_time_rem']['crit']
                warn_runtime_rem = daemon_ups.daemon.daemon_params['threshold_battery_time_rem']['warn']
                normal_sleep = daemon_ups.daemon.daemon_params['read_interval']['daemon']
                fault_sleep = daemon_ups.daemon.daemon_param_defaults['read_interval']['limit']

            # Read data from UPS
            out_power = int(daemon_ups.send_snmp_command(MiB.output_power, display=False))
            bat_load = int(daemon_ups.send_snmp_command(MiB.output_load, display=False))
            bat_capacity = int(daemon_ups.send_snmp_command(MiB.battery_capacity, display=False))
            time_on_bat = float(daemon_ups.send_snmp_command(MiB.time_on_battery, display=False))
            remain_run_time = float(daemon_ups.send_snmp_command(MiB.battery_runtime_remain, display=False))

            if UT_CONST.quit:
                print('[{}] {} Received Quit Signal'.format(
                    time_str, ups_states['warning']))
                sys.exit(0)

            # Check for load alarms
            if bat_load > crit_load_level:
                print('[{}] {} Overload Fault {}%, Suspend will execute with no resume possible. {}'.format(
                    time_str, ups_states['critical'], bat_load, reset_style))
                overload_fault = True
            elif bat_load > warn_load_level:
                print('[{}] {} Battery Load High: {}%'.format(
                    time_str, ups_states['warning'], bat_load))

            # Not on Battery
            if time_on_bat == 0.0:
                if args.verbose or ready_status:
                    print('[{}] {} Loading: {}%, Capacity: {}%, Power: {}W, Battery Status: {}'.format(
                        time_str, ups_states['ready'], bat_load, bat_capacity, out_power, bat_status))
                    ready_status = False
                if shutting_down:
                    daemon_ups.daemon.execute_script('cancel_shutdown_script')
                    print('[{}] {} Cancelling Shutdown'.format(
                        time_str, ups_states['good']))
                    shutting_down = False
                    ready_status = True
                if active_sleep != normal_sleep:
                    print('[{}] {} WARNING condition has ended: increase update interval from {} to {}'.format(
                          time_str, ups_states['ready'], active_sleep, normal_sleep))
                    active_sleep = normal_sleep
                    ready_status = True
                if bat_capacity < crit_bat_level:
                    print('[{}] {} Battery Nearly Exhausted, {:.2f}m/{}% remaining'.format(
                        time_str, ups_states['critical'], remain_run_time, bat_capacity))
                    print('[{}] {} UPS Battery Charging'.format(
                        time_str, ups_states['ready']))
                elif bat_capacity < warn_bat_level:
                    print('[{}] {} Battery Low, {}m/{}% remaining'.format(
                        time_str, ups_states['warning'], remain_run_time, bat_capacity))
                    print('[{}]: {} UPS Battery Charging {}'.format(
                        time_str, norm_style, reset_style))
            # On Battery condition
            elif time_on_bat > 0.0:
                print('[{}] {} System on UPS Power for {:.2f}min: {:.2f}m/{}% of battery remaining'.format(
                    time_str, ups_states['fault'], time_on_bat, remain_run_time, bat_capacity))
                if active_sleep != fault_sleep:
                    if (remain_run_time < warn_runtime_rem) or (bat_capacity < warn_bat_level):
                        # Warning Condition
                        print('[{}] {} battery low {:.2f}/{}%: reduce update interval from {} to {}'.format(
                              time_str, ups_states['warning'], remain_run_time, bat_capacity, normal_sleep, fault_sleep))
                        active_sleep = fault_sleep
                if not shutting_down:
                    if bat_status == 'Battery Low' or bat_capacity < crit_bat_level or remain_run_time < crit_runtime_rem:
                        # Call shutdown script
                        print('[{}] {} Battery Low Signal. Calling shutdown script'.format(
                            time_str, ups_states['critical']))
                        shutting_down = True
                        daemon_ups.daemon.execute_script('shutdown_script')

            if suspend_state:
                if time_on_bat < daemon_ups.daemon.daemon_params['threshold_time_on_battery']['crit'] and not overload_fault:
                    print('[{}] {} Running Resume Script'.format(time_str, ups_states['good']))
                    execute_result = daemon_ups.daemon.execute_script('resume_script')
                    if execute_result[0]:
                        print('[{}] {} Resume Script failed to execute - {}'.format(
                            time_str, ups_states['error'], execute_result))
                    suspend_state = False
                    ready_status = True

            if not suspend_state:
                if time_on_bat > daemon_ups.daemon.daemon_params['threshold_time_on_battery']['crit'] or bat_load > crit_load_level:
                    print('[{}] {} Running Suspend Script'.format(time_str, ups_states['warning']))
                    execute_result = daemon_ups.daemon.execute_script('suspend_script')
                    if execute_result[0]:
                        print('[{}] {} Suspend Script failed to execute - {}'.format(
                            time_str, ups_states['error'], execute_result))
                    suspend_state = True

            daemon_sleep(active_sleep)


if __name__ == "__main__":
    main()