cros/multimedia/cfm_facade_adapter.py

# Copyright 2017 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.

"""An adapter to remotely access the CFM facade on DUT."""

import logging
import os
import tempfile


class CFMFacadeRemoteAdapter(object):
    """CFMFacadeRemoteAdapter is an adapter to remotely control CFM on DUT.

    The Autotest host object representing the remote DUT, passed to this
    class on initialization, can be accessed from its _client property.
    """
    _RESTART_UI_DELAY = 10

    def __init__(self, host, remote_facade_proxy):
        """Construct a CFMFacadeRemoteAdapter.

        @param host: Host object representing a remote host.
        @param remote_facade_proxy: RemoteFacadeProxy object.
        """
        self._client = host
        self._proxy = remote_facade_proxy


    @property
    def _cfm_proxy(self):
        return self._proxy.cfm_main_screen


    @property
    def main_screen(self):
        """CFM main screen API."""
        return self._proxy.cfm_main_screen


    @property
    def mimo_screen(self):
        """CFM mimo screen API."""
        return self._proxy.cfm_mimo_screen


    def enroll_device(self):
        """Enroll device into CFM."""
        logging.info('Enrolling CfM device...')
        self._cfm_proxy.enroll_device()


    def restart_chrome_for_cfm(self, extra_chrome_args=None):
        """
        Restart chrome for CFM.

        @param extra_chrome_args a list with extra command line arguments for
                Chrome.
        """
        self._cfm_proxy.restart_chrome_for_cfm(extra_chrome_args)

    def reboot_device_with_chrome_api(self):
        """Reboot device using Chrome runtime API."""
        self._cfm_proxy.reboot_device_with_chrome_api()


    def skip_oobe_after_enrollment(self):
        """Skips oobe and goes to the app landing page after enrollment."""
        logging.info('Skipping OOBE...')
        self._cfm_proxy.skip_oobe_after_enrollment()


    def wait_for_telemetry_commands(self):
        """Wait for telemetry commands."""
        self._cfm_proxy.wait_for_telemetry_commands()


    def wait_for_hangouts_telemetry_commands(self):
        """Wait for Hangouts App telemetry commands."""
        logging.info('Waiting for Hangouts telemetry commands...')
        self._cfm_proxy.wait_for_hangouts_telemetry_commands()


    def wait_for_meetings_telemetry_commands(self):
        """Waits for Meet App telemetry commands."""
        logging.info('Waiting for Meet telemetry commands...')
        self._cfm_proxy.wait_for_meetings_telemetry_commands()


    def wait_for_meetings_in_call_page(self):
        """Waits for the in-call page to launch."""
        self._cfm_proxy.wait_for_meetings_in_call_page()


    def wait_for_meetings_landing_page(self):
        """Waits for the landing page screen."""
        self._cfm_proxy.wait_for_meetings_landing_page()


    # UI commands/functions
    def wait_for_oobe_start_page(self):
        """Wait for oobe start screen to launch."""
        self._cfm_proxy.wait_for_oobe_start_page()


    def skip_oobe_screen(self):
        """Skip Chromebox for Meetings oobe screen."""
        self._cfm_proxy.skip_oobe_screen()


    def is_oobe_start_page(self):
        """Check if device is on CFM oobe start screen.

        @return a boolean, based on oobe start page status.
        """
        return self._cfm_proxy.is_oobe_start_page()


    # Hangouts commands/functions
    def start_new_hangout_session(self, session_name):
        """Start a new hangout session.

        @param session_name: Name of the hangout session.
        """
        self._cfm_proxy.start_new_hangout_session(session_name)


    def end_hangout_session(self):
        """End current hangout session."""
        self._cfm_proxy.end_hangout_session()


    def take_screenshot(self):
        """
        Takes a screenshot on the DUT.

        @return The file path to the screenshot on the DUT or None.
        """
        # No suffix since cfm_proxy.take_screenshot() automactially appends one.
        with tempfile.NamedTemporaryFile() as f:
            basename = os.path.basename(f.name)
            return self._cfm_proxy.take_screenshot(basename)

    def get_latest_callgrok_file_path(self):
        """
        @return The path to the lastest callgrok log file, if any.
        """
        return self._cfm_proxy.get_latest_callgrok_file_path()


    def get_latest_pa_logs_file_path(self):
        """
        @return The path to the lastest packaged app log file, if any.
        """
        return self._cfm_proxy.get_latest_pa_logs_file_path()


    def get_all_pa_logs_file_path(self):
        """
        @return The path to all packaged app log files, if any.
        """
        return self._cfm_proxy.get_all_pa_logs_file_path()


    def is_in_hangout_session(self):
        """Check if device is in hangout session.

        @return a boolean, for hangout session state.
        """
        return self._cfm_proxy.is_in_hangout_session()


    def is_ready_to_start_hangout_session(self):
        """Check if device is ready to start a new hangout session.

        @return a boolean for hangout session ready state.
        """
        return self._cfm_proxy.is_ready_to_start_hangout_session()


    def join_meeting_session(self, session_name):
        """Join a meeting.

        @param session_name: Name of the meeting session.
        """
        self._cfm_proxy.join_meeting_session(session_name)


    def start_meeting_session(self):
        """Start a meeting.

        @return code for the started meeting.
        """
        return self._cfm_proxy.start_meeting_session()


    def end_meeting_session(self):
        """End current meeting session."""
        self._cfm_proxy.end_meeting_session()


    def get_participant_count(self):
        """Gets the total participant count in a call."""
        return self._cfm_proxy.get_participant_count()


    # Diagnostics commands/functions
    def is_diagnostic_run_in_progress(self):
        """Check if hotrod diagnostics is running.

        @return a boolean for diagnostic run state.
        """
        return self._cfm_proxy.is_diagnostic_run_in_progress()


    def wait_for_diagnostic_run_to_complete(self):
        """Wait for hotrod diagnostics to complete."""
        self._cfm_proxy.wait_for_diagnostic_run_to_complete()


    def run_diagnostics(self):
        """Run hotrod diagnostics."""
        self._cfm_proxy.run_diagnostics()


    def get_last_diagnostics_results(self):
        """Get latest hotrod diagnostics results.

        @return a dict with diagnostic test results.
        """
        return self._cfm_proxy.get_last_diagnostics_results()


    # Mic audio commands/functions
    def is_mic_muted(self):
        """Check if mic is muted.

        @return a boolean for mic mute state.
        """
        return self._cfm_proxy.is_mic_muted()


    def mute_mic(self):
        """Local mic mute from toolbar."""
        self._cfm_proxy.mute_mic()


    def unmute_mic(self):
        """Local mic unmute from toolbar."""
        self._cfm_proxy.unmute_mic()


    def remote_mute_mic(self):
        """Remote mic mute request from cPanel."""
        self._cfm_proxy.remote_mute_mic()


    def remote_unmute_mic(self):
        """Remote mic unmute request from cPanel."""
        self._cfm_proxy.remote_unmute_mic()


    def get_mic_devices(self):
        """Get all mic devices detected by hotrod."""
        return self._cfm_proxy.get_mic_devices()


    def get_preferred_mic(self):
        """Get mic preferred for hotrod.

        @return a str with preferred mic name.
        """
        return self._cfm_proxy.get_preferred_mic()


    def set_preferred_mic(self, mic):
        """Set preferred mic for hotrod.

        @param mic: String with mic name.
        """
        self._cfm_proxy.set_preferred_mic(mic)


    # Speaker commands/functions
    def get_speaker_devices(self):
        """Get all speaker devices detected by hotrod.

        @return a list of speaker devices.
        """
        return self._cfm_proxy.get_speaker_devices()


    def get_preferred_speaker(self):
        """Get speaker preferred for hotrod.

        @return a str with preferred speaker name.
        """
        return self._cfm_proxy.get_preferred_speaker()


    def set_preferred_speaker(self, speaker):
        """Set preferred speaker for hotrod.

        @param speaker: String with speaker name.
        """
        self._cfm_proxy.set_preferred_speaker(speaker)


    def set_speaker_volume(self, volume_level):
        """Set speaker volume.

        @param volume_level: String value ranging from 0-100 to set volume to.
        """
        self._cfm_proxy.set_speaker_volume(volume_level)


    def get_speaker_volume(self):
        """Get current speaker volume.

        @return a str value with speaker volume level 0-100.
        """
        return self._cfm_proxy.get_speaker_volume()


    def play_test_sound(self):
        """Play test sound."""
        self._cfm_proxy.play_test_sound()


    # Camera commands/functions
    def get_camera_devices(self):
        """Get all camera devices detected by hotrod.

        @return a list of camera devices.
        """
        return self._cfm_proxy.get_camera_devices()


    def get_preferred_camera(self):
        """Get camera preferred for hotrod.

        @return a str with preferred camera name.
        """
        return self._cfm_proxy.get_preferred_camera()


    def set_preferred_camera(self, camera):
        """Set preferred camera for hotrod.

        @param camera: String with camera name.
        """
        self._cfm_proxy.set_preferred_camera(camera)


    def is_camera_muted(self):
        """Check if camera is muted (turned off).

        @return a boolean for camera muted state.
        """
        return self._cfm_proxy.is_camera_muted()


    def mute_camera(self):
        """Turned camera off."""
        self._cfm_proxy.mute_camera()


    def unmute_camera(self):
        """Turned camera on."""
        self._cfm_proxy.unmute_camera()


    def move_camera(self, camera_motion):
        """
        Move camera(PTZ commands).

        @param camera_motion: Set of allowed commands
            defined in cfmApi.move_camera.
        """
        self._cfm_proxy.move_camera(camera_motion)

    def get_media_info_data_points(self):
        """
        Gets media info data points containing media stats.

        These are exported on the window object when the
        ExportMediaInfo mod is enabled.

        @returns A list with dictionaries of media info data points.
        @raises RuntimeError if the data point API is not available.
        """
        return self._cfm_proxy.get_media_info_data_points()