| # Lint as: python2, python3 |
| # -*- coding: utf-8 -*- |
| # Copyright 2014 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """System tools required for Chameleond execution.""" |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import logging |
| import os |
| import subprocess |
| import sys |
| import threading |
| |
| import six |
| |
| |
| class _SystemTools(object): |
| """A class to wrap the required tools for Chameleond execution.""" |
| |
| _TOOL_PATHS = { |
| "aplay": "/usr/bin/aplay", |
| "arecord": "/usr/bin/arecord", |
| "avsync": "/usr/bin/avsync", |
| "bluetoothd": "/usr/sbin/bluetoothd", |
| "btmon": "/usr/bin/btmon", |
| "chameleond": "/etc/init.d/chameleond", |
| "cp": "/usr/bin/cp", |
| "date": "/bin/date", |
| "dut-control": "/usr/local/bin/dut-control", |
| "grep": "/bin/grep", |
| "histogram": "/usr/bin/histogram", |
| "hpd_control": "/usr/bin/hpd_control", |
| "i2cdump": "/usr/local/sbin/i2cdump", |
| "i2cget": "/usr/local/sbin/i2cget", |
| "i2cset": "/usr/local/sbin/i2cset", |
| "id": "/usr/bin/id", |
| "killall": "/usr/bin/killall", |
| "logger": "/usr/bin/logger", |
| "lsmod": "/sbin/lsmod", |
| "lspci": "lspci", |
| "memtool": "/usr/bin/memtool", |
| "modinfo": "/sbin/modinfo", |
| "modprobe": "/sbin/modprobe", |
| "mpris-proxy": "/usr/bin/mpris-proxy", |
| "ofono": "/usr/sbin/ofonod", |
| "pacat.local": "/usr/local/bin/pacat", |
| "pacat": "/usr/bin/pacat", |
| "pactl.local": "/usr/local/bin/pactl", |
| "pactl": "/usr/bin/pactl", |
| "pgrep": "/usr/bin/pgrep", |
| "pipewire": "/usr/bin/pipewire", |
| "pixeldump": "/usr/bin/pixeldump", |
| "playerctl": "/usr/bin/playerctl", |
| "printer": "/usr/bin/printer", |
| "pulseaudio.local": "/usr/local/bin/pulseaudio", |
| "pulseaudio": "/usr/bin/pulseaudio", |
| "pw-play": "/usr/bin/pw-play", |
| "pw-record": "/usr/bin/pw-record", |
| "reboot": "/sbin/reboot", |
| "rm": "/usr/bin/rm", |
| "scp": "/usr/bin/scp", |
| "sed": "/usr/bin/sed", |
| "service": "/usr/sbin/service", |
| "servod": "/usr/local/bin/servod", |
| "ssh": "/usr/bin/ssh", |
| "su": "/bin/su", |
| # Although systemd should theoretically be in /usr/bin, |
| # the debian distro sometimes only installs it in /bin. |
| "systemctl": "/bin/systemctl", |
| "tail": "/usr/bin/tail", |
| "uname": "/bin/uname", |
| "wget": "/usr/bin/wget", |
| "wpctl": "/usr/bin/wpctl", |
| } |
| |
| def __init__(self): |
| """Constructs a _SystemTools object.""" |
| self._CheckRequiredTools() |
| self._pi_args = "su - pi -c".split() |
| # The pi environment is only needed when running bluetooth audio tests. |
| self._pi_env = {} |
| self._pi_uid = None |
| |
| def SetUpPiEnv(self, uid): |
| """Set up a proper environment for the user pi. |
| |
| Some audio processes have to be run as the user pi. |
| |
| Args: |
| uid: the pi user id |
| |
| We need to set XDG_RUNTIME_DIR for pi for pulseaudio tools to run |
| correctly. Setting environmental variable DBUS_SESSION_BUS_ADDRESS to force |
| the pulseaudio to connect to an existing DBSU session before starting a new |
| one. |
| """ |
| self._pi_uid = uid |
| self._pi_env = os.environ.copy() |
| self._pi_env["XDG_RUNTIME_DIR"] = f"/run/user/{uid}" |
| self._pi_env["DBUS_SESSION_BUS_ADDRESS"] = f"/run/user/{uid}/bus" |
| |
| def GetToolPath(self, name): |
| """Get the path of the tool. |
| |
| Args: |
| name: Name of the system tool. |
| |
| Returns: |
| The path of the tool; or None if not found. |
| """ |
| return self._TOOL_PATHS.get(name) |
| |
| def _CheckRequiredTools(self): |
| """Checks all the required tools exist. |
| |
| Raises: |
| SystemToolsError if missing a tool. |
| """ |
| for path in list(self._TOOL_PATHS.values()): |
| if not os.path.isfile(path): |
| # It is okay that some tools may not exist in a particular platform. |
| logging.warning("IOError: Required tool %s not existed", path) |
| |
| def _MakeCommandForPi(self, name, args_str, use_local_version=False): |
| """Combines the system tool and its parameters into a list. |
| |
| In python3, the following subprocess methods take env as a keyword |
| argument: |
| - subprocess.Popen |
| - subprocess.run |
| - subprocess.check_output |
| To pass environment variables to a method other than those above, e.g., |
| subprocess.call, the environment variables must be positioned before the |
| tool name. |
| |
| However, subprocess.check_output seems to have an issue such that the |
| env keyword argument does not take effect. Instead, the environment |
| string must be positioned before the tool name. To simplify and ensure |
| consistency across all subprocess methods, the environment string is |
| placed before the tool name, and the env keyword argument is not used. |
| |
| Args: |
| name: the name of the system tool. |
| args_str: the args string (not a list) |
| use_local_version: True to use a local version of the executable. |
| Default value is False to favor the native version. |
| |
| Returns: |
| A list representing the complete command. |
| """ |
| xdg_runtime_dir = self._pi_env.get("XDG_RUNTIME_DIR", "") |
| dbus_session_bus_addr = self._pi_env.get("DBUS_SESSION_BUS_ADDRESS", "") |
| s1 = f"XDG_RUNTIME_DIR={xdg_runtime_dir}" |
| s2 = f"DBUS_SESSION_BUS_ADDRESS={dbus_session_bus_addr}" |
| env_str = " ".join([s1, s2]) |
| |
| tool_path = self._FindPreferredExecutablePath(name, use_local_version) |
| command = self._pi_args + [" ".join([env_str, tool_path, args_str])] |
| logging.info("command: %s", str(command)) |
| return command |
| |
| def _MakeCommand(self, name, args): |
| """Combines the system tool and its parameters into a list. |
| |
| Args: |
| name: Name of the system tool. |
| args: List of arguments passed in by user. |
| |
| Returns: |
| A list representing the complete command. |
| """ |
| command = [self._TOOL_PATHS[name]] + [str(arg) for arg in args] |
| logging.info("command: %s", str(command)) |
| return command |
| |
| def OrigCall(self, name, *args): |
| """Calls the tool with arguments using subprocess.call(). |
| |
| Args: |
| name: The name of the tool. |
| *args: The arguments of the tool. |
| """ |
| command = self._MakeCommand(name, args) |
| return subprocess.call(command) |
| |
| def OrigCallAsPi(self, name, args_str): |
| """Calls the tool with arguments as Pi using subprocess.call(). |
| |
| This is primarily used on Raspberry Pi as the chameleon host. This is |
| needed as some commands, in pulseaudio and pipewire, are required |
| to run with pi as the user. The chameleond root user is not allowed to |
| access pulseaudio and pipewire for security concern. |
| |
| Args: |
| name: The name of the tool. |
| args_str: the args string (not a list) |
| """ |
| command = self._MakeCommandForPi(name, args_str) |
| return subprocess.call(command) |
| |
| def Call(self, name, *args): |
| """Calls the tool with arguments using subprocess.check_call(). |
| |
| Args: |
| name: The name of the tool. |
| *args: The arguments of the tool. |
| """ |
| command = self._MakeCommand(name, args) |
| subprocess.check_call(command) |
| |
| def Output(self, name, *args): |
| """Calls the tool with arguments and returns its output. |
| |
| Args: |
| name: The name of the tool. |
| *args: The arguments of the tool. |
| |
| Returns: |
| The output message of the call, including stderr message. |
| """ |
| command = self._MakeCommand(name, args) |
| try: |
| output = subprocess.check_output(command, stderr=subprocess.STDOUT) |
| except Exception as e: |
| logging.error("Output error: %s", e) |
| return "" |
| |
| if sys.version_info >= (3, 0): |
| return output.decode() |
| else: |
| return output |
| |
| def _FindPreferredExecutablePath(self, name, use_local_version=False): |
| """Find the preferred executable path. |
| |
| For some daemons, e.g., pulseaudio, there may exist a locally made |
| version. In this situation, the executables will reside in |
| /usr/local/bin instead of the original /usr/bin when they are installed. |
| |
| Args: |
| name: The name of the tool. |
| use_local_version: True to use a local version of the executable. |
| Default value is False to favor the native version. |
| |
| Returns: |
| The preferred executable path if a locally made version exists; |
| otherwise, the original installed path. |
| """ |
| local_name = name + ".local" |
| local_path = self._TOOL_PATHS.get(local_name) |
| |
| if ( |
| use_local_version |
| and bool(local_path) |
| and os.path.isfile(local_path) |
| ): |
| return local_path |
| else: |
| return self._TOOL_PATHS.get(name) |
| |
| def OutputAsPi(self, name, args_str, use_local_version): |
| """Calls the tool with arguments as Pi and returns its output. |
| |
| This is primarily used on Raspberry Pi as the chameleon host. This is |
| needed as some commands, e.g., pactl and pacat in pulseaudio, require |
| to run with pi as the user. The chameleond root user is not allowed to |
| access pulseaudio for security concern. |
| |
| Args: |
| name: The name of the tool. |
| args_str: the args string (not a list) |
| use_local_version: True to use a local version of the executable |
| |
| Returns: |
| The output message of the call, including stderr message. |
| """ |
| command = self._MakeCommandForPi( |
| name, args_str, use_local_version=use_local_version |
| ) |
| |
| try: |
| output = subprocess.check_output(command) |
| except Exception as e: |
| logging.error("Output error: %s", e) |
| return "" |
| |
| if sys.version_info >= (3, 0): |
| return output.decode() |
| else: |
| return output |
| |
| def DelayedCall(self, time, name, *args): |
| """Calls the tool with arguments after a given delay. |
| |
| The method returns first before the execution. |
| |
| Args: |
| time: The time in second. |
| name: The name of the tool. |
| *args: The arguments of the tool. |
| """ |
| threading.Timer(time, lambda: self.Call(name, *args)).start() |
| |
| def RunInSubprocess(self, name, *args): |
| """Calls the tool and run it in a separate process. |
| |
| This tool will be useful for starting and later killing aplay and arecord |
| processes which have to be interrupted. The command outputs are channelled |
| to stdout and/or stderr. |
| |
| Args: |
| name: The name of the tool. |
| *args: The arguments of the tool. |
| |
| Returns: |
| process: The subprocess spawned for the command. |
| """ |
| return self.RunInSubprocessOutputToFile(name, subprocess.PIPE, *args) |
| |
| def RunInSubprocessAsPi(self, name, args_str, use_local_version): |
| """Calls the tool and run it in a separate process as user Pi. |
| |
| This is primarily used on Raspberry Pi as the chameleon host. This is |
| needed as some commands, e.g., pactl and pacat in pulseaudio, require |
| to run with pi as the user. The chameleond root user is not allowed to |
| access pulseaudio for security concern. |
| |
| Args: |
| name: The name of the tool. |
| args_str: the args string (not a list) |
| use_local_version: True to use a local version of the executable |
| |
| Returns: |
| process: The subprocess spawned for the command. |
| """ |
| command = self._MakeCommandForPi( |
| name, args_str, use_local_version=use_local_version |
| ) |
| process = subprocess.Popen( |
| command, stdout=subprocess.PIPE, stderr=subprocess.PIPE |
| ) |
| process.name = name |
| return process |
| |
| def RunInSubprocessOutputToFile(self, name, handle, *args): |
| """Calls the tool and run it in a separate process, and outputs to a file. |
| |
| Args: |
| name: The name of the tool. |
| handle: The file handle to output stdout and stderr. |
| *args: The arguments of the tool. |
| |
| Returns: |
| process: The subprocess spawned for the command. |
| """ |
| command = self._MakeCommand(name, args) |
| process = subprocess.Popen(command, stdout=handle, stderr=handle) |
| process.name = name |
| return process |
| |
| def GetSubprocessOutput(self, process, decode=True): |
| """Returns the output of the command called in the process spawned. |
| |
| Args: |
| process: The subprocess in which a command is called. |
| decode: If true, decode output from byte strings to strings. |
| |
| Returns: |
| A tuple (return_code, out, err). |
| return_code: 0 on success, 1 on error. |
| out: Content of command output to stdout, usually when command succeeds. |
| err: Content of command output to stderr when an error occurs. |
| """ |
| out, err = process.communicate() |
| return_code = process.returncode |
| if decode: |
| return (return_code, six.ensure_text(out), six.ensure_text(err)) |
| return (return_code, out, six.ensure_text(err)) |
| |
| |
| # Singleton |
| SystemTools = _SystemTools() |