#  Copyright (C) 2025 The Android Open Source Project
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

# Lint as: python3
"""Utils for calling Telecom related APIs."""

from collections.abc import Callable
import datetime
import enum
import time

from mobly import asserts
from mobly.controllers import android_device

_DELAY_BETWEEN_ACTIONS = datetime.timedelta(seconds=1)
_DELAY_BETWEEN_CALL_STATE_CHECKS = datetime.timedelta(seconds=1)
_WAIT_CALL_CONNECTED_TIMEOUT = datetime.timedelta(seconds=10)
_WAIT_CALL_DISCONNECTED_TIMEOUT = datetime.timedelta(seconds=10)
_WAIT_CALL_RINGING_TIMEOUT = datetime.timedelta(seconds=10)


@enum.unique
class CallState(enum.IntEnum):
  """Represents the state of a call."""

  STATE_NEW = 0
  STATE_DIALING = 1
  STATE_RINGING = 2
  STATE_HOLDING = 3
  STATE_ACTIVE = 4
  STATE_DISCONNECTED = 7
  STATE_SELECT_PHONE_ACCOUNT = 8
  STATE_CONNECTING = 9
  STATE_DISCONNECTING = 10
  STATE_PULLING_CALL = 11
  STATE_AUDIO_PROCESSING = 12
  STATE_SIMULATED_RINGING = 13


@enum.unique
class CallType(enum.IntEnum):
  """The type of a call.

  Attributes:
    UNKNOWN: Unknown call type.
    LEGACY: Traditional circuit-switched voice call (e.g., 2G/3G).
    ViLTE: Video over LTE.
    VOLTE: Voice over LTE.
    VOWIFI: Voice over Wi-Fi.
  """

  UNKNOWN = 0
  LEGACY = 1
  VILTE = 2
  VOLTE = 3
  VOWIFI = 4


def wait_until_or_assert(
    condition: Callable[[], bool],
    error_msg: str,
    timeout: datetime.timedelta,
) -> None:
  """Waits until the condition is met, or asserts if timeout.

  Args:
    condition: Represents the condition to wait for.
    error_msg: The error message to be included in the assertion failure.
    timeout: The maximum time to wait for the condition to be met.

  Raises:
    mobly.signals.TestFailure: When the condition is not met within the timeout.
  """
  end_time = time.monotonic() + timeout.total_seconds()
  while time.monotonic() < end_time:
    if condition():
      return
    time.sleep(_DELAY_BETWEEN_ACTIONS.total_seconds())
  asserts.fail(f'{error_msg} within {timeout.total_seconds()} seconds')


def originate_call(
    ad: android_device.AndroidDevice,
    remote_phone_number: str,
) -> None:
  """Originates a call from DUT to the specified number."""
  ad.log.info(f'Originating call to {remote_phone_number}')
  ad.tele.telecomPlaceCall(remote_phone_number)
  wait_until_or_assert(
      lambda: ad.tele.telecomGetCallState(remote_phone_number)
      == CallState.STATE_DIALING,
      error_msg=f'{ad} Failed to place call to {remote_phone_number}',
      timeout=_WAIT_CALL_CONNECTED_TIMEOUT,
  )
  ad.log.info(f'Call with {remote_phone_number} is placed')


def terminate_call(
    ad: android_device.AndroidDevice, remote_phone_number: str
) -> None:
  """Terminates a call from DUT to the specified number."""
  ad.log.info(f'Waiting for call from {remote_phone_number}')
  wait_until_or_assert(
      lambda: ad.tele.telecomGetCallState(remote_phone_number)
      == CallState.STATE_RINGING,
      error_msg=f'{ad} Failed to receive call from {remote_phone_number}',
      timeout=_WAIT_CALL_RINGING_TIMEOUT,
  )
  ad.log.info(f'Call from {remote_phone_number} is ringing')

  ad.log.info(f'Terminating call from {remote_phone_number}')
  ad.tele.telecomAcceptRingingCall(remote_phone_number)
  wait_until_or_assert(
      lambda: ad.tele.telecomGetCallState(remote_phone_number)
      == CallState.STATE_ACTIVE,
      error_msg=f'{ad} Failed to accept call from {remote_phone_number}',
      timeout=_WAIT_CALL_CONNECTED_TIMEOUT,
  )
  ad.log.info(f'Call from {remote_phone_number} is accepted')


def monitor_call(
    ad: android_device.AndroidDevice,
    remote_phone_number: str,
    call_duration: datetime.timedelta,
    call_type: CallType | None = None,
) -> None:
  """Monitors the ongoing call with the specified duration."""
  end_time = time.monotonic() + call_duration.total_seconds()
  while time.monotonic() < end_time:
    asserts.assert_equal(
        ad.tele.telecomGetCallState(remote_phone_number),
        CallState.STATE_ACTIVE,
        f'{ad} Call is disconnected during the call duration',
    )
    if call_type is not None:
      current_call_type = ad.tele.telecomGetCallType(remote_phone_number)
      asserts.assert_equal(
          current_call_type,
          call_type,
          f'{ad} Call is changed to {CallType(current_call_type).name},'
          f' expected: {call_type.name}',
      )
    time.sleep(_DELAY_BETWEEN_CALL_STATE_CHECKS.total_seconds())


def end_call(
    ad: android_device.AndroidDevice, remote_phone_number: str
) -> None:
  """Ends the ongoing call."""
  ad.log.info(f'Ending call with {remote_phone_number}')
  ad.tele.telecomEndCall(remote_phone_number)
  wait_until_or_assert(
      lambda: ad.tele.telecomGetCallState(remote_phone_number)
      == CallState.STATE_DISCONNECTED,
      error_msg=(
          f'{ad} Failed to end the ongoing call with {remote_phone_number}'
      ),
      timeout=_WAIT_CALL_DISCONNECTED_TIMEOUT,
  )
  ad.log.info(f'Call with {remote_phone_number} is ended')


def set_default_dialer(
    ad: android_device.AndroidDevice, package_name: str
) -> None:
  """Sets the default dialer to the specified package."""
  ad.adb.shell(['telecom', 'set-default-dialer', package_name])
  ad.log.info(f'Set default dialer to {package_name}')


def reset_default_dialer(ad: android_device.AndroidDevice) -> None:
  """Resets the default dialer to the system dialer."""
  system_dialer = (
      ad.adb.shell(['telecom', 'get-system-dialer']).decode().strip()
  )
  ad.adb.shell(['telecom', 'set-default-dialer', system_dialer])
  ad.log.info(f'Reset default dialer to {system_dialer}')