Merge pull request #158 from ni/users/semoore/get-switch-sessions

Author: sean-moore3

STATUS.md

@@ -70,10 +70,10 @@
 | Custom          | SetCustomSession                                                        | set\_custom\_session                            | test\_custom\_instruments |
 | Custom          | GetCustomSession(s)                                                     | pins\_to\_custom\_session(s)                    | test\_custom\_instruments |
 | Custom          | GetAllCustomSessions                                                    | get\_all\_custom\_sessions                      | test\_custom\_instruments |
-| Multiplexer     | GetSwitchNames                                                          |                                                 |                           |
-| Multiplexer     | SetSwitchSession                                                        |                                                 |                           |
-| Multiplexer     | GetSwitchSession(s)                                                     |                                                 |                           |
-| Multiplexer     | GetAllSwitchSessions                                                    |                                                 |                           |
+| Multiplexer     | GetSwitchNames                                                          | get\_all\_switch\_names                         | test\_switch              |
+| Multiplexer     | SetSwitchSession                                                        | set\_switch\_session                            | test\_switch              |
+| Multiplexer     | GetSwitchSession(s)                                                     | pin\_to\_switch\_sessions                       | test\_switch              |
+| Multiplexer     | GetAllSwitchSessions                                                    | get\_all\_switch\_sessions                      | test\_switch              |
 
 ## Pin Query API
 | Class                           | .NET Method                                    | Python                                                                                             | Python System Tests |

requirements.txt

@@ -10,3 +10,4 @@ tox
 pytest
 pytest-cov
 black
+ni-python-styleguide

src/nitsm/tsmcontext.py

@@ -1,12 +1,13 @@
 """TSM Context Wrapper"""
-
+import ctypes.wintypes
 import time
 import typing
 
 import nitsm._pinmapinterfaces
 import nitsm.enums
 import nitsm.pinquerycontexts
 import pythoncom
+import win32com.client
 
 __all__ = ["SemiconductorModuleContext"]
 
@@ -18,7 +19,6 @@
     import nifgen
     import niscope
     import niswitch
-    import win32com.client.dynamic
 
     _Any = typing.Any
     _Tuple = typing.Tuple
@@ -32,6 +32,7 @@
     _CapabilityArg = _Union[nitsm.enums.Capability, str]
     _PinsArg = _Union[str, _Sequence[str]]  # argument that accepts 1 or more pins
     _StringTuple = _Tuple[str, ...]
+    _AnyTuple = _Tuple[_Any, ...]
 
     _NIDigitalSingleSessionPpmuQuery = _Tuple[_PinQueryContext, nidigital.Session, str]
     _NIDigitalMultipleSessionPpmuQuery = _Tuple[
@@ -69,6 +70,8 @@
         _PinQueryContext, _Tuple[niscope.Session, ...], _StringTuple
     ]
 
+    _SwitchQuery = _Tuple[_Tuple["SemiconductorModuleContext", ...], _AnyTuple, _StringTuple]
+
     _RelayDriverSingleSessionQuery = _Tuple[niswitch.Session, str]
     _RelayDriverMultipleSessionQuery = _Tuple[_Tuple[niswitch.Session, ...], _StringTuple]
 
@@ -93,16 +96,15 @@ class SemiconductorModuleContext:
 
     _sessions = {}
 
-    def __init__(self, tsm_com_obj: "_ISemiconductorModuleContext"):
+    def __init__(self, tsm_dispatch: "_ISemiconductorModuleContext"):
         """Wraps an instance of ISemiconductorModuleContext.
 
         Args:
-            tsm_com_obj: The win32com.client.dynamic.CDispatch object provided by TestStand.
+            tsm_dispatch: The win32com.client.dynamic.CDispatch object provided by TestStand.
         """
-        self._context = nitsm._pinmapinterfaces.ISemiconductorModuleContext(tsm_com_obj)
-        self._context._oleobj_ = tsm_com_obj._oleobj_.QueryInterface(
-            self._context.CLSID, pythoncom.IID_IDispatch
-        )
+        clsid = nitsm._pinmapinterfaces.ISemiconductorModuleContext.CLSID
+        interface = tsm_dispatch._oleobj_.QueryInterface(clsid, pythoncom.IID_IDispatch)
+        self._context = nitsm._pinmapinterfaces.ISemiconductorModuleContext(interface)
 
     # General and Advanced
 
@@ -910,6 +912,116 @@ def pins_to_niscope_sessions(self, pins: "_PinsArg") -> "_NIScopeMultipleSession
         )
         return pin_query_context, sessions, channel_lists
 
+    # Switching
+
+    def get_all_switch_names(self, multiplexer_type_id: "_InstrTypeIdArg") -> "_StringTuple":
+        """Returns the names of all switches of the type specified by the multiplexer_type_id in the
+        Semiconductor Module context. You can use switch names to open driver sessions.
+
+        Args:
+            multiplexer_type_id: Specifies the type ID for the multiplexer in the pin map file. When
+                you add a multiplexer to the pin map file, you can define a type ID for the
+                multiplexer, such as the driver name. Multiplexers in the pin map that do not
+                specify a type ID have a default ID of
+                nitsm.enums.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER.
+        """
+        if isinstance(multiplexer_type_id, nitsm.enums.InstrumentTypeIdConstants):
+            multiplexer_type_id = multiplexer_type_id.value
+        return self._context.GetSwitchNames(multiplexer_type_id)
+
+    def set_switch_session(
+        self, switch_name: str, session_data: "_Any", multiplexer_type_id: "_InstrTypeIdArg"
+    ) -> None:
+        """Associates an open switch session with the switch_name for a multiplexer of type
+        multiplexer_type_id.
+
+        Args:
+            switch_name: The instrument name in the pin map file for the corresponding session_data.
+            session_data: The instrument session for the corresponding switch_name and
+                multiplexer_type_id.
+            multiplexer_type_id: Specifies the type ID for the multiplexer in the pin map file. When
+                you add a multiplexer to the pin map file, you can define a type ID for the
+                multiplexer, such as the driver name. Multiplexers in the pin map that do not
+                specify a type ID have a default ID of
+                nitsm.enums.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER.
+        """
+        if isinstance(multiplexer_type_id, nitsm.enums.InstrumentTypeIdConstants):
+            multiplexer_type_id = multiplexer_type_id.value
+        session_id = id(session_data)
+        self._sessions[session_id] = session_data
+        return self._context.SetSwitchSession(multiplexer_type_id, switch_name, session_id)
+
+    def get_all_switch_sessions(self, multiplexer_type_id: "_InstrTypeIdArg") -> "_AnyTuple":
+        """Returns a tuple of all switch session data of the type specified by the
+        multiplexer_type_id in the Semiconductor Module context.
+
+        Args:
+            multiplexer_type_id: Specifies the type ID for the multiplexer in the pin map file. When
+                you add a multiplexer to the pin map file, you can define a type ID for the
+                multiplexer, such as the driver name. Multiplexers in the pin map that do not
+                specify a type ID have a default ID of
+                nitsm.enums.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER.
+        """
+        if isinstance(multiplexer_type_id, nitsm.enums.InstrumentTypeIdConstants):
+            multiplexer_type_id = multiplexer_type_id.value
+        session_ids = self._context.GetSwitchSessions(multiplexer_type_id)
+        return tuple(map(SemiconductorModuleContext._sessions.get, session_ids))
+
+    def pin_to_switch_sessions(
+        self, pin: str, multiplexer_type_id: "_InstrTypeIdArg"
+    ) -> "_SwitchQuery":
+        """Returns the switch sessions, switch routes, and new Semiconductor Module context objects
+        required to access the specified switched pin.
+
+        Args:
+            pin: The name of the pin to translate to session data and switch routes.
+            multiplexer_type_id: Specifies the type ID for the multiplexer in the pin map file. When
+                you add a multiplexer to the pin map file, you can define a type ID for the
+                multiplexer, such as the driver name. Multiplexers in the pin map that do not
+                specify a type ID have a default ID of
+                nitsm.enums.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER.
+
+        Returns:
+            semiconductor_module_contexts: A tuple of Semiconductor Module context objects. Each
+                element in the tuple represents a site that must be executed serially. Use each
+                Semiconductor Module context object to query the pin map and publish data.
+            session_data: A tuple of the session data required to access the switch that connects an
+                instrument channel to the pin.
+            switch_routes: The routes required to connect an instrument channel to the pin.
+        """
+        if isinstance(multiplexer_type_id, nitsm.enums.InstrumentTypeIdConstants):
+            multiplexer_type_id = multiplexer_type_id.value
+        # We have to use DumbDispatch here because pywin32 fails to recognize
+        # ISemiconductorModuleContext as deriving from IDispatch; most likely because it isn't
+        # natively supported. So, we fetch it as IUnknown instead.
+        vt_by_ref_array = pythoncom.VT_BYREF | pythoncom.VT_ARRAY
+        site_contexts = win32com.client.VARIANT(vt_by_ref_array | pythoncom.VT_UNKNOWN, [])
+        sessions = win32com.client.VARIANT(vt_by_ref_array | pythoncom.VT_VARIANT, [])
+        switch_routes = win32com.client.VARIANT(vt_by_ref_array | pythoncom.VT_BSTR, [])
+        dumb_context = win32com.client.dynamic.DumbDispatch(self._context)
+        dumb_context.GetSwitchSessions_2(
+            multiplexer_type_id, pin, site_contexts, sessions, switch_routes
+        )
+        # As of pywin32 303, there is a bug where SAFEARRAYs of IUnknown pointers leak references.
+        # See here: https://github.com/mhammond/pywin32/issues/1864
+        # As a work-around, we decrement the reference count until the only one left is held by
+        # Python. It will be released when the object is garbage collected.
+        add_ref = ctypes.WINFUNCTYPE(ctypes.wintypes.ULONG)(1, "AddRef")
+        release = ctypes.WINFUNCTYPE(ctypes.wintypes.ULONG)(2, "Release")
+        for site_context in site_contexts.value:
+            # address of IUnknown has to be parsed from the repr
+            # https://github.com/mhammond/pywin32/blob/main/com/win32com/src/PyIUnknown.cpp
+            address = ctypes.c_void_p(int(repr(site_context).split()[-1][:-1], 16))
+            # first add a reference in case the bug has been fixed; prevents count from reaching 0
+            add_ref(address)
+            # then release the reference until only one remains
+            while release(address) > 1:
+                pass
+        site_contexts = map(win32com.client.dynamic.DumbDispatch, site_contexts.value)
+        site_contexts = tuple(map(SemiconductorModuleContext, site_contexts))
+        sessions = tuple(map(SemiconductorModuleContext._sessions.get, sessions.value))
+        return site_contexts, sessions, switch_routes.value
+
     # Relay Driver
 
     def get_relay_driver_module_names(self) -> "_StringTuple":

systemtests/switch.pinmap

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="utf-8"?>
+<PinMap schemaVersion="1.5" xmlns="http://www.ni.com/TestStand/SemiconductorModule/PinMap.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<Instruments>
+		<NIDigitalPatternInstrument name="DigitalPattern1" numberOfChannels="32" group="Digital" />
+		<Multiplexer name="Multiplexer1" multiplexerTypeId="SimulatedMultiplexer" />
+	</Instruments>
+	<Pins>
+		<DUTPin name="DUTPin1" />
+	</Pins>
+	<PinGroups></PinGroups>
+	<Sites>
+		<Site siteNumber="0" />
+		<Site siteNumber="1" />
+	</Sites>
+	<Connections>
+		<MultiplexedConnection instrument="DigitalPattern1" channel="0">
+			<MultiplexedDUTPinRoute pin="DUTPin1" siteNumber="0" multiplexer="Multiplexer1" routeName="DUTPin1Site0" />
+			<MultiplexedDUTPinRoute pin="DUTPin1" siteNumber="1" multiplexer="Multiplexer1" routeName="DUTPin1Site1" />
+		</MultiplexedConnection>
+	</Connections>
+</PinMap>

systemtests/switch.seq

@@ -0,0 +1,45 @@
+import nitsm.codemoduleapi as tsm
+
+MULTIPLEXER_TYPE_ID = "SimulatedMultiplexer"
+
+
+@tsm.code_module
+def open_sessions(tsm_context: tsm.SemiconductorModuleContext):
+    switch_names = tsm_context.get_all_switch_names(MULTIPLEXER_TYPE_ID)
+    for switch_name in switch_names:
+        tsm_context.set_switch_session(switch_name, switch_name, MULTIPLEXER_TYPE_ID)
+    # nidigital sessions are required to satisfy the pin map but won't be used
+    instrument_names = tsm_context.get_all_nidigital_instrument_names()
+    for instrument_name in instrument_names:
+        tsm_context.set_nidigital_session(instrument_name, ...)
+
+
+@tsm.code_module
+def measure(
+    tsm_context: tsm.SemiconductorModuleContext, pin, expected_switch_names, expected_switch_routes
+):
+    site_contexts, switch_names, switch_routes = tsm_context.pin_to_switch_sessions(
+        pin, MULTIPLEXER_TYPE_ID
+    )
+    expected_names_and_routes = set(zip(expected_switch_names, expected_switch_routes))
+    valid_names_and_routes = []
+
+    for site_context, switch_name, switch_route in zip(site_contexts, switch_names, switch_routes):
+        # check switch route we received is in the set of switch routes we expected
+        actual_name_and_route = (switch_name, switch_route)
+        valid_name_and_route = actual_name_and_route in expected_names_and_routes
+        valid_names_and_routes.append(valid_name_and_route)
+        expected_names_and_routes.discard(actual_name_and_route)
+        # get a pin query context and publish result
+        pin_query = site_context.pins_to_nidigital_session_for_ppmu(pin)[0]
+        pin_query.publish(valid_name_and_route)
+
+    # publish missing switch routes for all sites
+    pin_query = tsm_context.pins_to_nidigital_session_for_ppmu(pin)[0]
+    num_missing_routes = [len(expected_names_and_routes)] * len(site_contexts)
+    pin_query.publish(num_missing_routes, "NumMissing")
+
+
+@tsm.code_module
+def close_sessions(tsm_context: tsm.SemiconductorModuleContext):
+    tsm_context.get_all_switch_sessions(MULTIPLEXER_TYPE_ID)

systemtests/switch_codemodules.py

@@ -50,3 +50,8 @@ def test_site_and_global_data(system_test_runner):
 @pytest.mark.sequence_file("specifications.seq")
 def test_specifications(system_test_runner):
     assert system_test_runner.run()
+
+
+@pytest.mark.sequence_file("switch.seq")
+def test_switch(system_test_runner):
+    assert system_test_runner.run()

systemtests/test_system.py

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="utf-8"?>
+<PinMap schemaVersion="1.5" xmlns="http://www.ni.com/TestStand/SemiconductorModule/PinMap.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<Instruments>
+		<NIDigitalPatternInstrument name="DigitalPattern1" numberOfChannels="32" group="Digital" />
+		<Multiplexer name="Multiplexer1" multiplexerTypeId="NIGenericMultiplexer" />
+	</Instruments>
+	<Pins>
+		<DUTPin name="DUTPin1" />
+	</Pins>
+	<PinGroups></PinGroups>
+	<Sites>
+		<Site siteNumber="0" />
+		<Site siteNumber="1" />
+	</Sites>
+	<Connections>
+		<MultiplexedConnection instrument="DigitalPattern1" channel="0">
+			<MultiplexedDUTPinRoute pin="DUTPin1" siteNumber="0" multiplexer="Multiplexer1" routeName="DUTPin1Site0" />
+			<MultiplexedDUTPinRoute pin="DUTPin1" siteNumber="1" multiplexer="Multiplexer1" routeName="DUTPin1Site1" />
+		</MultiplexedConnection>
+	</Connections>
+</PinMap>

tests/switch.pinmap

@@ -0,0 +1,66 @@
+import pytest
+import nitsm.codemoduleapi as tsm
+
+
+@pytest.fixture
+def simulated_switch_sessions(standalone_tsm_context):
+    switch_names = standalone_tsm_context.get_all_switch_names(
+        tsm.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER
+    )
+    for switch_name in switch_names:
+        standalone_tsm_context.set_switch_session(
+            switch_name, switch_name, tsm.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER
+        )
+    yield switch_names
+
+
+@pytest.mark.pin_map("switch.pinmap")
+@pytest.mark.parametrize(
+    "multiplexer_type_id",
+    ["NIGenericMultiplexer", tsm.InstrumentTypeIdConstants.NI_GENERIC_MULTIPLEXER],
+)
+class TestSwitch:
+    switches = ["Multiplexer1"]
+
+    def test_get_all_switch_names(self, standalone_tsm_context, multiplexer_type_id):
+        switch_names = standalone_tsm_context.get_all_switch_names(multiplexer_type_id)
+        assert isinstance(switch_names, tuple)
+        assert len(switch_names) == len(self.switches)
+        for switch_name in switch_names:
+            assert isinstance(switch_name, str)
+            assert switch_name in self.switches
+
+    def test_set_switch_session(self, standalone_tsm_context, multiplexer_type_id):
+        switch_names = standalone_tsm_context.get_all_nidmm_instrument_names()
+        for switch_name in switch_names:
+            standalone_tsm_context.set_switch_session(switch_name, switch_name, multiplexer_type_id)
+            assert tsm.SemiconductorModuleContext._sessions[id(switch_name)] is switch_name
+
+    def test_get_all_switch_sessions(
+        self, standalone_tsm_context, simulated_switch_sessions, multiplexer_type_id
+    ):
+        queried_sessions = standalone_tsm_context.get_all_switch_sessions(multiplexer_type_id)
+        assert isinstance(queried_sessions, tuple)
+        assert len(queried_sessions) == len(simulated_switch_sessions)
+        for queried_session in queried_sessions:
+            assert isinstance(queried_session, str)
+            assert queried_session in simulated_switch_sessions
+
+    def test_pin_to_switch_sessions(
+        self, standalone_tsm_context, simulated_switch_sessions, multiplexer_type_id
+    ):
+        tsm_contexts, sessions, switch_routes = standalone_tsm_context.pin_to_switch_sessions(
+            "DUTPin1", multiplexer_type_id
+        )
+        assert isinstance(tsm_contexts, tuple)
+        assert isinstance(sessions, tuple)
+        assert isinstance(switch_routes, tuple)
+        assert len(tsm_contexts) == len(sessions)
+        assert len(sessions) == len(switch_routes)
+        for tsm_context, session, switch_route in zip(tsm_contexts, sessions, switch_routes):
+            assert isinstance(tsm_context, tsm.SemiconductorModuleContext)
+            assert len(tsm_context.site_numbers) == 1
+            assert isinstance(session, str)
+            assert session in simulated_switch_sessions
+            assert isinstance(switch_route, str)
+            assert switch_route == f"DUTPin1Site{tsm_context.site_numbers[0]}"

tests/test_switch.py

@@ -1,9 +1,13 @@
+"""Generates _pinmapinterfaces.py
+
+You must register the correct version of TSM with TestStand Version Selector prior to running this
+script.
+"""
+
 import sys
 import os.path
-import winreg
 from win32com.client import makepy
 
-
 output_file = os.path.join(os.path.dirname(__file__), "_pinmapinterfaces.py")
 teststand_public_path = os.environ["TestStandPublic64"]
 pmi_type_library = os.path.join(