xref: /aosp_15_r20/external/autotest/client/cros/multimedia/system_facade.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Facade to access the system-related functionality."""

import six
import os
import threading
import time

from autotest_lib.client.bin import utils


class SystemFacadeLocalError(Exception):
    """Error in SystemFacadeLocal."""
    pass


class SystemFacadeLocal(object):
    """Facede to access the system-related functionality.

    The methods inside this class only accept Python native types.

    """
    SCALING_GOVERNOR_MODES = [
            'performance',
            'powersave',
            'userspace',
            'ondemand',
            'conservative',
            'schedutil',
            'interactive', # deprecated since kernel v4.14
            'sched' # deprecated since kernel v4.14
            ]

    def __init__(self):
        self._bg_worker = None

    def set_scaling_governor_mode(self, index, mode):
        """Set mode of CPU scaling governor on one CPU.

        @param index: CPU index starting from 0.

        @param mode: Mode of scaling governor, accept 'interactive' or
                     'performance'.

        @returns: The original mode.

        """
        if mode not in self.SCALING_GOVERNOR_MODES:
            raise SystemFacadeLocalError('mode %s is invalid' % mode)

        governor_path = os.path.join(
                '/sys/devices/system/cpu/cpu%d' % index,
                'cpufreq/scaling_governor')
        if not os.path.exists(governor_path):
            raise SystemFacadeLocalError(
                    'scaling governor of CPU %d is not available' % index)

        original_mode = utils.read_one_line(governor_path)
        utils.open_write_close(governor_path, mode)

        return original_mode


    def get_cpu_usage(self):
        """Returns machine's CPU usage.

        Returns:
            A dictionary with 'user', 'nice', 'system' and 'idle' values.
            Sample dictionary:
            {
                'user': 254544,
                'nice': 9,
                'system': 254768,
                'idle': 2859878,
            }
        """
        return utils.get_cpu_usage()


    def compute_active_cpu_time(self, cpu_usage_start, cpu_usage_end):
        """Computes the fraction of CPU time spent non-idling.

        This function should be invoked using before/after values from calls to
        get_cpu_usage().
        """
        return utils.compute_active_cpu_time(cpu_usage_start,
                                                  cpu_usage_end)


    def get_mem_total(self):
        """Returns the total memory available in the system in MBytes."""
        return utils.get_mem_total()


    def get_mem_free(self):
        """Returns the currently free memory in the system in MBytes."""
        return utils.get_mem_free()

    def get_mem_free_plus_buffers_and_cached(self):
        """
        Returns the free memory in MBytes, counting buffers and cached as free.

        This is most often the most interesting number since buffers and cached
        memory can be reclaimed on demand. Note however, that there are cases
        where this as misleading as well, for example used tmpfs space
        count as Cached but can not be reclaimed on demand.
        See https://www.kernel.org/doc/Documentation/filesystems/tmpfs.txt.
        """
        return utils.get_mem_free_plus_buffers_and_cached()

    def get_ec_temperatures(self):
        """Uses ectool to return a list of all sensor temperatures in Celsius.
        """
        return utils.get_ec_temperatures()

    def get_current_temperature_max(self):
        """
        Returns the highest reported board temperature (all sensors) in Celsius.
        """
        return utils.get_current_temperature_max()

    def get_current_board(self):
        """Returns the current device board name."""
        return utils.get_current_board()


    def get_chromeos_release_version(self):
        """Returns chromeos version in device under test as string. None on
        fail.
        """
        return utils.get_chromeos_release_version()

    def get_num_allocated_file_handles(self):
        """
        Returns the number of currently allocated file handles.
        """
        return utils.get_num_allocated_file_handles()

    def get_storage_statistics(self, device=None):
        """
        Fetches statistics for a storage device.
        """
        return utils.get_storage_statistics(device)

    def get_energy_usage(self):
        """
        Gets the energy counter value as a string.
        """
        return utils.get_energy_usage()

    def start_bg_worker(self, command):
        """
        Start executing the command in a background worker.
        """
        self._bg_worker = BackgroundWorker(command, do_process_output=True)
        self._bg_worker.start()

    def get_and_discard_bg_worker_output(self):
        """
        Returns the output collected so far since the last call to this method.
        """
        if self._bg_worker is None:
            SystemFacadeLocalError('Background worker has not been started.')

        return self._bg_worker.get_and_discard_output()

    def stop_bg_worker(self):
        """
        Stop the worker.
        """
        if self._bg_worker is None:
            SystemFacadeLocalError('Background worker has not been started.')

        self._bg_worker.stop()
        self._bg_worker = None


class BackgroundWorker(object):
    """
    Worker intended for executing a command in the background and collecting its
    output.
    """

    def __init__(self, command, do_process_output=False):
        self._bg_job = None
        self._command = command
        self._do_process_output = do_process_output
        self._output_lock = threading.Lock()
        self._process_output_thread = None
        self._stdout = six.StringIO()

    def start(self):
        """
        Start executing the command.
        """
        self._bg_job = utils.BgJob(self._command, stdout_tee=self._stdout)
        self._bg_job.sp.poll()
        if self._bg_job.sp.returncode is not None:
            self._exit_bg_job()

        if self._do_process_output:
            self._process_output_thread = threading.Thread(
                    target=self._process_output)
            self._process_output_thread.start()

    def _process_output(self, sleep_interval=0.01):
        while self._do_process_output:
            with self._output_lock:
                self._bg_job.process_output()
            time.sleep(sleep_interval)

    def get_and_discard_output(self):
        """
        Returns the output collected so far and then clears the output buffer.
        In other words, subsequent calls to this method will not include output
        that has already been returned before.
        """
        output = ""
        with self._output_lock:
            self._stdout.flush()
            output = self._stdout.getvalue()
            self._stdout.truncate(0)
            self._stdout.seek(0)
        return output

    def stop(self):
        """
        Stop executing the command.
        """
        if self._do_process_output:
            self._do_process_output = False
            self._process_output_thread.join(1)
        self._exit_bg_job()

    def _exit_bg_job(self):
        utils.nuke_subprocess(self._bg_job.sp)
        utils.join_bg_jobs([self._bg_job])
        if self._bg_job.result.exit_status > 0:
            raise SystemFacadeLocalError('Background job failed: %s' %
                                          self._bg_job.result.command)