diff --git a/framework/core/audioAmplifier/__init__.py b/framework/core/audioAmplifier/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/framework/core/audioAmplifier/base.py b/framework/core/audioAmplifier/base.py new file mode 100644 index 0000000..2dec4d2 --- /dev/null +++ b/framework/core/audioAmplifier/base.py @@ -0,0 +1,107 @@ +from abc import ABC, abstractmethod + +class AudioAmplifier(ABC): + """ + Abstract base class defining the interface for an audio amplifier controller. + + Implementations must provide methods for controlling power, volume, + input source, mute state, sound mode, and retrieving status information. + """ + + @abstractmethod + def power_on(self): + """Power on the amplifier.""" + pass + + @abstractmethod + def power_off(self): + """Power off the amplifier.""" + pass + + @abstractmethod + def set_volume(self, volume: float): + """ + Set the amplifier volume. + + :param volume: Desired volume level. + """ + pass + + @abstractmethod + def mute(self, state: bool): + """ + Mute or unmute the amplifier. + + :param state: True to mute, False to unmute. + """ + pass + + @abstractmethod + def list_inputs(self) -> list[str]: + """ + Get the list of available input sources supported by the amplifier. + """ + pass + + @abstractmethod + def list_sound_modes(self) -> list[str]: + """ + Get the list of available sound modes supported by the amplifier. + """ + pass + + @abstractmethod + def set_input(self, input_name: str): + """ + Set the input source of the amplifier. + + :param input_name: Name of the input source (e.g., "TV", "CD"). + """ + pass + + @abstractmethod + def set_sound_mode(self, input_name: str): + """ + Set the sound mode of the amplifier. + + :param input_name: Name of the sound mode (e.g., "TV", "CD"). + """ + pass + + @abstractmethod + def update_state(self): + """ + Refresh the internal state from the amplifier. + + Typically required before retrieving status properties. + """ + pass + + @abstractmethod + def get_power(self) -> str: + """Get the current power state (e.g., "ON", "OFF").""" + pass + + @abstractmethod + def get_volume(self) -> float: + """Get the current volume level.""" + pass + + @abstractmethod + def get_input(self) -> str: + """Get the currently selected input source.""" + pass + + @abstractmethod + def is_muted(self) -> bool: + """Check whether the amplifier is muted.""" + pass + + @abstractmethod + def get_status(self) -> dict: + """ + Get a dictionary of key status information (power, volume, input, mute). + + :return: Dictionary of current amplifier state. + """ + pass diff --git a/framework/core/audioAmplifier/denon_controller.py b/framework/core/audioAmplifier/denon_controller.py new file mode 100644 index 0000000..a41690c --- /dev/null +++ b/framework/core/audioAmplifier/denon_controller.py @@ -0,0 +1,78 @@ +import asyncio +from denonavr import DenonAVR +from .base import AudioAmplifier + +class DenonAVRController(AudioAmplifier): + + def __init__(self, host: str): + self.receiver = DenonAVR(host) + self.setup() + + def setup(self): + asyncio.run(self.receiver.async_setup()) + + def power_on(self): + asyncio.run(self.receiver.async_power_on()) + self.update_state() + + def power_off(self): + asyncio.run(self.receiver.async_power_off()) + self.update_state() + + + def set_volume(self, volume: float): + asyncio.run(self.receiver.async_set_volume(volume)) + self.update_state() + + def mute(self, state: bool): + asyncio.run(self.receiver.async_mute(state)) + self.update_state() + + def list_inputs(self) -> list[str]: + self.update_state() + return self.receiver.input_func_list + + def list_sound_modes(self) -> list[str]: + self.update_state() + return self.receiver.sound_mode_list + + def set_input(self, input_name: str): + available = self.list_inputs() + if input_name not in available: + raise ValueError(f"Invalid input: {input_name}. Available inputs: {available}") + asyncio.run(self.receiver.async_set_input_func(input_name)) + self.update_state() + + def set_sound_mode(self, mode: str): + available = self.list_sound_modes() + if mode not in available: + raise ValueError(f"Invalid sound mode: {mode}. Available modes: {available}") + asyncio.run(self.receiver.async_set_sound_mode(mode)) + self.update_state() + + def update_state(self): + asyncio.run(self.receiver.async_update()) + + def get_power(self) -> str: + return self.receiver.power + + def get_volume(self) -> float: + return self.receiver.volume + + def is_muted(self) -> bool: + return self.receiver.muted + + def get_input(self) -> str: + return self.receiver.input_func + + def get_sound_mode(self) -> str: + return self.receiver.sound_mode + + def get_status(self): + return { + "power": self.get_power(), + "volume": self.get_volume(), + "muted": self.is_muted(), + "input": self.get_input(), + "sound_mode": self.get_sound_mode(), + } diff --git a/framework/core/audioAmplifierController.py b/framework/core/audioAmplifierController.py new file mode 100644 index 0000000..e05a1dd --- /dev/null +++ b/framework/core/audioAmplifierController.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * If not stated otherwise in this file or this component's LICENSE file the +# * following copyright and licenses apply: +# * +# * Copyright 2023 RDK Management +# * +# * 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. +# * +#* ****************************************************************************** +#* +#* ** Project : RAFT +#* ** @addtogroup : core +#* ** @date : 01/07/2025 +#* ** +#* ** @brief : audio amplifier controller +#* ** +#* ****************************************************************************** + +from framework.core.logModule import logModule +from framework.core.audioAmplifier.denon_controller import DenonAVRController + +class audioAmplifierController(): + + def __init__(self, log:logModule, config:dict): + self._log = log + self.controllerType = config.get("type") + self.host = config.get("host") + + if self.controllerType == "denon": + self.audioAmplifier = DenonAVRController(self.host) + + def power_on(self): + self._log.info("Powering ON audio amplifier") + self.audioAmplifier.power_on() + + def power_off(self): + self._log.info("Powering OFF audio amplifier") + self.audioAmplifier.power_off() + + def set_volume(self, volume: float): + self._log.info("Setting audio amplifier volume") + self.audioAmplifier.set_volume(volume) + + def mute(self, state: bool): + self._log.info("Muting audio amplifier") + self.audioAmplifier.mute(state) + + def list_inputs(self) -> list[str]: + self._log.info("Listing the audio amplifier available inputs") + return self.audioAmplifier.list_inputs() + + def list_sound_modes(self) -> list[str]: + self._log.info("Listing the audio amplifier available sound modes") + return self.audioAmplifier.list_sound_modes() + + def set_input(self, input_name: str): + self._log.info("Setting audio amplifier input") + self.audioAmplifier.set_input(input_name) + + def set_sound_mode(self, mode: str): + self.audioAmplifier.set_sound_mode(mode) + + def get_power(self) -> str: + self._log.info("Getting audio amplifier power") + return self.audioAmplifier.get_power() + + def get_volume(self) -> float: + self._log.info("Getting audio amplifier volume") + return self.audioAmplifier.get_volume() + + def is_muted(self) -> bool: + self._log.info("Getting audio amplifier mute state") + return self.audioAmplifier.is_muted() + + def get_input(self) -> str: + self._log.info("Getting audio amplifier input") + return self.audioAmplifier.get_input() + + def get_sound_mode(self) -> str: + self._log.info("Getting audio amplifier sound mode") + return self.audioAmplifier.get_sound_mode() + + def get_status(self): + self._log.info("Getting audio amplifier status") + return self.audioAmplifier.get_status() diff --git a/framework/core/streamToFile.py b/framework/core/streamToFile.py index f0df434..0f1840f 100644 --- a/framework/core/streamToFile.py +++ b/framework/core/streamToFile.py @@ -21,8 +21,6 @@ def writeStreamToFile(self, inputStream: IOBase) -> None: Args: inputStream (IOBase): The input stream to be read from. - outFileName (str): The path of the output file where the stream data will be written. - If only a file name is given, the file will be written in the current tests log directory. """ self._fileHandle = open(self._filePath, 'a+', encoding='utf-8') self._stopThread = False @@ -84,10 +82,9 @@ def readUntil(self, searchString:str, retries: int = 5) -> None: if read_line == write_line: time.sleep(1) else: - while read_line < write_line: + while read_line < write_line and len(result) == 0: if searchString in out_lines[read_line]: result = out_lines[:read_line] - break read_line+=1 retry += 1 self._readLine = read_line diff --git a/tests/audioAmp_test.py b/tests/audioAmp_test.py new file mode 100644 index 0000000..a0a6813 --- /dev/null +++ b/tests/audioAmp_test.py @@ -0,0 +1,94 @@ + +import os +import sys +import json + +# Add the framework path to system +dir_path = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(dir_path+"/../") + +from framework.core.logModule import logModule +from framework.core.audioAmplifierController import audioAmplifierController + +if __name__ == "__main__": + + LOG = logModule("audio amplifier test", logModule.DEBUG) + CONFIGS = [ + { + 'type': 'denon', + 'host': '' # Needs to be be filled out with IP address + }] + + for config in CONFIGS: + + LOG.setFilename(os.path.abspath('./logs/'),'audioAmplifier-%sTest.log' % config.get('type')) + LOG.stepStart('Testing with %s' % json.dumps(config)) + + controller = audioAmplifierController(LOG, config) + + # Power ON test + try: + controller.power_on() + power = controller.get_power() + assert power == "ON" + print("PASSED: Power ON") + except: + print("FAILED: Power ON") + + # Volume test + try: + volume = -51 + controller.set_volume(volume) + updated_volume = controller.get_volume() + assert updated_volume == volume + print("PASSED: Volume change") + except: + print(f"FAILED: Volume change. Expected: {volume}, actual {updated_volume}") + + # Mute test + try: + controller.mute(True) + muted = controller.is_muted() + assert muted is True + print("PASSED: Mute") + except: + print("FAILED: Mute") + + try: + controller.mute(False) + muted = controller.is_muted() + assert muted is False + print("PASSED: Unmute") + except: + print("FAILED: Unmute") + + # Input test + try: + print("Available inputs:", controller.list_inputs()) + input = "AUX2" + controller.set_input(input) + updated_input = controller.get_input() + assert updated_input == input + print("PASSED: Input source set correctly.") + except: + print(f"FAILED: Input source not set correctly. Expected: {input}, actual: {updated_input}") + + # Sound Mode test + try: + print("Available sound modes:", controller.list_sound_modes()) + sound_mode = "MOVIE" + controller.set_sound_mode(sound_mode) + updated_sound_mode = controller.get_sound_mode() + assert updated_sound_mode == sound_mode + print("PASSED: Sound mode set correctly") + except: + print(f"FAILED: Sound mode not set correctly. Expected: {sound_mode}, actual: {updated_sound_mode}") + + # Power OFF test + try: + controller.power_off() + power = controller.get_power() + assert power.upper() == "OFF" + print("PASSED: Power OFF") + except: + print("FAILED: Power OFF")