Repository: terrapower/armi
Pull request #1560 (2023-12-19 15:55:12)

Create method to retrieve active interfaces
## What is the change?
Adds a method on the Operator to retrieve the active interfaces. Aims to reduce copy and pasted code and improve testing for what interfaces should be active for a given interact state.

## Why is the change being made?
Being able to check which interfaces are active for a given interact state is particularly useful for ARMI Applications.
<!-- MANDATORY: Explain why the change is necessary -->
<!-- Optional: Link to any related GitHub Issues -->

---

## Checklist

<!--
    You (the pull requester) should put an `x` in the boxes below you have completed.
    If you're unsure about any of them, don't hesitate to ask. We're here to help!
    (If a checkbox requires no action for this PR, put an `x` in the box.)
-->

- [x] This PR has only [one purpose or idea](https://terrapower.github.io/armi/developer/tooling.html#one-idea-one-pr).
- [x] [Tests](https://terrapower.github.io/armi/developer/tooling.html#test-it) have been added/updated to verify that the new/changed code works.

<!-- Check the code quality -->

- [x] The code style follows [good practices](https://terrapower.github.io/armi/developer/standards_and_practices.html).
- [x] The commit message(s) follow [good practices](https://terrapower.github.io/armi/developer/tooling.html).

<!-- Check the project-level cruft -->

- [x] The [release notes](https://terrapower.github.io/armi/release/index.html) (location `doc/release/0.X.rst`) are up-to-date with any important changes.
- [x] The [documentation](https://terrapower.github.io/armi/developer/tooling.html#document-it) is still up-to-date in the `doc` folder.
- [x] No [requirements](https://terrapower.github.io/armi/developer/tooling.html#watch-for-requirements) were altered.
- [x] The dependencies are still up-to-date in `pyproject.toml`.



Patch:
diff --git a/armi/operators/operator.py b/armi/operators/operator.py
index 4cf08a1d..c17875d5 100644
--- a/armi/operators/operator.py
+++ b/armi/operators/operator.py
@@ -27,6 +27,7 @@ import os
 import re
 import shutil
 import time
+from typing import Tuple
 
 from armi import context
 from armi import interfaces
@@ -43,6 +44,8 @@ from armi.settings.fwSettings.globalSettings import (
     CONF_TIGHT_COUPLING,
     CONF_TIGHT_COUPLING_MAX_ITERS,
     CONF_CYCLES_SKIP_TIGHT_COUPLING_INTERACTION,
+    CONF_DEFERRED_INTERFACE_NAMES,
+    CONF_DEFERRED_INTERFACES_CYCLE,
 )
 from armi.utils import codeTiming
 from armi.utils import (
@@ -574,8 +577,7 @@ class Operator:
 
     def interactAllInit(self):
         """Call interactInit on all interfaces in the stack after they are initialized."""
-        allInterfaces = self.interfaces[:]  # copy just in case
-        self._interactAll("Init", allInterfaces)
+        self._interactAll("Init", self.getInterfaces())
 
     def interactAllBOL(self, excludedInterfaceNames=()):
         """
@@ -583,30 +585,15 @@ class Operator:
 
         All enabled or bolForce interfaces will be called excluding interfaces with excludedInterfaceNames.
         """
-        activeInterfaces = [
-            ii
-            for ii in self.interfaces
-            if (ii.enabled() or ii.bolForce()) and ii.name not in excludedInterfaceNames
-        ]
-        activeInterfaces = [
-            ii
-            for ii in activeInterfaces
-            if ii.name not in self.cs["deferredInterfaceNames"]
-        ]
+        activeInterfaces = self.getActiveInterfaces("BOL", excludedInterfaceNames)
         self._interactAll("BOL", activeInterfaces)
 
     def interactAllBOC(self, cycle):
         """Interact at beginning of cycle of all enabled interfaces."""
-        activeInterfaces = [ii for ii in self.interfaces if ii.enabled()]
-        if cycle < self.cs["deferredInterfacesCycle"]:
-            activeInterfaces = [
-                ii
-                for ii in activeInterfaces
-                if ii.name not in self.cs["deferredInterfaceNames"]
-            ]
+        activeInterfaces = self.getActiveInterfaces("BOC", cycle=cycle)
         return self._interactAll("BOC", activeInterfaces, cycle)
 
-    def interactAllEveryNode(self, cycle, tn, excludedInterfaceNames=None):
+    def interactAllEveryNode(self, cycle, tn, excludedInterfaceNames=()):
         """
         Call the interactEveryNode hook for all enabled interfaces.
 
@@ -621,22 +608,12 @@ class Operator:
         excludedInterfaceNames : list, optional
             Names of interface names that will not be interacted with.
         """
-        excludedInterfaceNames = excludedInterfaceNames or ()
-        activeInterfaces = [
-            ii
-            for ii in self.interfaces
-            if ii.enabled() and ii.name not in excludedInterfaceNames
-        ]
+        activeInterfaces = self.getActiveInterfaces("EveryNode", excludedInterfaceNames)
         self._interactAll("EveryNode", activeInterfaces, cycle, tn)
 
-    def interactAllEOC(self, cycle, excludedInterfaceNames=None):
+    def interactAllEOC(self, cycle, excludedInterfaceNames=()):
         """Interact end of cycle for all enabled interfaces."""
-        excludedInterfaceNames = excludedInterfaceNames or ()
-        activeInterfaces = [
-            ii
-            for ii in self.interfaces
-            if ii.enabled() and ii.name not in excludedInterfaceNames
-        ]
+        activeInterfaces = self.getActiveInterfaces("EOC", excludedInterfaceNames)
         self._interactAll("EOC", activeInterfaces, cycle)
 
     def interactAllEOL(self):
@@ -648,11 +625,8 @@ class Operator:
         If the interfaces are flagged to be reversed at EOL, they are separated from the main stack and appended
         at the end in reverse order. This allows, for example, an interface that must run first to also run last.
         """
-        activeInterfaces = [ii for ii in self.interfaces if ii.enabled()]
-        interfacesAtEOL = [ii for ii in activeInterfaces if not ii.reverseAtEOL]
-        activeReverseInterfaces = [ii for ii in activeInterfaces if ii.reverseAtEOL]
-        interfacesAtEOL.extend(reversed(activeReverseInterfaces))
-        self._interactAll("EOL", interfacesAtEOL)
+        activeInterfaces = self.getActiveInterfaces("EOL")
+        self._interactAll("EOL", activeInterfaces)
 
     def interactAllCoupled(self, coupledIteration):
         """
@@ -670,8 +644,7 @@ class Operator:
             :id: I_ARMI_OPERATOR_PHYSICS1
             :implements: R_ARMI_OPERATOR_PHYSICS
         """
-        activeInterfaces = [ii for ii in self.interfaces if ii.enabled()]
-
+        activeInterfaces = self.getActiveInterfaces("Coupled")
         # Store the previous iteration values before calling interactAllCoupled
         # for each interface.
         for interface in activeInterfaces:
@@ -679,7 +652,6 @@ class Operator:
                 interface.coupler.storePreviousIterationValue(
                     interface.getTightCouplingValue()
                 )
-
         self._interactAll("Coupled", activeInterfaces, coupledIteration)
 
         return self._checkTightCouplingConvergence(activeInterfaces)
@@ -934,7 +906,13 @@ class Operator:
         return candidateI
 
     def interfaceIsActive(self, name):
-        """True if named interface exists and is active."""
+        """True if named interface exists and is enabled.
+
+        Notes
+        -----
+        This logic is significantly simpler that getActiveInterfaces. This logic only
+        touches the enabled() flag, but doesn't take into account the case settings.
+        """
         i = self.getInterface(name)
         return i and i.enabled()
 
@@ -952,6 +930,63 @@ class Operator:
         """
         return self.interfaces[:]
 
+    def getActiveInterfaces(
+        self,
+        interactState: str,
+        excludedInterfaceNames: Tuple[str] = (),
+        cycle: int = 0,
+    ):
+        """Retrieve the interfaces which are active for a given interaction state.
+
+        Parameters
+        ----------
+        interactState: str
+            A string dictating which interaction state the interfaces should be pulled for.
+        excludedInterfaceNames: Tuple[str]
+            A tuple of strings dictating which interfaces should be manually skipped.
+        cycle: int
+            The given cycle. 0 by default.
+
+        Returns
+        -------
+        activeInterfaces: List[Interfaces]
+            The interfaces deemed active for the given interactState.
+        """
+        # Validate the inputs
+        if excludedInterfaceNames is None:
+            excludedInterfaceNames = ()
+
+        if interactState not in ("BOL", "BOC", "EveryNode", "EOC", "EOL", "Coupled"):
+            raise ValueError(f"{interactState} is an unknown interaction state!")
+
+        # Ensure the interface is enabled.
+        enabled = lambda i: i.enabled()
+        if interactState == "BOL":
+            enabled = lambda i: i.enabled() or i.bolForce()
+
+        # Ensure the name of the interface isn't in some exclusion list.
+        nameCheck = lambda i: True
+        if interactState == "EveryNode" or interactState == "EOC":
+            nameCheck = lambda i: i.name not in excludedInterfaceNames
+        elif interactState == "BOC" and cycle < self.cs[CONF_DEFERRED_INTERFACES_CYCLE]:
+            nameCheck = lambda i: i.name not in self.cs[CONF_DEFERRED_INTERFACE_NAMES]
+        elif interactState == "BOL":
+            nameCheck = (
+                lambda i: i.name not in self.cs[CONF_DEFERRED_INTERFACE_NAMES]
+                and i.name not in excludedInterfaceNames
+            )
+
+        # Finally, find the active interfaces.
+        activeInterfaces = [i for i in self.interfaces if enabled(i) and nameCheck(i)]
+
+        # Special Case: At EOL we reverse the order of some interfaces.
+        if interactState == "EOL":
+            actInts = [ii for ii in activeInterfaces if not ii.reverseAtEOL]
+            actInts.extend(reversed([ii for ii in activeInterfaces if ii.reverseAtEOL]))
+            activeInterfaces = actInts
+
+        return activeInterfaces
+
     def reattach(self, r, cs=None):
         """Add links to globally-shared objects to this operator and all interfaces.
 


Test patch:
diff --git a/armi/operators/tests/test_operators.py b/armi/operators/tests/test_operators.py
index 0f17ea0a..581a2a92 100644
--- a/armi/operators/tests/test_operators.py
+++ b/armi/operators/tests/test_operators.py
@@ -35,6 +35,8 @@ from armi.settings.fwSettings.globalSettings import (
     CONF_TIGHT_COUPLING,
     CONF_CYCLES_SKIP_TIGHT_COUPLING_INTERACTION,
     CONF_TIGHT_COUPLING_SETTINGS,
+    CONF_DEFERRED_INTERFACE_NAMES,
+    CONF_DEFERRED_INTERFACES_CYCLE,
 )
 from armi.tests import mockRunLogs
 from armi.utils import directoryChangers
@@ -171,6 +173,45 @@ class OperatorTests(unittest.TestCase):
         self.assertTrue(self.o.interfaceIsActive("main"))
         self.assertFalse(self.o.interfaceIsActive("Fake-o"))
 
+    def test_getActiveInterfaces(self):
+        """Ensure that the right interfaces are returned for a given interaction state."""
+        self.o.cs[CONF_DEFERRED_INTERFACES_CYCLE] = 1
+        self.o.cs[CONF_DEFERRED_INTERFACE_NAMES] = ["history"]
+
+        # Test invalid inputs.
+        with self.assertRaises(ValueError):
+            self.o.getActiveInterfaces("notAnInterface")
+
+        # Test BOL
+        interfaces = self.o.getActiveInterfaces(
+            "BOL", excludedInterfaceNames=("xsGroups")
+        )
+        interfaceNames = [interface.name for interface in interfaces]
+        self.assertNotIn("xsGroups", interfaceNames)
+        self.assertNotIn("history", interfaceNames)
+
+        # Test BOC
+        interfaces = self.o.getActiveInterfaces("BOC", cycle=0)
+        interfaceNames = [interface.name for interface in interfaces]
+        self.assertNotIn("history", interfaceNames)
+
+        # Test EveryNode and EOC
+        interfaces = self.o.getActiveInterfaces(
+            "EveryNode", excludedInterfaceNames=("xsGroups")
+        )
+        interfaceNames = [interface.name for interface in interfaces]
+        self.assertIn("history", interfaceNames)
+        self.assertNotIn("xsGroups", interfaceNames)
+
+        # Test EOL
+        interfaces = self.o.getActiveInterfaces("EOL")
+        self.assertEqual(interfaces[-1].name, "main")
+
+        # Test Coupled
+        interfaces = self.o.getActiveInterfaces("Coupled")
+        for test, ref in zip(interfaces, self.activeInterfaces):
+            self.assertEqual(test.name, ref.name)
+
     def test_loadStateError(self):
         """The ``loadTestReactor()`` test tool does not have any history in the DB to load from."""
         # a first, simple test that this method fails correctly