Possible driver manager concept

src/fixate/__main__.py

@@ -13,6 +13,7 @@
 from fixate.core.ui import user_info_important, user_serial, user_ok
 from fixate.reporting import register_csv, unregister_csv
 from fixate.ui_cmdline import register_cmd_line, unregister_cmd_line
+from fixate.core.common import TestScript
 import fixate.sequencer
 
 parser = ArgumentParser(
@@ -335,7 +336,7 @@ def ui_run(self):
                 return ReturnCodes.ERROR
 
 
-def retrieve_test_data(test_suite, index):
+def retrieve_test_data(test_suite, index) -> TestScript:
     """
     Tries to retrieve test data from the loaded test_suite module
     :param test_suite: Imported module with tests available
@@ -346,7 +347,7 @@ def retrieve_test_data(test_suite, index):
         data = test_suite.test_data
     except AttributeError:
         # Try legacy API
-        return test_suite.TEST_SEQUENCE
+        return test_suite.TEST_SCRIPT
     try:
         sequence = data[index]
     except KeyError as e:

src/fixate/core/common.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+import dataclasses
 import re
 import sys
 import threading
@@ -7,6 +9,8 @@
 import warnings
 from functools import wraps
 from collections import namedtuple
+from typing import TypeVar, Generic, List, Optional, Union, Iterable
+
 from fixate.core.exceptions import ParameterError, InvalidScalarQuantityError
 
 logger = logging.getLogger(__name__)
@@ -326,16 +330,19 @@ def inner(*args, **kwargs):
     return inner
 
 
+DmType = TypeVar("DmType")
+
+
 # The first line of the doc string will be reflected in the test logs. Please don't change.
-class TestList:
+class TestList(Generic[DmType]):
     """
     Test List
     The TestList is a container for TestClasses and TestLists to set up a test hierarchy.
     They operate similar to a python list except that it has additional methods that can be overridden to provide additional functionality
     """
 
-    def __init__(self, seq=None):
-        self.tests = []
+    def __init__(self, seq: Optional[List[Union[TestClass[DmType], TestList[DmType]]]] = None):
+        self.tests: List[Union[TestClass[DmType], TestList[DmType]]] = []
         if seq is None:
             seq = []
         self.tests.extend(seq)
@@ -352,56 +359,60 @@ def __init__(self, seq=None):
                 self.test_desc = doc_string[0]
                 self.test_desc_long = "\\n".join(doc_string[1:])
 
-    def __getitem__(self, item):
+    def __getitem__(self, item) -> Union[TestClass[DmType], TestList[DmType]]:
         return self.tests.__getitem__(item)
 
-    def __contains__(self, item):
+    def __contains__(self, item) -> bool:
         return self.tests.__contains__(item)
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: int, value:  Union[TestClass[DmType], TestList[DmType]]) -> None:
         return self.tests.__setitem__(key, value)
 
-    def __delitem__(self, key):
+    def __delitem__(self, key: int) -> None:
         return self.tests.__delitem__(key)
 
-    def __len__(self):
+    def __len__(self) -> int:
         return self.tests.__len__()
 
-    def append(self, p_object):
+    def append(self, p_object: Union[TestClass[DmType], TestList[DmType]]) -> None:
         self.tests.append(p_object)
 
-    def extend(self, iterable):
+    def extend(self, iterable: Iterable[Union[TestClass[DmType], TestList[DmType]]]):
         self.tests.extend(iterable)
 
-    def insert(self, index, p_object):
+    def insert(self, index: int, p_object: Union[TestClass[DmType], TestList[DmType]]):
         self.tests.insert(index, p_object)
 
-    def index(self, value, start=None, stop=None):
+    def index(self, value: Union[TestClass[DmType], TestList[DmType]], start: int =None, stop: int=None):
         self.tests.index(value, start, stop)
 
-    def set_up(self):
+    def set_up(self, dm: DmType):
         """
         Optionally override this to be called before the set_up of the included TestClass and/or TestList within this TestList
         """
+        pass
 
-    def tear_down(self):
+    def tear_down(self, dm: DmType):
         """
         Optionally override this to be called after the tear_down of the included TestClass's and/or TestList's within this TestList
         This will be called if the set_up has been called regardless of the success of the included TestClass's and/or TestList's
         """
+        pass
 
-    def enter(self):
+    def enter(self, dm: DmType):
         """
         This is called when being pushed onto the stack
         """
+        pass
 
-    def exit(self):
+    def exit(self, dm: DmType):
         """
         This is called when being popped from the stack
         """
+        pass
 
 
-class TestClass:
+class TestClass(Generic[DmType]):
     """
     This class is an abstract base class to implement tests.
     The first line of the docstring of the class that inherits this class will be recognised by logging and UI
@@ -416,7 +427,6 @@ class TestClass:
     test_desc = None
     test_desc_long = None
     attempts = 1
-    tests = []
     retry_type = RT_PROMPT
     retry_exceptions = [BaseException]  # Depreciated
     skip_exceptions = []
@@ -438,19 +448,28 @@ def __init__(self, skip=False):
                     self.test_desc = doc_string[0]
                     self.test_desc_long = "\\n".join(doc_string[1:])
 
-    def set_up(self):
+    def set_up(self, dm: DmType):
         """
         Optionally override this code that is executed before the test method is called
         """
+        pass
 
-    def tear_down(self):
+    def tear_down(self, dm: DmType):
         """
         Optionally override this code that is always executed at the end of the test whether it was successful or not
         """
+        pass
 
-    def test(self):
+    def test(self, dm: DmType):
         """
         This method should be overridden with the test code
         This is the test sequence code
         Use chk functions to set the pass fail criteria for the test
         """
+        raise NotImplementedError
+
+
+@dataclasses.dataclass
+class TestScript(Generic[DmType]):
+    test_list: TestList[DmType]
+    dm_type: type(DmType)

src/fixate/drivers/dmm/__init__.py

@@ -18,7 +18,7 @@
 
 
 def open() -> DMM:
-    for DMM in (Fluke8846A, Keithley6500):
+    for driver_class in (Fluke8846A, Keithley6500):
         instrument = find_instrument_by_id(DMM.REGEX_ID)
         if instrument is not None:
             # We've found a configured instrument so try to open it
@@ -30,7 +30,7 @@ def open() -> DMM:
                     f"Unable to open DMM: {instrument.address}"
                 ) from e
             # Instantiate driver with connected instrument
-            driver = DMM(resource)
+            driver = driver_class(resource)
             fixate.drivers.log_instrument_open(driver)
             return driver
     raise InstrumentNotFoundError

src/fixate/sequencer.py

@@ -1,8 +1,11 @@
+import inspect
 import sys
 import time
 import re
+from typing import Optional, Union, TypeVar, Generic
+
 from pubsub import pub
-from fixate.core.common import TestList, TestClass
+from fixate.core.common import TestList, TestClass, TestScript
 from fixate.core.exceptions import SequenceAbort, CheckFail
 from fixate.core.ui import user_retry_abort_fail
 from fixate.core.checks import CheckResult
@@ -94,7 +97,8 @@ def get_parent_level(level):
 
 class Sequencer:
     def __init__(self):
-        self.tests = TestList()
+        self.test_script: Optional[TestScript] = None
+        self._driver_manager = None
         self._status = "Idle"
         self.active_test = None
         self.ABORT = False
@@ -159,9 +163,9 @@ def status(self, val):
             else:
                 self._status = val
 
-    def load(self, val):
-        self.tests.append(val)
-        self.context.push(self.tests)
+    def load(self, test_script: TestScript):
+        self.context.push(test_script.test_list)
+        self.test_script = test_script
         self.end_status = "N/A"
 
     def count_tests(self):
@@ -225,6 +229,32 @@ def run_sequence(self):
                     top.current().exit()
                 self.context.pop()
 
+    @property
+    def driver_manager(self):
+        if self._driver_manager is None:
+            self._driver_manager = self.test_script.dm_type()
+        return self._driver_manager
+
+    def _cleanup_driver_manager(self):
+        """
+        Attempt to call close on each instrument on the driver manager.
+
+        We assume any non-private attribute on the driver manager (i.e. not starting with '_')
+        is potentially a driver to be closed. Iterate over all such items and if they
+        have a close method call it.
+
+        Finally, set the driver_manager instance back to None, so that all drivers get
+        re-instantiated if they are needed again.
+        """
+        drivers = [
+            driver
+            for name, driver in inspect.getmembers(self._driver_manager)
+            if not name.startswith("_")
+        ]
+        for driver in drivers:
+            if hasattr(driver, "close"):
+                driver.close()
+
     def run_once(self):
         """
         Runs through the tests once as are pushed onto the context stack.
@@ -244,7 +274,7 @@ def run_once(self):
                             data=top.testlist,
                             test_index=self.levels(),
                         )
-                        top.testlist.exit()
+                        top.testlist.exit(self.driver_manager)
                         if self.context:
                             self.context.top().index += 1
                     elif isinstance(top.current(), TestClass):
@@ -261,7 +291,7 @@ def run_once(self):
                             data=top.current(),
                             test_index=self.levels(),
                         )
-                        top.current().enter()
+                        top.current().enter(self.driver_manager)
                         self.context.push(top.current())
                     else:
                         raise SequenceAbort("Unknown Test Item Type")
@@ -271,6 +301,7 @@ def run_once(self):
                         exception=sys.exc_info()[1],
                         test_index=self.levels(),
                     )
+                    self._cleanup_driver_manager()
                     pub.sendMessage("Sequence_Abort", exception=e)
                     self._handle_sequence_abort()
                     return
@@ -315,11 +346,11 @@ def run_test(self):
                 # Run the test
                 try:
                     for index_context, current_level in enumerate(self.context):
-                        current_level.current().set_up()
-                    active_test.test()
+                        current_level.current().set_up(self.driver_manager)
+                    active_test.test(self.driver_manager)
                 finally:
                     for current_level in self.context[index_context::-1]:
-                        current_level.current().tear_down()
+                        current_level.current().tear_down(self.driver_manager)
                 if not self.chk_fail:
                     active_test_status = "PASS"
                     self.tests_passed += 1
@@ -343,6 +374,8 @@ def run_test(self):
                     exception=sys.exc_info()[1],
                     test_index=self.levels(),
                 )
+                self._cleanup_driver_manager()
+
                 attempts = 0
                 active_test_status = "ERROR"
                 if not self.retry_prompt():
@@ -358,9 +391,11 @@ def run_test(self):
                     exception=sys.exc_info()[1],
                     test_index=self.levels(),
                 )
+                self._cleanup_driver_manager()
 
             # Retry Logic
             pub.sendMessage("Test_Retry", data=active_test, test_index=self.levels())
+            self._cleanup_driver_manager()
         pub.sendMessage(
             "Test_Complete",
             data=active_test,