Ticket 6742: Generalise IOC_Test_Framework

utils/ca_interface.py

@@ -0,0 +1,79 @@
+import sys
+from caproto.threading.client import Context
+from caproto._utils import CaprotoTimeoutError
+from caproto import ChannelType
+
+def exception_translator(func):
+    def inner_function(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except CaprotoTimeoutError:
+            raise UnableToConnectToPVException(sys.exc_info()[1])
+    return inner_function
+
+
+class CaChannelWrapper:
+    def __init__(self, default_ca_timeout=5):
+        self._context = Context(timeout=default_ca_timeout)
+        self._timeout = self._context.timeout
+
+    def add_monitor(self, pv_name, callback=None):
+        """
+        callback function must have call signature:
+
+            def f(monitor, response):
+              ...
+
+        The name of pv that fired the monitor event is available as
+        monitor.pv.name
+
+        see https://nsls-ii.github.io/caproto/threading-client.html
+        """
+
+        pv = self._context.get_pvs(pv_name)[0]
+        monitor = pv.subscribe()
+        if callback:
+            monitor.add_callback(callback)
+
+    def poll(self):
+        pass
+
+    def errorLogFunc(self, *args, **kwargs):
+        return None
+
+    def pv_exists(self, pv_name, timeout=None):
+        timeout = timeout or self._timeout
+        pv = self._context.get_pvs(pv_name)[0]
+        try:
+            pv.read()
+            return True
+        except CaprotoTimeoutError:
+            return False
+
+    @exception_translator
+    def set_pv_value(self, pv_name, value, wait=False, timeout=None):
+        timeout = timeout or self._timeout
+        pv = self._context.get_pvs(pv_name)[0]
+        pv.write(value, wait=wait, timeout=timeout)
+
+    @exception_translator
+    def get_pv_value(self, pv_name):
+        pv = self._context.get_pvs(pv_name)[0]
+        result = pv.read()
+        received_data = result.data
+        #Deal with string/bytes in P3
+        if result.data_type == ChannelType.STRING:
+            received_data = self.decode_bytes_if_necessary(received_data)
+        return received_data if len(received_data) > 1 else received_data[0]
+
+    @staticmethod
+    def decode_bytes_if_necessary(data_list):
+        if not type(data_list[0]) == bytes:
+            return data_list
+        return [element.decode() for element in data_list]
+
+
+
+class UnableToConnectToPVException(Exception):
+    pass
+

utils/channel_access.py

@@ -6,7 +6,7 @@
 import operator
 import ctypes
 from contextlib import contextmanager
-from genie_python.genie_cachannel_wrapper import CaChannelWrapper, UnableToConnectToPVException
+from .ca_interface import CaChannelWrapper, UnableToConnectToPVException
 
 from functools import partial
 

utils/emulator_launcher.py

@@ -4,6 +4,7 @@
 import abc
 import os
 import subprocess
+import psutil
 
 import sys
 import datetime
@@ -332,8 +333,8 @@ class LewisLauncher(EmulatorLauncher):
     Launches Lewis.
     """
 
-    _DEFAULT_PY_PATH = os.path.join("C:\\", "Instrument", "Apps", "Python3")
-    _DEFAULT_LEWIS_PATH = os.path.join(_DEFAULT_PY_PATH, "scripts")
+    _DEFAULT_PY_PATH, _ = os.path.split(sys.executable) #os.path.join("C:\\", "Instrument", "Apps", "Python3")
+    _DEFAULT_LEWIS_PATH = os.path.join(_DEFAULT_PY_PATH)
 
     def __init__(self, test_name, device, var_dir, port, options):
         """
@@ -348,7 +349,6 @@ def __init__(self, test_name, device, var_dir, port, options):
         super(LewisLauncher, self).__init__(test_name, device, var_dir, port, options)
 
         self._lewis_path = options.get("lewis_path", LewisLauncher._DEFAULT_LEWIS_PATH)
-        self._python_path = options.get("python_path", os.path.join(LewisLauncher._DEFAULT_PY_PATH, "python.exe"))
         self._lewis_protocol = options.get("lewis_protocol", "stream")
         self._lewis_additional_path = options.get("lewis_additional_path", DEVICE_EMULATOR_PATH)
         self._lewis_package = options.get("lewis_package", "lewis_emulators")
@@ -364,8 +364,15 @@ def _close(self):
         Closes the Lewis session by killing the process.
         """
         print("Terminating Lewis")
+        # If launching with shell=True then simply doing process.terminate() 
+        # kills the shell but not Lewis. Solution is
+        # https://stackoverflow.com/questions/4789837/how-to-terminate-a-python-subprocess-launched-with-shell-true
         if self._process is not None:
-            self._process.terminate()
+            #self._process.terminate()
+            process = psutil.Process(self._process.pid)
+            for p in process.children(recursive=True):
+                p.kill()
+            process.kill()
         if self._logFile is not None:
             self._logFile.close()
             print("Lewis log written to {0}".format(self._log_filename()))
@@ -379,9 +386,9 @@ def _open(self):
         """
 
         self._control_port = str(get_free_ports(1)[0])
-        lewis_command_line = [self._python_path, "-m", "lewis",
+        lewis_command_line = ["lewis",
                               "-r", "127.0.0.1:{control_port}".format(control_port=self._control_port)]
-        lewis_command_line.extend(["-p", "{protocol}: {{bind_address: 127.0.0.1, port: {port}}}"
+        lewis_command_line.extend(["-p", '"{protocol}: {{bind_address: 127.0.0.1, port: {port}}}"'
                                   .format(protocol=self._lewis_protocol, port=self._port)])
         if self._lewis_additional_path is not None:
             lewis_command_line.extend(["-a", self._lewis_additional_path])
@@ -395,8 +402,10 @@ def _open(self):
         self._logFile = open(self._log_filename(), "w")
         self._logFile.write("Started Lewis with '{0}'\n".format(" ".join(lewis_command_line)))
 
-        self._process = subprocess.Popen(lewis_command_line,
-                                         creationflags=subprocess.CREATE_NEW_CONSOLE,
+        # Convert *args into str because Lewis throws error 
+        # when passing quote marks (which are need for the -p argument)
+        self._process = subprocess.Popen(" ".join(lewis_command_line),
+                                         shell=True,
                                          stdout=self._logFile,
                                          stderr=subprocess.STDOUT)
         self._connected = True

utils/free_ports.py

@@ -12,7 +12,7 @@ def get_free_ports(n):
     ports = list()
     for i in range(0, n):
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+        #s.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
         s.bind(("", 0))
         ports.append(s.getsockname()[1])
         socks.append(s)
@@ -34,7 +34,7 @@ def get_free_ports_from_list(n, port_low, port_high):
     ports = list()
     for i in range(0, n):
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+        #s.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) # Don't use this when runnign under linux
         for j in range(port_low, port_high):
             try:
                 s.bind(("", j))

utils/ioc_launcher.py

@@ -3,6 +3,7 @@
 """
 import subprocess
 import os
+import sys
 import time
 from contextlib import contextmanager
 
@@ -22,8 +23,8 @@
 
 APPS_BASE = os.path.join("C:\\", "Instrument", "Apps")
 EPICS_TOP = os.environ.get("KIT_ROOT", os.path.join(APPS_BASE, "EPICS"))
-PYTHON = os.environ.get("PYTHON", os.path.join(APPS_BASE, "Python", "python.exe"))
-PYTHON3 = os.environ.get("PYTHON3", os.path.join(APPS_BASE, "Python3", "python.exe"))
+PYTHON = os.environ.get("PYTHON", sys.executable)
+PYTHON3 = os.environ.get("PYTHON3", sys.executable)
 
 DEFAULT_IOC_START_TEXT = "epics>"
 MAX_TIME_TO_WAIT_FOR_IOC_TO_START = 120
@@ -176,7 +177,7 @@ def open(self):
             # To be able to see the IOC output for debugging, remove the redirection of stdin, stdout and stderr.
             # This does mean that the IOC will need to be closed manually after the tests.
             # Make sure to revert before checking code in
-            self._process = subprocess.Popen(" ".join(self.command_line), creationflags=subprocess.CREATE_NEW_CONSOLE,
+            self._process = subprocess.Popen(" ".join(self.command_line),shell=True ,
                                              cwd=self._directory, stdin=subprocess.PIPE,
                                              stdout=self.log_file_manager.log_file, stderr=subprocess.STDOUT,
                                              env=settings)
@@ -547,15 +548,15 @@ def __init__(self, test_name, ioc, test_mode, var_dir):
         super(IocLauncher, self).__init__(test_name, ioc, test_mode, var_dir)
 
     def _command_line(self):
-        run_ioc_path = os.path.join(self._directory, 'runIOC.bat')
+        #run_ioc_path = os.path.join(self._directory, 'runIOC.bat')
         st_cmd_path = os.path.join(self._directory, 'st.cmd')
 
-        if not os.path.isfile(run_ioc_path):
-            print("Run IOC path not found: '{0}'".format(run_ioc_path))
+        #if not os.path.isfile(run_ioc_path):
+        #    print("Run IOC path not found: '{0}'".format(run_ioc_path))
         if not os.path.isfile(st_cmd_path):
             print("St.cmd path not found: '{0}'".format(st_cmd_path))
 
-        return [run_ioc_path, st_cmd_path]
+        return [st_cmd_path]
 
     def close(self):
         """