GH166 - audio amp

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

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(),
+        }

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()

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")