Ecs 6432 Add ability to ignore specific upstream devices in load

docs/source/upcoming_release_notes/297-ECS-6432_Add_ability_to_ignore_specific_upstream_devices_in_load.rst

@@ -0,0 +1,22 @@
+297 ECS-6432 Add ability to ignore specific upstream devices in load
+#################
+
+API Changes
+-----------
+- N/A
+
+Features
+--------
+- Added ability to ignore specific upstream devices when loading hutch-python
+
+Bugfixes
+--------
+- N/A
+
+Maintenance
+-----------
+- N/A
+
+Contributors
+------------
+- janeliu-slac

hutch_python/constants.py

@@ -39,6 +39,7 @@
     'daq_host',
     'obj_config',
     'session_timer',
+    'exclude_devices'
 )
 NO_LOG_EXCEPTIONS = (KeyboardInterrupt, SystemExit)
 LOG_DOMAINS = {".pcdsn", ".slac.stanford.edu"}

hutch_python/happi.py

@@ -29,7 +29,8 @@ def get_happi_objs(
     db: str,
     light_ctrl: LightController,
     endstation: str,
-    load_level: DeviceLoadLevel = DeviceLoadLevel.STANDARD
+    load_level: DeviceLoadLevel = DeviceLoadLevel.STANDARD,
+    exclude_devices: list[str] = None
 ) -> dict[str, ophyd.Device]:
     """
     Get the relevant items for ``endstation`` from the happi database ``db``.
@@ -57,6 +58,11 @@ def get_happi_objs(
     objs: ``dict``
         A mapping from item name to item
     """
+
+    # Explicitly set exclude_devices to empty list to avoid mutable default arguments issue.
+    if not exclude_devices:
+        exclude_devices = []
+
     # Load the happi Client
     if None not in (light_ctrl, lightpath):
         client = light_ctrl.client
@@ -67,47 +73,51 @@ def get_happi_objs(
     if load_level == DeviceLoadLevel.ALL:
         results = client.search(active=True)
         containers.extend(res.item for res in results)
-        return _load_devices(*containers)
 
-    if light_ctrl is None or (endstation.upper() not in light_ctrl.beamlines):
+    elif light_ctrl is None or (endstation.upper() not in light_ctrl.beamlines):
         # lightpath was unavailable, search by beamline name
         reqs = dict(beamline=endstation.upper(), active=True)
         results = client.search(**reqs)
         containers.extend(res.item for res in results)
-        return _load_devices(*containers)
-
-    # if lightpath exists, we can grab upstream devices
-    dev_names = set()
-    paths = light_ctrl.beamlines[endstation.upper()]
-    for path in paths:
-        dev_names.update(path)
-
-    # gather happi items for each of these
-    for name in dev_names:
-        results = client.search(name=name)
-        containers.extend(res.item for res in results)
 
-    if load_level >= DeviceLoadLevel.STANDARD:
-        # also any device with the same beamline name
-        # since lightpath only grabs lightpath-active devices
-        beamlines = {it.beamline for it in containers}
-        beamlines.add(endstation.upper())
-
-        for line in beamlines:
-            # Assume we want hutch items that are active
-            # items can be lightpath-inactive
-            reqs = dict(beamline=line, active=True)
-            results = client.search(**reqs)
-            blc = [res.item for res in results
-                   if res.item.name not in dev_names]
-            # Add the beamline containers to the complete list
-            if blc:
-                containers.extend(blc)
+    else:
+        # if lightpath exists, we can grab upstream devices
+        dev_names = set()
+        paths = light_ctrl.beamlines[endstation.upper()]
+        for path in paths:
+            dev_names.update(path)
+
+        # gather happi items for each of these
+        for name in dev_names:
+            results = client.search(name=name)
+            containers.extend(res.item for res in results)
+
+        if load_level >= DeviceLoadLevel.STANDARD:
+            # also any device with the same beamline name
+            # since lightpath only grabs lightpath-active devices
+            beamlines = {it.beamline for it in containers}
+            beamlines.add(endstation.upper())
+
+            for line in beamlines:
+                # Assume we want hutch items that are active
+                # items can be lightpath-inactive
+                reqs = dict(beamline=line, active=True)
+                results = client.search(**reqs)
+                blc = [res.item for res in results
+                       if res.item.name not in dev_names]
+                # Add the beamline containers to the complete list
+                if blc:
+                    containers.extend(blc)
 
     if len(containers) < 1:
         logger.warning(f'{len(containers)} active devices found for '
                        'this beampath')
 
+    # Do not load excluded devices
+    for device in containers.copy():
+        if device.name in exclude_devices:
+            containers.remove(device)
+
     return _load_devices(*containers)
 
 
@@ -132,7 +142,7 @@ def _load_devices(*containers: happi.HappiItem):
     return dev_namespace.__dict__
 
 
-def get_lightpath(db, hutch) -> LightController:
+def get_lightpath(db: str, hutch: str) -> LightController:
     """
     Create a ``lightpath.LightController`` from relevant ``happi`` objects.
 
@@ -155,10 +165,14 @@ def get_lightpath(db, hutch) -> LightController:
     if None in (lightpath, beamlines):
         logger.warning('Lightpath module is not available.')
         return None
+
     # Load the happi Client
     client = happi.Client(path=db)
+
     # Allow the lightpath module to create a path
     lc = lightpath.LightController(client, endstations=[hutch.upper()])
+
     # Return paths (names only) seen by the LightController
     # avoid loding the devices so hutch-python can keep track of it
+
     return lc

hutch_python/load_conf.py

@@ -297,6 +297,20 @@ def load_conf(conf, hutch_dir=None, args=None):
         daq_platform = 0
         logger.info('Selected default hutch-python daq platform: 0')
 
+    try:
+        # This is a list of all devices that should not be loaded
+        exclude_devices = conf['exclude_devices']
+        if not isinstance(exclude_devices, list):
+            logger.error(
+                'Invalid exclude_devices conf, must be a list.')
+            exclude_devices = []
+        else:
+            exclude_devices = [device_name.strip() for device_name in exclude_devices]
+    except KeyError:
+        exclude_devices = []
+        logger.info(
+            'Missing exclude_devices in conf. Will load all devices.')
+
     # Set the session timeout duration
     try:
         hutch_python.ipython_session_timer.configure_timeout(
@@ -460,7 +474,8 @@ def load_conf(conf, hutch_dir=None, args=None):
             lc = get_lightpath(db, hutch)
 
             # Gather relevant objects given the BeamPath
-            happi_objs = get_happi_objs(db, lc, hutch, load_level=load_level)
+            happi_objs = get_happi_objs(
+                db, lc, hutch, load_level=load_level, exclude_devices=exclude_devices)
             cache(**happi_objs)
 
             # create and store beampath

hutch_python/tests/test_happi.py

@@ -17,9 +17,6 @@ def test_happi_objs():
     logger.debug("test_happi_objs")
     db = os.path.join(os.path.abspath(os.path.dirname(__file__)),
                       'happi_db.json')
-    # patch lightpath configs to include test db beamline
-    conftest.beamlines['TST'] = ['X0']
-    conftest.sources.append('X0')
     # Only select active objects
     lc = get_lightpath(db, 'tst')
     objs = get_happi_objs(db, lc, 'tst')
@@ -44,9 +41,6 @@ def test_load_level(load_level: DeviceLoadLevel, num_devices: int):
     logger.debug("test_load_level")
     db = os.path.join(os.path.abspath(os.path.dirname(__file__)),
                       'happi_db.json')
-    # patch lightpath configs to include test db beamline
-    conftest.beamlines['TST'] = ['X0']
-    conftest.sources.append('X0')
     # Only select active objects
     lc = get_lightpath(db, 'tst')
     objs = get_happi_objs(db, lc, 'tst', load_level=load_level)
@@ -63,3 +57,30 @@ def test_get_lightpath():
     # Check that we created a valid BeamPath with no inactive objects
     assert obj.name == 'TST'
     assert len(obj.devices) == 3
+
+
+# run test_happi_objs() with exclude_devices
+@conftest.requires_lightpath
+def test_happi_objs_with_exclude_devices():
+    # This test checks whether devices from exclude_devices have been removed from the
+    # output of get_happi_objs(). For comparison, get_happi_objs() is first called without
+    # exclude_devices and then called again with exclude_devices.
+    exclude_devices = ['tst_device_5', 'tst_device_1']
+
+    logger.debug("test_happi_objs")
+    db = os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                      'happi_db.json')
+    # Only select active objects
+    lc = get_lightpath(db, 'tst')
+
+    # Call get_happi_objs() without exclude_devices
+    objs = get_happi_objs(db, lc, 'tst', DeviceLoadLevel.STANDARD)
+    assert len(objs) == 4
+
+    # Call get_happi_objs() with exclude_devices
+    objs_exclude_devices = get_happi_objs(db, lc, 'tst', DeviceLoadLevel.STANDARD, exclude_devices)
+    assert len(objs_exclude_devices) == 2
+
+    # Check that none of the loaded devices are in the exclude_devices list
+    for obj in objs_exclude_devices:
+        assert obj not in exclude_devices