From 053bffe4780b12d0ba9e6e718c87acaa4fce5ffa Mon Sep 17 00:00:00 2001 From: bhanucbp <141142298+bhanucbp@users.noreply.github.com> Date: Mon, 24 Nov 2025 18:11:32 +0000 Subject: [PATCH 1/4] gh#197: Add Virtual HDMI-CEC client - Add virtual HDMI-CEC client - Add utcontroller class --- examples/configs/example_rack_config.yml | 5 +- framework/core/hdmiCECController.py | 16 ++ .../hdmicecModules/abstractCECController.py | 2 +- ...ec_print_device_network_configuration.yaml | 24 ++ .../hdmicecModules/virtualCECController.py | 211 ++++++++++++++++++ framework/core/utPlaneController.py | 109 +++++++++ 6 files changed, 364 insertions(+), 3 deletions(-) create mode 100644 framework/core/hdmicecModules/configuration/virtual_cec_print_device_network_configuration.yaml create mode 100644 framework/core/hdmicecModules/virtualCECController.py create mode 100644 framework/core/utPlaneController.py diff --git a/examples/configs/example_rack_config.yml b/examples/configs/example_rack_config.yml index 5fb433f..4b94cb4 100644 --- a/examples/configs/example_rack_config.yml +++ b/examples/configs/example_rack_config.yml @@ -91,7 +91,7 @@ rackConfig: # [type: "tapo", ip: "", username: "", password: "", outlet: "optional"] # [type: "hs100", ip:"", port:"optional" ] kara also supports hs100 # [type: "apc", ip:"", username:"", password:"" ] rack apc switch - # [type: "olimex", ip:"", port:"optional", relay:"" ] + # [type: "olimex", ip:"", port:"optional", relay:"" ] # [type: "SLP", ip:"", username: "", password: "", outlet_id:"", port:"optional"] # [type: "none" ] if section doesn't exist then type:none will be used @@ -99,7 +99,8 @@ rackConfig: # supported types: # [type: "cec-client", adaptor: "/dev/ttycec"] # [type: "remote-cec-client", adaptor: "/dev/ttycec", address: "192.168.99.1", username(optional): "testuser", password(optional): "testpswd", port(optional): "22"] - + # [type: "virtual-cec-client", device_type: ["TV", "RecordingDevice", "Tuner", "PlaybackDevice", "AudioSystem", "VideoProcessor", "PureCecSwitch", "Reserved"] + # address: "127.0.0.1", username: "testuser", password: "testpswd", port: "5522", control_port: 8080, device_network_configuration: "path to device network configuration file" ] # [ avSyncController: optional] - Specifiec AVSyncController for the slot # supported types: # [type: "SyncOne2", port: "/dev/ttyACM0", extended_mode (optional): true|false, audio_input (optional): "AUTO|EXTERNAL|INTERNAL", speaker_distance (optional): "1.5"] diff --git a/framework/core/hdmiCECController.py b/framework/core/hdmiCECController.py index d16a0ea..1bcbbe2 100644 --- a/framework/core/hdmiCECController.py +++ b/framework/core/hdmiCECController.py @@ -39,6 +39,7 @@ from framework.core.logModule import logModule from framework.core.streamToFile import StreamToFile from framework.core.hdmicecModules import CECClientController, RemoteCECClient, CECDeviceType +from framework.core.hdmicecModules.virtualCECController import virtualCECController class HDMICECController(): """ @@ -72,6 +73,17 @@ def __init__(self, log: logModule, config: dict): password=config.get('password',''), port=config.get('port',22), prompt=config.get('prompt', ':~')) + elif self.controllerType.lower() == 'virtual-cec-client': + self.controller = virtualCECController(self.cecAdaptor, + self._log, + self._stream, + address=config.get('address'), + username=config.get('username',''), + password=config.get('password',''), + port=config.get('port',22), + prompt=config.get('prompt', '~#'), + device_configuration=config.get('device_network_configuration',''), + control_port=config.get('control_port', 8080)) self._read_line = 0 def sendMessage(self, sourceAddress: str, destAddress: str, opCode: str, payload: list = None) -> None: @@ -106,6 +118,10 @@ def checkMessageReceived(self, sourceAddress: str, destAddress: str, opCode: str Returns: boolean: True if message is received. False otherwise. """ + if self.controllerType.lower() == 'virtual-cec-client': + self._log.debug('checkMessageReceived is mock implementation for virtual-cec-client controller. Defaulting to True.') + return True + result = False payload_string = '' if isinstance(payload, list): diff --git a/framework/core/hdmicecModules/abstractCECController.py b/framework/core/hdmicecModules/abstractCECController.py index 89e5cf3..39f0f04 100644 --- a/framework/core/hdmicecModules/abstractCECController.py +++ b/framework/core/hdmicecModules/abstractCECController.py @@ -49,7 +49,7 @@ def __init__(self, adaptor_path:str, logger:logModule, streamLogger: StreamToFil def sendMessage(cls, sourceAddress: str, destAddress: str, opCode: str, payload: list = None, deviceType: CECDeviceType=None) -> None: """ Sends an opCode from a specified source and to a specified destination. - + Args: sourceAddress (str): The logical address of the source device (0-9 or A-F). destAddress (str): The logical address of the destination device (0-9 or A-F). diff --git a/framework/core/hdmicecModules/configuration/virtual_cec_print_device_network_configuration.yaml b/framework/core/hdmicecModules/configuration/virtual_cec_print_device_network_configuration.yaml new file mode 100644 index 0000000..c5a2d5e --- /dev/null +++ b/framework/core/hdmicecModules/configuration/virtual_cec_print_device_network_configuration.yaml @@ -0,0 +1,24 @@ +#** ***************************************************************************** +# * +# * If not stated otherwise in this file or this component's LICENSE file the +# * following copyright and licenses apply: +# * +# * Copyright 2025 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. +# * +#** ****************************************************************************** +HdmiCec: + command: print + description: Print the current HDMI CEC device map \ No newline at end of file diff --git a/framework/core/hdmicecModules/virtualCECController.py b/framework/core/hdmicecModules/virtualCECController.py new file mode 100644 index 0000000..eed2150 --- /dev/null +++ b/framework/core/hdmicecModules/virtualCECController.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * If not stated otherwise in this file or this component's LICENSE file the +# * following copyright and licenses apply: +# * +# * Copyright 2024 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. +# * +#* ****************************************************************************** + +import os +import sys +import time +import re +import yaml + +dir_path = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(dir_path) +sys.path.append(os.path.join(dir_path, "../")) + +from framework.core.logModule import logModule +from framework.core.streamToFile import StreamToFile +from utPlaneController import utPlaneController +from .abstractCECController import CECInterface +from framework.core.commandModules.sshConsole import sshConsole + +HDMICEC_DEVICE_LIST_FILE = "/tmp/hdmi_cec_device_list_info.txt" +HDMICEC_PRINT_CEC_NETWORK_CONFIG_FILE = os.path.join(dir_path, "configuration", "virtual_cec_print_device_network_configuration.yaml") + +class virtualCECController(CECInterface): + """ + HDMI CEC related utility functions + """ + def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, + address: str, username: str = '', password: str = '', port: int = 22, prompt = '~#', + device_configuration:str='', control_port:int=8080): + """ + Initializes the virtualCECController class for HDMI CEC device communication. + + Args: + adaptor (str): The adaptor type/name for the parent class. + logger (logModule): Logger module instance for logging operations. + streamLogger (StreamToFile): Stream logger for file-based logging. + address (str): IP address or hostname of the remote device. + username (str, optional): SSH username for authentication. Defaults to ''. + password (str, optional): SSH password for authentication. Defaults to ''. + port (int, optional): SSH port number for connection. Defaults to 22. + prompt (str, optional): Command prompt string for the SSH session. Defaults to '~#'. + device_configuration (str, optional): Path to the HDMI CEC device network configuration YAML file. Defaults to ''. + control_port (int, optional): Port number for ut-controller communication. Defaults to 8080. + + """ + super().__init__(adaptor, logger, streamLogger) + + _console = sshConsole(self._log, address, username, password, port=port, prompt=prompt) + + self.control_port = control_port + self.session = _console + self.commandPrompt = prompt + + # Load the HDMI CEC device network configuration file + with open(device_configuration, "r") as f: + config_dict = yaml.safe_load(f) + + self.cecDeviceNetworkConfigString = yaml.dump(config_dict) + self.cecDeviceNetworkConfigString = self.cecDeviceNetworkConfigString.replace('"', '\\"') + + # Load the print configuration file + with open(HDMICEC_PRINT_CEC_NETWORK_CONFIG_FILE, "r") as f: + print_dict = yaml.safe_load(f) + + self.printConfigString = yaml.dump(print_dict) + self.printConfigString = self.printConfigString.replace('"', '\\"') + + self.utPlaneController = utPlaneController(self.session, port=self.control_port) + + def loadCecDeviceNetworkConfiguration(self, configString: str): + """ + Loads the HDMI CEC device network configuration file on to the vComponent. + """ + + self.utPlaneController.sendMessage(configString) + + def readDeviceNetworkList(self) -> list: + """ + Reads the device network list from the HDMI CEC device. + + Returns: + list: A list of dictionaries representing discovered devices with details. + """ + result = self.session.read_until(self.commandPrompt) + + result = re.sub(r'\x1b\[[0-9;]*m', '', result) # remove ANSI color codes + result = result.replace('\r', '') # normalize newlines + result = re.sub(r'root@[\w\-\:\/# ]+', '', result) # remove shell prompt lines + result = re.sub(r'curl:.*?\n', '', result, flags=re.DOTALL) # remove curl noise if any + + devices = [] + + # Regex to match device lines + pattern = re.compile( + r"- Name:\s*(?P[^,]+),.*?" + r"Active Source:\s*(?P\d+),.*?" + r"Logical-1:\s*(?P-?\d+),.*?" + r"Physical:\s*(?P[\d\.]+)", + re.MULTILINE + ) + + for match in pattern.finditer(result): + devices.append({ + "name": match.group("name").strip(), + "physical address": match.group("physical"), + "logical address": int(match.group("logical1")), + "active source": int(match.group("active")), + }) + + return devices + + def listDevices(self) -> list: + """ + Lists the devices currently available on the HDMI CEC network. + + Returns: + list: A list of dictionaries representing discovered devices with details. + { + "name": "name", + "physical address": "0.0.0.0", + "logical address": 0, + "active source": 0, + } + """ + # send command to CEC network to print device configuration + self.utPlaneController.sendMessage(self.printConfigString) + + devices = self.readDeviceNetworkList() + + if devices is None or len(devices) == 0: + self.session.write("cat " + HDMICEC_DEVICE_LIST_FILE) + time.sleep(2) + devices = self.readDeviceNetworkList() + + return devices + + def checkMessageReceived(self, cecMessage:dict, timeout: int = 10) -> bool: + """ + This function checks to see if a specified opCode has been received. + + Args: + cecMessage (dict): A dictionary containing the following keys: + sourceAddress (int): The logical address of the source device (0-15). + destAddress (int): The logical address of the destination device (0-15). + opCode (str): Operation code to check as an hexadecimal string e.g 0x81. + payload (list): List of hexadecimal strings to be checked with the opCode. + timeout (int): The maximum amount of time, in seconds, that the method will + wait for the message to be received. Defaults to 10. + Returns: + bool: True if the message was received, False otherwise. + """ + # Note: for vComponent, this function always returns True as we cannot verify message receipt + return True + + def sendMessage(self, sourceAddress: int, destAddress: int, opCode: str, payload: list = None) -> bool: + """ + This function sends a specified opCode. + + Args: + sourceAddress (int): The logical address of the source device (0-15). + destAddress (int): The logical address of the destination device (0-15). + opCode (str): Operation code to send as an hexadecimal string e.g 0x81. + payload (list): List of hexadecimal strings to be sent with the opCode. Optional. + + Returns: + bool: True if the message was sent successfully, False otherwise. + """ + # Format the payload: source, destination, opCode, and payload + msg_payload = [f"0x{sourceAddress}{destAddress}", opCode] + if payload: + msg_payload.extend(payload) + + yaml_content = ( + "HdmiCec:\n" + " command: cec_message\n" + " description: Send a CEC message\n" + " message:\n" + " user_defined: true\n" + f" payload: {msg_payload}\n" + ) + + # Send the command to ut-controller + self.utPlaneController.sendMessage(yaml_content) + + return True + + def start(self): + self.loadCecDeviceNetworkConfiguration(self.cecDeviceNetworkConfigString) + + def stop(self): + pass diff --git a/framework/core/utPlaneController.py b/framework/core/utPlaneController.py new file mode 100644 index 0000000..cfd667c --- /dev/null +++ b/framework/core/utPlaneController.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +#** ***************************************************************************** +# * +# * If not stated otherwise in this file or this component's LICENSE file the +# * following copyright and licenses apply: +# * +# * Copyright 2025 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. +# * +#* ****************************************************************************** + +import os +import sys + +dir_path = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(dir_path) +sys.path.append(os.path.join(dir_path, "..", "..", "..")) + +from framework.core.logModule import logModule + +class utPlaneController(): + """ + UT Plane Controller class for managing communication with the ut-controller. + + This class provides an interface to interact with the ut-controller running on a device. + It facilitates sending commands and YAML configuration files to the controller via HTTP requests + using curl commands. The controller operates over a configurable port (default: 8080) and uses + the session object to execute commands on the target device. + + Typical usage involves: + 1. Creating an instance with an active session and optional port/log configuration + 2. Preparing YAML files with test commands or configuration + 3. Sending these files to the ut-controller using sendMessage() + + Attributes: + session (object): Active session object for device communication + port (int): Port number where ut-controller service is listening (default: 8080) + log (logModule): Logger instance for recording controller activities and debugging + + Example: + >>> controller = utPlaneController(session, port=8080) + >>> controller.sendMessage("/path/on/dut/to/test_config.yaml") + >>> controller.sendMessage("yaml string directly") + """ + + def __init__(self, session:object, port: int = 8080, log:logModule=None): + """ + Initializes UT Plane Controller class. + + Args: + session (class): The session object to communicate with the device + port (int): The port number for the controller + log (class, optional): Parent log class. Defaults to None. + """ + self.log = log + if log is None: + self.log = logModule(self.__class__.__name__) + self.log.setLevel( self.log.INFO ) + + self.session = session + self.port = port + + def sendMessage(self, yamlInput: str) -> bool: + """ + Sends a command to the ut-controller via curl. + + Args: + yamlInput (str): Either a YAML string or path to a YAML file. + + Returns: + bool: True if the message was sent successfully, False otherwise. + """ + try: + # Validate input + if not yamlInput or not isinstance(yamlInput, str): + self.log.error("Invalid input provided") + return False + + # Check if input is a file path + if os.path.isfile(yamlInput): + # It's a file path - use --data-binary with file reference + yaml_content = yamlInput + cmd = f'curl -X POST -H "Content-Type: application/x-yaml" --data-binary @"{yaml_content}" "http://localhost:{self.port}/api/postKVP"' + else: + # It's a direct YAML string - escape quotes and send inline + yaml_content = yamlInput.replace('"', '\\"') + cmd = f'curl -X POST -H "Content-Type: application/x-yaml" --data-binary "{yaml_content}" "http://localhost:{self.port}/api/postKVP"' + + # Send command + self.session.write(cmd) + + self.log.info(f"Message sent successfully to ut-controller on port {self.port}") + return True + + except Exception as e: + self.log.error(f"Failed to send message to ut-controller: {str(e)}") + return False From b2a951ec455f14028f93f82f4167e5d9a7612a49 Mon Sep 17 00:00:00 2001 From: bhanucbp <141142298+bhanucbp@users.noreply.github.com> Date: Tue, 25 Nov 2025 14:17:42 +0000 Subject: [PATCH 2/4] gh #197: Add Virtual HDMI-CEC client - Addressed review comments --- examples/configs/example_rack_config.yml | 5 +- .../hdmicecModules/virtualCECController.py | 54 ++++++++----------- framework/core/utPlaneController.py | 16 ++++-- 3 files changed, 35 insertions(+), 40 deletions(-) diff --git a/examples/configs/example_rack_config.yml b/examples/configs/example_rack_config.yml index 4b94cb4..8e12cd2 100644 --- a/examples/configs/example_rack_config.yml +++ b/examples/configs/example_rack_config.yml @@ -95,12 +95,11 @@ rackConfig: # [type: "SLP", ip:"", username: "", password: "", outlet_id:"", port:"optional"] # [type: "none" ] if section doesn't exist then type:none will be used - # [ hdmiCECController: optional ] - Specific hdmiCECController for the slot + # [ hdmiCECController: optional ] - Specifies hdmiCECController for the slot # supported types: # [type: "cec-client", adaptor: "/dev/ttycec"] # [type: "remote-cec-client", adaptor: "/dev/ttycec", address: "192.168.99.1", username(optional): "testuser", password(optional): "testpswd", port(optional): "22"] - # [type: "virtual-cec-client", device_type: ["TV", "RecordingDevice", "Tuner", "PlaybackDevice", "AudioSystem", "VideoProcessor", "PureCecSwitch", "Reserved"] - # address: "127.0.0.1", username: "testuser", password: "testpswd", port: "5522", control_port: 8080, device_network_configuration: "path to device network configuration file" ] + # [type: "virtual-cec-client", address: "127.0.0.1", username: "testuser", password: "testpswd", port: "5522", control_port: 8080, device_network_configuration: "path to device network configuration file" ] # [ avSyncController: optional] - Specifiec AVSyncController for the slot # supported types: # [type: "SyncOne2", port: "/dev/ttyACM0", extended_mode (optional): true|false, audio_input (optional): "AUTO|EXTERNAL|INTERNAL", speaker_distance (optional): "1.5"] diff --git a/framework/core/hdmicecModules/virtualCECController.py b/framework/core/hdmicecModules/virtualCECController.py index eed2150..e6e18ad 100644 --- a/framework/core/hdmicecModules/virtualCECController.py +++ b/framework/core/hdmicecModules/virtualCECController.py @@ -33,7 +33,7 @@ from framework.core.logModule import logModule from framework.core.streamToFile import StreamToFile -from utPlaneController import utPlaneController +from framework.core.utPlaneController import utPlaneController from .abstractCECController import CECInterface from framework.core.commandModules.sshConsole import sshConsole @@ -46,7 +46,7 @@ class virtualCECController(CECInterface): """ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, address: str, username: str = '', password: str = '', port: int = 22, prompt = '~#', - device_configuration:str='', control_port:int=8080): + device_configuration:str = '', control_port:int = 8080): """ Initializes the virtualCECController class for HDMI CEC device communication. @@ -71,21 +71,29 @@ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, self.session = _console self.commandPrompt = prompt - # Load the HDMI CEC device network configuration file - with open(device_configuration, "r") as f: - config_dict = yaml.safe_load(f) + try: + # Load the HDMI CEC device network configuration file + with open(device_configuration, "r") as f: + config_dict = yaml.safe_load(f) - self.cecDeviceNetworkConfigString = yaml.dump(config_dict) - self.cecDeviceNetworkConfigString = self.cecDeviceNetworkConfigString.replace('"', '\\"') + self.cecDeviceNetworkConfigString = yaml.dump(config_dict) - # Load the print configuration file - with open(HDMICEC_PRINT_CEC_NETWORK_CONFIG_FILE, "r") as f: - print_dict = yaml.safe_load(f) + # Load the print configuration file + with open(HDMICEC_PRINT_CEC_NETWORK_CONFIG_FILE, "r") as f: + print_dict = yaml.safe_load(f) - self.printConfigString = yaml.dump(print_dict) - self.printConfigString = self.printConfigString.replace('"', '\\"') + self.printConfigString = yaml.dump(print_dict) - self.utPlaneController = utPlaneController(self.session, port=self.control_port) + self.utPlaneController = utPlaneController(self.session, port=self.control_port) + except FileNotFoundError: + self._log.critical(f"Device config file not found") + raise + except yaml.YAMLError as e: + self._log.critical(f"Invalid YAML in device config file: {e}") + raise + except Exception as e: + self._log.critical(f"Failed to load device configuration: {e}") + raise def loadCecDeviceNetworkConfiguration(self, configString: str): """ @@ -154,24 +162,6 @@ def listDevices(self) -> list: return devices - def checkMessageReceived(self, cecMessage:dict, timeout: int = 10) -> bool: - """ - This function checks to see if a specified opCode has been received. - - Args: - cecMessage (dict): A dictionary containing the following keys: - sourceAddress (int): The logical address of the source device (0-15). - destAddress (int): The logical address of the destination device (0-15). - opCode (str): Operation code to check as an hexadecimal string e.g 0x81. - payload (list): List of hexadecimal strings to be checked with the opCode. - timeout (int): The maximum amount of time, in seconds, that the method will - wait for the message to be received. Defaults to 10. - Returns: - bool: True if the message was received, False otherwise. - """ - # Note: for vComponent, this function always returns True as we cannot verify message receipt - return True - def sendMessage(self, sourceAddress: int, destAddress: int, opCode: str, payload: list = None) -> bool: """ This function sends a specified opCode. @@ -179,7 +169,7 @@ def sendMessage(self, sourceAddress: int, destAddress: int, opCode: str, payload Args: sourceAddress (int): The logical address of the source device (0-15). destAddress (int): The logical address of the destination device (0-15). - opCode (str): Operation code to send as an hexadecimal string e.g 0x81. + opCode (str): Operation code to send as a hexadecimal string e.g 0x81. payload (list): List of hexadecimal strings to be sent with the opCode. Optional. Returns: diff --git a/framework/core/utPlaneController.py b/framework/core/utPlaneController.py index cfd667c..120b5f1 100644 --- a/framework/core/utPlaneController.py +++ b/framework/core/utPlaneController.py @@ -23,6 +23,7 @@ import os import sys +import yaml dir_path = os.path.dirname(os.path.realpath(__file__)) sys.path.append(dir_path) @@ -55,7 +56,7 @@ class utPlaneController(): >>> controller.sendMessage("yaml string directly") """ - def __init__(self, session:object, port: int = 8080, log:logModule=None): + def __init__(self, session:object, port: int = 8080, log: logModule = None): """ Initializes UT Plane Controller class. @@ -64,6 +65,11 @@ def __init__(self, session:object, port: int = 8080, log:logModule=None): port (int): The port number for the controller log (class, optional): Parent log class. Defaults to None. """ + + # Validate session + if session is None: + raise ValueError("session cannot be None") + self.log = log if log is None: self.log = logModule(self.__class__.__name__) @@ -72,12 +78,13 @@ def __init__(self, session:object, port: int = 8080, log:logModule=None): self.session = session self.port = port - def sendMessage(self, yamlInput: str) -> bool: + def sendMessage(self, yamlInput: str, isFile: bool = False) -> bool: """ Sends a command to the ut-controller via curl. Args: yamlInput (str): Either a YAML string or path to a YAML file. + isFile (bool): Flag indicating if yamlInput is a file path. Defaults to False. Returns: bool: True if the message was sent successfully, False otherwise. @@ -88,9 +95,8 @@ def sendMessage(self, yamlInput: str) -> bool: self.log.error("Invalid input provided") return False - # Check if input is a file path - if os.path.isfile(yamlInput): - # It's a file path - use --data-binary with file reference + if isFile: + # It's a file path on target device - use --data-binary with file reference yaml_content = yamlInput cmd = f'curl -X POST -H "Content-Type: application/x-yaml" --data-binary @"{yaml_content}" "http://localhost:{self.port}/api/postKVP"' else: From e34e8a364b824e12ba9c7767553e99a2ddf6ac49 Mon Sep 17 00:00:00 2001 From: bhanucbp <141142298+bhanucbp@users.noreply.github.com> Date: Wed, 26 Nov 2025 14:47:52 +0000 Subject: [PATCH 3/4] gh #197: Add Virtual HDMI-CEC client - Addressed review comments --- framework/core/hdmiCECController.py | 2 ++ framework/core/hdmicecModules/virtualCECController.py | 5 ++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/framework/core/hdmiCECController.py b/framework/core/hdmiCECController.py index 1bcbbe2..e8e5cbd 100644 --- a/framework/core/hdmiCECController.py +++ b/framework/core/hdmiCECController.py @@ -119,6 +119,8 @@ def checkMessageReceived(self, sourceAddress: str, destAddress: str, opCode: str boolean: True if message is received. False otherwise. """ if self.controllerType.lower() == 'virtual-cec-client': + # For a virtual component, the device network is only a simulation, so the controller + # does not generate any logs when messages are sent or received. Therefore, it always returns True. self._log.debug('checkMessageReceived is mock implementation for virtual-cec-client controller. Defaulting to True.') return True diff --git a/framework/core/hdmicecModules/virtualCECController.py b/framework/core/hdmicecModules/virtualCECController.py index e6e18ad..f5e2803 100644 --- a/framework/core/hdmicecModules/virtualCECController.py +++ b/framework/core/hdmicecModules/virtualCECController.py @@ -65,10 +65,7 @@ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, """ super().__init__(adaptor, logger, streamLogger) - _console = sshConsole(self._log, address, username, password, port=port, prompt=prompt) - self.control_port = control_port - self.session = _console self.commandPrompt = prompt try: @@ -84,6 +81,8 @@ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, self.printConfigString = yaml.dump(print_dict) + self.session = sshConsole(self._log, address, username, password, port=port, prompt=prompt) + self.utPlaneController = utPlaneController(self.session, port=self.control_port) except FileNotFoundError: self._log.critical(f"Device config file not found") From 6e51ea2e5f46a53469925df67e58d428fd361df9 Mon Sep 17 00:00:00 2001 From: bhanucbp <141142298+bhanucbp@users.noreply.github.com> Date: Wed, 26 Nov 2025 16:53:54 +0000 Subject: [PATCH 4/4] gh #197: Add Virtual HDMI-CEC client - Addressed review comments --- framework/core/hdmiCECController.py | 6 --- .../hdmicecModules/virtualCECController.py | 42 ++++++++++++++----- framework/core/utPlaneController.py | 7 ++-- 3 files changed, 36 insertions(+), 19 deletions(-) diff --git a/framework/core/hdmiCECController.py b/framework/core/hdmiCECController.py index e8e5cbd..9834b78 100644 --- a/framework/core/hdmiCECController.py +++ b/framework/core/hdmiCECController.py @@ -118,12 +118,6 @@ def checkMessageReceived(self, sourceAddress: str, destAddress: str, opCode: str Returns: boolean: True if message is received. False otherwise. """ - if self.controllerType.lower() == 'virtual-cec-client': - # For a virtual component, the device network is only a simulation, so the controller - # does not generate any logs when messages are sent or received. Therefore, it always returns True. - self._log.debug('checkMessageReceived is mock implementation for virtual-cec-client controller. Defaulting to True.') - return True - result = False payload_string = '' if isinstance(payload, list): diff --git a/framework/core/hdmicecModules/virtualCECController.py b/framework/core/hdmicecModules/virtualCECController.py index f5e2803..1f57ac7 100644 --- a/framework/core/hdmicecModules/virtualCECController.py +++ b/framework/core/hdmicecModules/virtualCECController.py @@ -4,7 +4,7 @@ # * If not stated otherwise in this file or this component's LICENSE file the # * following copyright and licenses apply: # * -# * Copyright 2024 RDK Management +# * Copyright 2025 RDK Management # * # * Licensed under the Apache License, Version 2.0 (the "License"); # * you may not use this file except in compliance with the License. @@ -39,6 +39,7 @@ HDMICEC_DEVICE_LIST_FILE = "/tmp/hdmi_cec_device_list_info.txt" HDMICEC_PRINT_CEC_NETWORK_CONFIG_FILE = os.path.join(dir_path, "configuration", "virtual_cec_print_device_network_configuration.yaml") +HOST_CMD_EXEC_TIMEOUT = 2 class virtualCECController(CECInterface): """ @@ -51,7 +52,7 @@ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, Initializes the virtualCECController class for HDMI CEC device communication. Args: - adaptor (str): The adaptor type/name for the parent class. + adaptor (str): The adaptor file path for the parent class. logger (logModule): Logger module instance for logging operations. streamLogger (StreamToFile): Stream logger for file-based logging. address (str): IP address or hostname of the remote device. @@ -156,18 +157,18 @@ def listDevices(self) -> list: if devices is None or len(devices) == 0: self.session.write("cat " + HDMICEC_DEVICE_LIST_FILE) - time.sleep(2) + time.sleep(HOST_CMD_EXEC_TIMEOUT) devices = self.readDeviceNetworkList() return devices - def sendMessage(self, sourceAddress: int, destAddress: int, opCode: str, payload: list = None) -> bool: + def sendMessage(self, sourceAddress: str, destAddress: str, opCode: str, payload: list = None) -> bool: """ This function sends a specified opCode. Args: - sourceAddress (int): The logical address of the source device (0-15). - destAddress (int): The logical address of the destination device (0-15). + sourceAddress (str): The logical address of the source device ('0'-'F'). + destAddress (str): The logical address of the destination device ('0'-'F'). opCode (str): Operation code to send as a hexadecimal string e.g 0x81. payload (list): List of hexadecimal strings to be sent with the opCode. Optional. @@ -188,13 +189,34 @@ def sendMessage(self, sourceAddress: int, destAddress: int, opCode: str, payload f" payload: {msg_payload}\n" ) - # Send the command to ut-controller - self.utPlaneController.sendMessage(yaml_content) - - return True + try: + result = self.utPlaneController.sendMessage(yaml_content) + return bool(result) + except Exception as e: + self._log.critical(f"Failed to send CEC message: {e}") + return False def start(self): self.loadCecDeviceNetworkConfiguration(self.cecDeviceNetworkConfigString) def stop(self): pass + + def receiveMessage(self,sourceAddress: str, destAddress: str, opCode: str, timeout: int = 10, payload: list = None) -> bool: + """ + For a virtual component, the device network is only a simulation, so the controller + does not generate any logs when messages are sent or received. Therefore, it always returns expected message. + + Args: + sourceAddress (str): The logical address of the source device (0-9 or A-F). + destAddress (str): The logical address of the destination device (0-9 or A-F). + opCode (str): Operation code to send as an hexidecimal string e.g 0x81. + timeout (int): The maximum amount of time, in seconds, that the method will + wait for the message to be received. Defaults to 10. + payload (list): List of hexidecimal strings to be sent with the opCode. Optional. + + Returns: + list: list of strings containing found message. + """ + message = "Received expected message" + return message diff --git a/framework/core/utPlaneController.py b/framework/core/utPlaneController.py index 120b5f1..386d044 100644 --- a/framework/core/utPlaneController.py +++ b/framework/core/utPlaneController.py @@ -23,7 +23,6 @@ import os import sys -import yaml dir_path = os.path.dirname(os.path.realpath(__file__)) sys.path.append(dir_path) @@ -104,11 +103,13 @@ def sendMessage(self, yamlInput: str, isFile: bool = False) -> bool: yaml_content = yamlInput.replace('"', '\\"') cmd = f'curl -X POST -H "Content-Type: application/x-yaml" --data-binary "{yaml_content}" "http://localhost:{self.port}/api/postKVP"' + result = True + # Send command - self.session.write(cmd) + result = self.session.write(cmd) self.log.info(f"Message sent successfully to ut-controller on port {self.port}") - return True + return result except Exception as e: self.log.error(f"Failed to send message to ut-controller: {str(e)}")