Update #103: Creation of remote hdmicec client

examples/configs/example_rack_config.yml

@@ -94,6 +94,7 @@ rackConfig:
                     # [ hdmiCECController: optional ] - Specific 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"]
                 - pi2:
                     ip: "192.168.99.1"
                     description: "local pi4"

framework/core/hdmiCECController.py

@@ -37,7 +37,7 @@
 MY_DIR = path.dirname(MY_PATH)
 sys.path.append(path.join(MY_DIR,'../../'))
 from framework.core.logModule import logModule
-from hdmicecModules import CECClientController, MonitoringType
+from framework.core.hdmicecModules import CECClientController, RemoteCECClient, MonitoringType
 
 class HDMICECController():
     """
@@ -58,8 +58,15 @@ def __init__(self, log: logModule, config: dict):
         self.cecAdaptor = config.get('adaptor')
         if self.controllerType.lower() == 'cec-client':
             self.controller = CECClientController(self.cecAdaptor, self._log)
+        elif self.controllerType.lower() == 'remote-cec-client':
+            self.controller = RemoteCECClient(self.cecAdaptor,
+                                              self._log,
+                                              address=config.get('address'),
+                                              username=config.get('username',''),
+                                              password=config.get('password',''),
+                                              port=config.get('port',22))
         self._read_line = 0
-        self._monitoringLog = path.join(self._log.logPath, 'cecMonitor.log')
+        self._monitoringLog = path.abspath(path.join(self._log.logPath, 'cecMonitor.log'))
 
     def send_message(self, message: str) -> bool:
         """
@@ -148,11 +155,18 @@ def listDevices(self) -> list:
     CONFIGS = [
         {
             'type': 'cec-client',
-            'adaptor': '/dev/ttyACM0'
-            },
+            'adaptor': '/dev/ttyACM0' # This is default for pulse 8
+        },
+        {
+            'type': 'remote-cec-client',
+            'adaptor': '/dev/cec0', # This is default for Raspberry Pi
+            'address': '', # Needs to be be filled out with IP address
+            'username': '', # Needs to be filled out with login username
+            'password': '' # Needs to be filled out with login password
+        }
     ]
     for config in CONFIGS:
-        LOG.setFilename('./logs/','CECTEST%s.log' % config.get('type'))
+        LOG.setFilename(path.abspath('./logs/'),'CECTEST%s.log' % config.get('type'))
         LOG.stepStart('Testing with %s' % json.dumps(config))
         CEC = HDMICECController(LOG, config)
         DEVICES = CEC.listDevices()

framework/core/hdmicecModules/__init__.py

@@ -27,4 +27,5 @@
 #* ******************************************************************************
 
 from .cecClient import CECClientController
+from .remoteCECClient import RemoteCECClient
 from .cecTypes import MonitoringType

framework/core/hdmicecModules/abstractCECController.py

@@ -40,6 +40,7 @@ def __init__(self, adaptor_path:str, logger:logModule):
         self.adaptor = adaptor_path
         self._log = logger
         self._monitoring = False
+        self._monitoring_log = None
 
     @property
     def monitoring(self) -> bool:

framework/core/hdmicecModules/cecClient.py

@@ -59,8 +59,7 @@ def __init__(self, adaptor_path:str, logger:logModule):
             AttributeError: If the specified CEC adaptor is not found.
         """
 
-        self._log = logger
-        self.adaptor = adaptor_path
+        super().__init__(adaptor_path=adaptor_path, logger=logger)
         self._log.debug('Initialising CECClientController for [%s]' % self.adaptor)
         if self.adaptor not in map(lambda x: x.get('com port'),self._getAdaptors()):
             raise AttributeError('CEC Adaptor specified not found')
@@ -163,41 +162,25 @@ def _splitDeviceSectionsToDicts(self,command_output:str) -> list:
 
     def startMonitoring(self, monitoringLog: str, device_type: MonitoringType = MonitoringType.RECORDER) -> None:
         self._monitoring = True
+        self._monitoring_log = monitoringLog
         try:
             self._m_proc = subprocess.Popen(f'cec-client {self.adaptor} -m -d 0 -t {device_type.value}'.split(),
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.STDOUT,
                                 text=True)
-            self._m_stdout_thread = Thread(target=self._write_monitoring_log,
-                                        args=[self._m_proc.stdout, monitoringLog],
-                                        daemon=True)
-            self._m_stdout_thread.start()
+            self._log.logStreamToFile(self._m_proc.stdout, self._monitoring_log)
         except Exception as e:
             self.stopMonitoring()
             raise
 
-    def _write_monitoring_log(self,streamIn: IOBase, logFilePath: str) -> None:
-        """
-        Writes the output of the monitoring process to a log file.
-
-        Args:
-            stream_in (IOBase): The input stream from the monitoring process.
-            logFilePath (str): File path to write the monitoring log out to.
-        """
-        while True:
-            chunk = streamIn.readline()
-            if chunk == '':
-                break
-            with open(logFilePath, 'a+',) as out:
-                out.write(chunk)
-
     def stopMonitoring(self) -> None:
         self._log.debug('Stopping monitoring of adaptor [%s]' % self.adaptor)
         if self.monitoring is False:
             return
         self._m_proc.terminate()
         exit_code = self._m_proc.wait()
-        self._m_stdout_thread.join()
+        self._log.stopStreamedLog(self._monitoring_log)
+        self._monitoring_log = None
         self._monitoring = False
 
     def __del__(self):

framework/core/hdmicecModules/remoteCECClient.py

@@ -0,0 +1,158 @@
+#!/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        : 02/10/2024
+#*   **
+#*   ** @brief : Abstract class for CEC controller types.
+#*   **
+#* ******************************************************************************
+
+import re
+
+from framework.core.logModule import logModule
+from framework.core.commandModules.sshConsole import sshConsole
+from .abstractCECController import CECInterface
+from .cecTypes import MonitoringType
+
+class RemoteCECClient(CECInterface):
+
+    def __init__(self, adaptor: str,logger: logModule, address: str, port: int = 22, username: str = '', password: str = ''):
+        super().__init__(adaptor, logger)
+        self._console = sshConsole(self._log,address, username, password, port=port)
+        self._log.debug('Initialising RemoteCECClient controller')
+        try:
+            self._console.open()
+        except:
+            self._log.critical('Could not open connection to RemoteCECClient controller')
+            raise
+        if self.adaptor not in map(lambda x: x.get('com port'),self._getAdaptors()):
+            raise AttributeError('CEC Adaptor specified not found')
+        self._monitoringLog = None
+
+    @property
+    def monitoring(self) -> bool:
+        return self._monitoring
+
+    def _getAdaptors(self) -> list:
+        """
+        Retrieves a list of available CEC adaptors using `cec-client`.
+
+        Returns:
+            list: A list of dictionaries representing available adaptors with details like COM port.
+        """
+        self._console.write(f'cec-client -l')
+        stdout = self._console.read_until('currently active source')
+        stdout = stdout.replace('\r\n','\n')
+        adaptor_count = re.search(r'Found devices: ([0-9]+)',stdout, re.M).group(1)
+        adaptors = self._splitDeviceSectionsToDicts(stdout)
+        return adaptors
+
+    def sendMessage(self, message:str) -> bool:
+        """
+        Send a CEC message to the CEC network.
+
+        Args:
+            message (str): The CEC message to be sent.
+
+        Returns:
+            bool: True if the message was sent successfully, False otherwise.
+        """
+        return self._console.write(f'echo "{message}" | cec-client {self.adaptor}')
+
+    def listDevices(self) -> list:
+        """
+        List CEC devices on CEC network.
+
+        The list returned contains dicts in the following format:
+            {
+            'name': 'TV'
+            'address': '0.0.0.0',
+            'active source': True,
+            'vendor': 'Unknown',
+            'osd string': 'TV', 
+            'CEC version': '1.3a', 
+            'power status': 'on', 
+            'language': 'eng', 
+            }
+        Returns:
+            list: A list of dictionaries representing discovered devices.
+        """
+        self.sendMessage('scan')
+        output = self._console.read_until('currently active source')
+        devices = self._splitDeviceSectionsToDicts(output.replace('\r\n','\n'))
+        for device in devices:
+            device['name'] = device.get('osd string')
+            if device.get('active source') == 'yes':
+                device['active source'] = True
+            else:
+                device['active source'] = False
+        return devices
+
+    def startMonitoring(self, monitoringLog: str, deviceType: MonitoringType=MonitoringType.RECORDER) -> None:
+        """
+        Starts monitoring CEC messages with a specified device type.
+
+        Args:
+            deviceType (MonitoringType, optional): The type of device to monitor (default: MonitoringType.RECORDER).
+            monitoringLog (str) : Path to write the monitoring log out
+        """
+        self._monitoringLog = monitoringLog
+        self._console.write(f'cec-client -m -t{deviceType.value}')
+        self._console.shell.set_combine_stderr(True)
+        self._log.logStreamToFile(self._console.shell.makefile(), self._monitoringLog)
+        self._monitoring = True
+
+    def stopMonitoring(self) -> None:
+        """
+        Stops the CEC monitoring process.
+        """
+        if self.monitoring is False:
+            return
+        self._console.write('\x03')
+        self._log.stopStreamedLog(self._monitoringLog)
+
+    def _splitDeviceSectionsToDicts(self,command_output:str) -> list:
+            """
+            Splits the output of a `cec-client` command into individual device sections and parses them into dictionaries.
+
+            Args:
+                command_output (str): The output string from the `cec-client` command.
+
+            Returns:
+                list: A list of dictionaries, each representing a single CEC device with its attributes.
+            """
+            devices = []
+            device_sections = re.findall(r'^device[ #0-9]{0,}:[\s\S]+?(?:type|language): +[\S ]+$',
+                                        command_output,
+                                        re.M)
+            if device_sections:
+                for section in device_sections:
+                    device_dict = {}
+                    for line in section.split('\n'):
+                        line_split = re.search(r'^([\w #]+): +?(\S[\S ]{0,})$',line)
+                        if line_split:
+                            device_dict[line_split.group(1)] = line_split.group(2)
+                    devices.append(device_dict)
+            return devices

framework/core/logModule.py

@@ -31,7 +31,9 @@
 #* ******************************************************************************
 
 import os
+from io import IOBase
 import logging
+from threading import Thread
 import time
 import datetime
 #from datetime import datetime
@@ -110,6 +112,7 @@ def __init__(self, moduleName, level=INFO):
         self.path = None
         self.logFile = None
         self.csvLogFile = None
+        self._loggingThreads = {}
 
     def __del__(self):
         """Deletes the logger instance.
@@ -486,4 +489,57 @@ def stepResult(self, result, message):
         message = "[{}]: RESULT : [{}]: {}".format(self.stepNum,resultMessage, message)
         self.step("=====================Step End======================",showStepNumber=False)
         self.stepResultMessage(message)
-
+
+    def logStreamToFile(self, inputStream: IOBase, outFileName: str) -> None:
+        """
+        Starts a new thread to write the contents of an input stream to a file.
+
+        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.
+        """
+        outPath = path.join(self.logPath,outFileName)
+        if path.isabs(outFileName):
+            outPath = outFileName
+        newThread = Thread(target=self._writeLogFile,
+                                        args=[inputStream, outPath],
+                                        daemon=True)
+
+        self._loggingThreads.update({outFileName: newThread})
+        newThread.start()
+
+    def stopStreamedLog(self, outFileName: str) -> None:
+        """
+        Stops a previously started thread that is writing to a log file.
+
+        Args:
+            outFileName (str): The path of the output file associated with the thread to be stopped.
+
+        Raises:
+            AttributeError: If the specified thread cannot be found.
+        """
+        log_thread = self._loggingThreads.get(outFileName)
+        if log_thread:
+            log_thread.join(timeout=30)
+        else:
+            raise AttributeError(f'Could not find requested logging thread to stop. [{outFileName}]')
+
+    def _writeLogFile(self,streamIn: IOBase, logFilePath: str) -> None:
+        """
+        Writes the input stream to a log file.
+
+        Args:
+            stream_in (IOBase): The stream from a process.
+            logFilePath (str): File path to write the log out to.
+        """
+        while True:
+            chunk = streamIn.readline()
+            if chunk == '':
+                break
+            with open(logFilePath, 'a+',) as out:
+                out.write(chunk)
+
+    def __del__(self):
+        for thread in self._loggingThreads.values():
+            thread.join()