feat: configuration and auxiliaries can be started and used in any py…

…thon script

.github/dependabot.yml

@@ -7,7 +7,6 @@ updates:
       interval: "monthly"
     reviewers:
       - "sebastianpfischer"
-      - "BKaDamien"
       - "Pog3k"
       - "sebclrsn"
 

docs/getting_started/pykiso_as_simulator.rst

@@ -0,0 +1,56 @@
+.. _pykiso_as_simulator:
+
+** NEW ** Pykiso as a simulator
+-------------------------------
+
+Introduction
+~~~~~~~~~~~~
+
+Pykiso consists of two main components:
+
+* The testing framework
+* The test environment creation and management
+
+Together, these components enable embedded software engineers to efficiently test their software in a familiar manner.
+Pykiso's testing framework adopts an approach similar to popular embedded testing frameworks,
+such as Google Test, making it intuitive for engineers with experience in embedded systems.
+
+Over the years, we have identified two main groups of Pykiso users:
+
+* Those who embrace our opinionated approach to testing embedded software.
+* Those who appreciate the core principles and resources but prefer a different approach to testing.
+
+By decoupling the testing framework from the test environment creation,
+Pykiso now caters to both groups, offering flexibility while retaining the benefits of its robust structure.
+
+
+
+Workflow Overview
+~~~~~~~~~~~~~~~~~
+
+**Create a test environment**
+Begin by defining your test environment in a configuration file. (Refer to :ref:`basic_config_file` for more details.)
+
+**Strip the test suites section**
+If needed, you can remove the `Test Suites` section from the configuration file to simplify the setup.
+
+**Write your (test) script**
+Create a Python script, import the Pykiso library, import the test environment and use the auxiliaries to interact with the system under test.
+
+
+This workflow allows users to leverage Pykiso's testing capabilities while maintaining flexibility in how they define and manage their test environments.
+
+
+
+Example
+~~~~~~~
+
+Definition of the test environment:
+
+.. literalinclude:: ../../examples/next_pykiso2/pykiso_as_simulator/serial.yaml
+    :language: yaml
+
+Creation of the test script:
+
+.. literalinclude:: ../../examples/next_pykiso2/pykiso_as_simulator/serial_simulation.py
+    :language: python

docs/getting_started/user_guide.rst

@@ -4,3 +4,5 @@
 .. include:: basic_config_file.rst
 
 .. include:: basic_tests.rst
+
+.. include:: pykiso_as_simulator.rst

docs/whats_new/version_ongoing.rst

@@ -15,7 +15,16 @@ By default the CCPCanCan will use the trace size define in during the initialisa
 
 The log path is now initialise if set at None.
 
+
 Results can be exported to Xray
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 see :ref:`xray`
+
+pykiso modules detachment
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The pykiso modules are now detached from the main testing framework.
+This enable users to define their hw setup and load it in python. The auxiliaries
+can now be used in a more flexible way in python.
+See :ref:`pykiso_as_simulator` for more details.

examples/next_pykiso2/pykiso_as_simulator/serial.yaml

@@ -0,0 +1,12 @@
+auxiliaries:
+  com_aux_sender:
+    connectors:
+        com: loopback
+    type: pykiso.lib.auxiliaries.communication_auxiliary:CommunicationAuxiliary
+  com_aux_receiver:
+    connectors:
+        com: loopback
+    type: pykiso.lib.auxiliaries.communication_auxiliary:CommunicationAuxiliary
+connectors:
+  loopback:
+    type: pykiso.lib.connectors.cc_raw_loopback:CCLoopback

examples/next_pykiso2/pykiso_as_simulator/serial_simulation.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+
+# Import pykiso
+import pykiso
+
+# Load the test environment configuration
+pykiso.load_config(Path(__file__).parent.resolve() / "serial.yaml")
+
+# From the pykiso library, import the type of auxiliary you defined in the configuration
+# Here, it would be the CommunicationAuxiliary class
+from pykiso.lib.auxiliaries.communication_auxiliary import CommunicationAuxiliary
+
+
+def first_test():
+    """Ping-pong test between sender and receiver (with no context manager)
+    """
+    # Get the instance of the sender and receiver defined in the configuration
+    sender = CommunicationAuxiliary.get_instance('com_aux_sender')
+    receiver = CommunicationAuxiliary.get_instance('com_aux_receiver')
+    # Start the sender and receiver
+    sender.start()
+    receiver.start()
+    # Use the auxiliaries for my test
+    sender.send_message("Hello, World!")
+    assert receiver.receive_message(timeout_in_s = 2) == "Hello, World!"
+    print("Test passed!")
+    # Stop the sender and receiver
+    sender.stop()
+    receiver.stop()
+
+def second_test():
+    """Ping-pong test between sender and receiver (with context manager)
+    """
+    # Get the instance of the sender and receiver defined in the configuration
+    sender = CommunicationAuxiliary.get_instance('com_aux_sender')
+    receiver = CommunicationAuxiliary.get_instance('com_aux_receiver')
+    # Start the auxiliaries with a context manager
+    with sender as sender, receiver as receiver:
+        # Use the auxiliaries for my test
+        sender.send_message("Hello, World!")
+        assert receiver.receive_message(timeout_in_s = 2) == "Hello, World!"
+    print("Second test passed!")
+
+if __name__ == "__main__":
+    first_test()
+    second_test()

src/pykiso/__init__.py

@@ -45,3 +45,17 @@
 )
 
 logging_initializer.add_internal_log_levels()
+
+# Experimental - load configuration and create auxiliaries
+from .config_parser import parse_config
+from .test_setup.config_registry import ConfigRegistry
+
+
+def load_config(config_file: str):
+    """Enable any user to load a pykiso yaml file from any script
+
+    :param config_file: path to the pykiso yaml file
+
+    """
+    cfg = parse_config(config_file)
+    ConfigRegistry.register_aux_con(cfg)

src/pykiso/auxiliary.py

@@ -24,7 +24,9 @@
 import queue
 import threading
 from enum import Enum, unique
-from typing import Any, Callable, List, Optional
+from typing import Any, Callable, List, Optional, Self
+
+from pykiso.test_setup.config_registry import ConfigRegistry
 
 from .exceptions import AuxiliaryCreationError, AuxiliaryNotStarted
 from .logging_initializer import add_internal_log_levels, initialize_loggers
@@ -48,6 +50,15 @@ class AuxiliaryInterface(abc.ABC):
     for the reception and one for the transmmission.
     """
 
+    @classmethod
+    def get_instance(cls, name: str) -> Self:
+        """Experimental - Get an auxiliary instance by its name."""
+        auxiliary = ConfigRegistry.get_aux_by_alias(name)
+        # Verify if the auxiliary is of the right type
+        if not isinstance(auxiliary, cls):
+            raise ValueError(f"Requested auxiliary {name} is not of type {cls}")
+        return auxiliary
+
     def __init__(
         self,
         name: str = None,
@@ -201,7 +212,8 @@ def _start_tx_task(self) -> None:
 
         task_name = f"{self.name}_tx"
         log.internal_debug("start transmit task %s", task_name)
-        self.tx_thread = threading.Thread(name=task_name, target=self._transmit_task)
+        # Any created thread should disappear after main-thread exit
+        self.tx_thread = threading.Thread(name=task_name, target=self._transmit_task, daemon=True)
         self.tx_thread.start()
 
     def _start_rx_task(self) -> None:
@@ -212,7 +224,8 @@ def _start_rx_task(self) -> None:
         with self.rx_lock:
             task_name = f"{self.name}_rx"
             log.internal_debug("start reception task %s", task_name)
-            self.rx_thread = threading.Thread(name=task_name, target=self._reception_task)
+            # Any created thread should disappear after main-thread exit
+            self.rx_thread = threading.Thread(name=task_name, target=self._reception_task, daemon=True)
             self.rx_thread.start()
 
     def _stop_tx_task(self) -> None:
@@ -257,6 +270,21 @@ def stop(self) -> bool:
         """
         return self.delete_instance()
 
+    def __enter__(self) -> Self:
+        """Context manager entry point"""
+        if self.start():
+            return self
+        else:
+            raise AuxiliaryNotStarted(f"Failed to start auxiliary {self.name}")
+
+    def __exit__(self, type, value, traceback):
+        """Context manager exit point"""
+        stop_status = self.stop()
+        if traceback:
+            log.error(f"Error occurred during auxiliary {self.name} execution: {type=}, {value=}, {traceback=}")
+        if not stop_status:
+            raise RuntimeError(f"Failed to stop auxiliary {self.name}")
+
     def suspend(self) -> bool:
         """Supend current auxiliary's run.
 

src/pykiso/lib/auxiliaries/proxy_auxiliary.py

@@ -357,7 +357,7 @@ def _receive_message(self, timeout_in_s: float = 0) -> None:
             if received_data is not None:
                 self.logger.debug(
                     "received response : data %s || channel : %s",
-                    received_data.hex(),
+                    received_data,
                     self.channel.name,
                 )
                 for conn in self.proxy_channels:

src/pykiso/lib/connectors/cc_raw_loopback.py

@@ -19,6 +19,7 @@
 
 """
 import threading
+import time
 from collections import deque
 from typing import Dict, Optional
 
@@ -62,8 +63,12 @@ def _cc_receive(self, timeout: float) -> Dict[str, Optional[bytes]]:
         :return: dictionary containing the received bytes if successful, otherwise None
         """
         with self.lock:
-            try:
-                recv_msg = self._loopback_buffer.popleft()
-                return {"msg": recv_msg}
-            except IndexError:
-                return {"msg": None}
+            # Simulate a blocking on receive
+            start = time.time_ns()
+            while (time.time_ns() - start) < timeout * 1e9:
+                try:
+                    recv_msg = self._loopback_buffer.popleft()
+                    return {"msg": recv_msg}
+                except IndexError:
+                    time.sleep(0.1)
+            return {"msg": None}

tests/dummy_serial.yaml

@@ -0,0 +1,12 @@
+auxiliaries:
+  com_aux_sender:
+    connectors:
+        com: loopback
+    type: pykiso.lib.auxiliaries.communication_auxiliary:CommunicationAuxiliary
+  com_aux_receiver:
+    connectors:
+        com: loopback
+    type: pykiso.lib.auxiliaries.communication_auxiliary:CommunicationAuxiliary
+connectors:
+  loopback:
+    type: pykiso.lib.connectors.cc_raw_loopback:CCLoopback