Repository: terrapower/armi
Pull request #1307 (2023-06-13 18:30:05)

Adding unit tests to the LatticePhysicsWriter
## What is the change?

I am adding unit tests to the `LatticePhysicsWriter` class.

## Why is the change being made?

This was requested by @opotowsky in ticket #1303

---

## 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!
    Learn what a "good PR" looks like here:
    https://terrapower.github.io/armi/developer/tooling.html#good-pull-requests
-->

- [X] This PR has only one purpose or idea.
- [X] Tests 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 is still up-to-date in the `doc` folder.
- [X] The dependencies are still up-to-date in `setup.py`.



Patch:
diff --git a/armi/bookkeeping/snapshotInterface.py b/armi/bookkeeping/snapshotInterface.py
index 097f922f..722ce29c 100644
--- a/armi/bookkeeping/snapshotInterface.py
+++ b/armi/bookkeeping/snapshotInterface.py
@@ -47,15 +47,27 @@ class SnapshotInterface(interfaces.Interface):
     name = "snapshot"
 
     def interactBOL(self):
+        """Active the default snapshots at BOL"""
         interfaces.Interface.interactBOL(self)
         if self.cs["defaultSnapshots"]:
             self.activateDefaultSnapshots()
 
     def interactEveryNode(self, cycle, node):
+        """
+        Call the snapshot interface to copy files at each node, if requested
+        """
         snapText = getCycleNodeStamp(cycle, node)  # CCCNNN
         if self.cs["dumpSnapshot"] and snapText in self.cs["dumpSnapshot"]:
             self.o.snapshotRequest(cycle, node)
 
+    def interactCoupled(self, iteration):
+        """
+        Call the snapshot interface to copy files for coupled iterations, if requested
+        """
+        snapText = getCycleNodeStamp(self.r.p.cycle, self.r.p.timeNode)  # CCCNNN
+        if self.cs["dumpSnapshot"] and snapText in self.cs["dumpSnapshot"]:
+            self.o.snapshotRequest(self.r.p.cycle, self.r.p.timeNode, iteration)
+
     def activateDefaultSnapshots(self):
         """Figure out and assign some default snapshots (BOL, MOL, EOL)."""
         if self.cs["runType"] == operators.RunTypes.EQUILIBRIUM:
diff --git a/armi/operators/operator.py b/armi/operators/operator.py
index c80334ab..c1b0e599 100644
--- a/armi/operators/operator.py
+++ b/armi/operators/operator.py
@@ -41,6 +41,9 @@ from armi.bookkeeping.report import reportingUtils
 from armi.operators import settingsValidation
 from armi.operators.runTypes import RunTypes
 from armi.physics.fuelCycle.settings import CONF_SHUFFLE_LOGIC
+from armi.physics.neutronics.globalFlux.globalFluxInterface import (
+    GlobalFluxInterfaceUsingExecuters,
+)
 from armi.settings.fwSettings.globalSettings import (
     CONF_TIGHT_COUPLING,
     CONF_TIGHT_COUPLING_MAX_ITERS,
@@ -1072,7 +1075,7 @@ class Operator:  # pylint: disable=too-many-public-methods
 
         dbi.loadState(cycle, timeNode, timeStepName, fileName)
 
-    def snapshotRequest(self, cycle, node):
+    def snapshotRequest(self, cycle, node, iteration=None):
         """
         Process a snapshot request at this time.
 
@@ -1112,28 +1115,40 @@ class Operator:  # pylint: disable=too-many-public-methods
         for fileName in os.listdir("."):
             if "mcc" in fileName and re.search(r"[A-Z]AF?\d?.inp", fileName):
                 base, ext = os.path.splitext(fileName)
+                if iteration is not None:
+                    newFile = "{0}_{1:03d}_{2:d}_{4}{3}".format(
+                        base, cycle, node, ext, iteration
+                    )
+                else:
+                    newFile = "{0}_{1:03d}_{2:d}{3}".format(base, cycle, node, ext)
                 # add the cycle and timenode to the XS input file names so that a rx-coeff case that runs
                 # in here won't overwrite them.
-                shutil.copy(
-                    fileName,
-                    os.path.join(
-                        newFolder, "{0}_{1:03d}_{2:d}{3}".format(base, cycle, node, ext)
-                    ),
-                )
+                shutil.copy(fileName, os.path.join(newFolder, newFile))
             if "rzmflx" in fileName:
                 pathTools.copyOrWarn("rzmflx for snapshot", fileName, newFolder)
 
-        isoFName = f"ISOTXS-c{cycle}n{node}"
-        if not os.path.exists(isoFName):
-            isoFName = f"ISOTXS-c{cycle}"
+        fileNamePossibilities = [
+            f"ISOTXS-c{cycle}n{node}",
+            f"ISOTXS-c{cycle}",
+        ]
+        if iteration is not None:
+            fileNamePossibilities = [
+                f"ISOTXS-c{cycle}n{node}i{iteration}"
+            ] + fileNamePossibilities
+
+        for isoFName in fileNamePossibilities:
+            if os.path.exists(isoFName):
+                break
         pathTools.copyOrWarn(
             "ISOTXS for snapshot", isoFName, pathTools.armiAbsPath(newFolder, "ISOTXS")
         )
-        pathTools.copyOrWarn(
-            "DIF3D output for snapshot",
-            self.cs.caseTitle + "{0:03d}.out".format(cycle),
-            newFolder,
+        globalFluxLabel = GlobalFluxInterfaceUsingExecuters.getLabel(
+            self.cs.caseTitle, cycle, node, iteration
         )
+        globalFluxInput = globalFluxLabel + ".inp"
+        globalFluxOutput = globalFluxLabel + ".out"
+        pathTools.copyOrWarn("DIF3D input for snapshot", globalFluxInput, newFolder)
+        pathTools.copyOrWarn("DIF3D output for snapshot", globalFluxOutput, newFolder)
         pathTools.copyOrWarn(
             "Shuffle logic for snapshot", self.cs[CONF_SHUFFLE_LOGIC], newFolder
         )
diff --git a/doc/release/0.2.rst b/doc/release/0.2.rst
index cd36edfc..0e683a7c 100644
--- a/doc/release/0.2.rst
+++ b/doc/release/0.2.rst
@@ -11,7 +11,7 @@ What's new in ARMI
 #. Calculate weighted-average percent burnup of BlockCollections. (`PR#1265 <https://github.com/terrapower/armi/pull/1265>`_)
 #. Added ``Composite.sort()`` to allow the user to recursively sort any piece of the ``Reactor``. (`PR#1280 <https://github.com/terrapower/armi/pull/1280>`_)
 #  Revert "when using non-uniform mesh, detailed fission/activation products have cross sections generated to avoid blocks without xs data" and move capability to lattice physics (`PR#1280 <https://github.com/terrapower/armi/pull/1298>`_)
-#. TBD
+#. Add interactCoupled method for snapshotInterface (`PR#1294 <https://github.com/terrapower/armi/pull/1294>`_)
 
 Bug fixes
 ---------


Test patch:
diff --git a/armi/bookkeeping/tests/test_snapshot.py b/armi/bookkeeping/tests/test_snapshot.py
index 17243320..80fb7b13 100644
--- a/armi/bookkeeping/tests/test_snapshot.py
+++ b/armi/bookkeeping/tests/test_snapshot.py
@@ -15,9 +15,23 @@
 """Test Snapshots."""
 # pylint: disable=missing-function-docstring,missing-class-docstring,protected-access,invalid-name,no-self-use,no-method-argument,import-outside-toplevel
 import unittest
+from unittest.mock import patch
 
 from armi.bookkeeping import snapshotInterface
 from armi import settings
+from armi.operators.operator import Operator
+
+
+class MockReactor:
+    def __init__(self, cs):
+        self.p = MockReactorParams()
+        self.o = Operator(cs)
+
+
+class MockReactorParams:
+    def __init__(self):
+        self.cycle = 0
+        self.timeNode = 1
 
 
 class TestSnapshotInterface(unittest.TestCase):
@@ -27,7 +41,23 @@ class TestSnapshotInterface(unittest.TestCase):
 
     def setUp(self):
         self.cs.revertToDefaults()
-        self.si = snapshotInterface.SnapshotInterface(None, self.cs)
+        self.si = snapshotInterface.SnapshotInterface(MockReactor(self.cs), self.cs)
+
+    @patch("armi.operators.operator.Operator.snapshotRequest")
+    def test_interactEveryNode(self, mockSnapshotRequest):
+        newSettings = {}
+        newSettings["dumpSnapshot"] = ["000001"]
+        self.si.cs = self.si.cs.modified(newSettings=newSettings)
+        self.si.interactEveryNode(0, 1)
+        self.assertTrue(mockSnapshotRequest.called)
+
+    @patch("armi.operators.operator.Operator.snapshotRequest")
+    def test_interactCoupled(self, mockSnapshotRequest):
+        newSettings = {}
+        newSettings["dumpSnapshot"] = ["000001"]
+        self.si.cs = self.si.cs.modified(newSettings=newSettings)
+        self.si.interactCoupled(2)
+        self.assertTrue(mockSnapshotRequest.called)
 
     def test_activeateDefaultSnapshots_30cycles2BurnSteps(self):
         self.assertEqual([], self.cs["dumpSnapshot"])
diff --git a/armi/operators/tests/test_operators.py b/armi/operators/tests/test_operators.py
index 85a3241c..96f08524 100644
--- a/armi/operators/tests/test_operators.py
+++ b/armi/operators/tests/test_operators.py
@@ -17,6 +17,7 @@
 # pylint: disable=missing-function-docstring,missing-class-docstring,protected-access,invalid-name,no-method-argument,import-outside-toplevel
 import os
 import unittest
+from unittest.mock import patch
 import collections
 
 from armi import settings
@@ -123,12 +124,22 @@ class OperatorTests(unittest.TestCase):
         cs = self.o.setStateToDefault(self.o.cs)
         self.assertEqual(cs[CONF_RUN_TYPE], "Standard")
 
-    def test_snapshotRequest(self):
+    @patch("shutil.copy")
+    @patch("os.listdir")
+    def test_snapshotRequest(self, fakeDirList, fakeCopy):
+        fakeDirList.return_value = ["mccAA.inp"]
         with TemporaryDirectoryChanger():
             with mockRunLogs.BufferLog() as mock:
                 self.o.snapshotRequest(0, 1)
                 self.assertIn("ISOTXS-c0", mock.getStdout())
-                self.assertIn("DIF3D output for snapshot", mock.getStdout())
+                self.assertIn(
+                    "DIF3D input for snapshot: armiRun-flux-c0n1.inp",
+                    mock.getStdout(),
+                )
+                self.assertIn(
+                    "DIF3D output for snapshot: armiRun-flux-c0n1.out",
+                    mock.getStdout(),
+                )
                 self.assertIn("Shuffle logic for snapshot", mock.getStdout())
                 self.assertIn("Geometry file for snapshot", mock.getStdout())
                 self.assertIn("Loading definition for snapshot", mock.getStdout())
@@ -136,6 +147,25 @@ class OperatorTests(unittest.TestCase):
                 self.assertIn("Pressure history for snapshot", mock.getStdout())
             self.assertTrue(os.path.exists("snapShot0_1"))
 
+        with TemporaryDirectoryChanger():
+            with mockRunLogs.BufferLog() as mock:
+                self.o.snapshotRequest(0, 2, iteration=1)
+                self.assertIn("ISOTXS-c0", mock.getStdout())
+                self.assertIn(
+                    "DIF3D input for snapshot: armiRun-flux-c0n2i1.inp",
+                    mock.getStdout(),
+                )
+                self.assertIn(
+                    "DIF3D output for snapshot: armiRun-flux-c0n2i1.out",
+                    mock.getStdout(),
+                )
+                self.assertIn("Shuffle logic for snapshot", mock.getStdout())
+                self.assertIn("Geometry file for snapshot", mock.getStdout())
+                self.assertIn("Loading definition for snapshot", mock.getStdout())
+                self.assertIn("Flow history for snapshot", mock.getStdout())
+                self.assertIn("Pressure history for snapshot", mock.getStdout())
+            self.assertTrue(os.path.exists("snapShot0_2"))
+
 
 class TestTightCoupling(unittest.TestCase):
     def setUp(self):
diff --git a/armi/physics/neutronics/latticePhysics/tests/test_latticeWriter.py b/armi/physics/neutronics/latticePhysics/tests/test_latticeWriter.py
index 0fe1bedb..bf3c4f09 100644
--- a/armi/physics/neutronics/latticePhysics/tests/test_latticeWriter.py
+++ b/armi/physics/neutronics/latticePhysics/tests/test_latticeWriter.py
@@ -14,9 +14,13 @@
 
 """Test the Lattice Physics Writer."""
 # pylint: disable=missing-function-docstring,missing-class-docstring,protected-access,invalid-name,no-self-use,no-method-argument,import-outside-toplevel
+from collections import defaultdict
 import unittest
 
 from armi.physics.neutronics.const import CONF_CROSS_SECTION
+from armi.physics.neutronics.fissionProductModel.fissionProductModelSettings import (
+    CONF_FP_MODEL,
+)
 from armi.physics.neutronics.latticePhysics.latticePhysicsWriter import (
     LatticePhysicsWriter,
 )
@@ -53,32 +57,129 @@ class FakeLatticePhysicsWriter(LatticePhysicsWriter):
 class TestLatticePhysicsWriter(unittest.TestCase):
     """Test Lattice Physics Writer."""
 
-    def test_LatticePhysicsWriter(self):
-        """Super basic test of the LatticePhysicsWriter."""
-        o, r = loadTestReactor(TEST_ROOT)
-        cs = o.cs
-        o.cs[CONF_CROSS_SECTION].setDefaults(
-            cs[CONF_XS_BLOCK_REPRESENTATION],
-            cs[CONF_DISABLE_BLOCK_TYPE_EXCLUSION_IN_XS_GENERATION],
+    def setUp(self):
+        self.o, self.r = loadTestReactor(TEST_ROOT)
+        self.cs = self.o.cs
+        self.cs[CONF_CROSS_SECTION].setDefaults(
+            self.cs[CONF_XS_BLOCK_REPRESENTATION],
+            self.cs[CONF_DISABLE_BLOCK_TYPE_EXCLUSION_IN_XS_GENERATION],
         )
-        block = r.core.getFirstBlock()
-        w = FakeLatticePhysicsWriter(block, r, o)
-
-        self.assertEqual(w.xsId, "AA")
-        self.assertFalse(w.modelFissionProducts)
-        self.assertEqual(w.driverXsID, "")
-        self.assertAlmostEqual(w.minimumNuclideDensity, 1e-15, delta=1e-16)
+        self.block = self.r.core.getFirstBlock()
+        self.w = FakeLatticePhysicsWriter(self.block, self.r, self.o)
 
-        self.assertEqual(w.testOut, "")
-        self.assertEqual(str(w), "<FakeLatticePhysicsWriter - XS ID AA (Neutron XS)>")
+    def test_latticePhysicsWriter(self):
+        """Super basic test of the LatticePhysicsWriter."""
+        self.assertEqual(self.w.xsId, "AA")
+        self.assertFalse(self.w.modelFissionProducts)
+        self.assertEqual(self.w.driverXsID, "")
+        self.assertAlmostEqual(self.w.minimumNuclideDensity, 1e-15, delta=1e-16)
+
+        self.assertEqual(self.w.testOut, "")
+        self.assertEqual(
+            str(self.w), "<FakeLatticePhysicsWriter - XS ID AA (Neutron XS)>"
+        )
 
-        w._writeTitle(None)
-        self.assertIn("ARMI generated case for caseTitle armiRun", w.testOut)
+        self.w._writeTitle(None)
+        self.assertIn("ARMI generated case for caseTitle armiRun", self.w.testOut)
 
-        nucs = w._getAllNuclidesByTemperatureInC(None)
+        nucs = self.w._getAllNuclidesByTemperatureInC(None)
         self.assertEqual(len(nucs.keys()), 1)
         self.assertAlmostEqual(list(nucs.keys())[0], 450.0, delta=0.1)
 
+    def test_writeTitle(self):
+        self.w._writeTitle("test_writeTitle")
+        self.assertIn("ARMI generated case for caseTitle", self.w.testOut)
+
+    def test_isSourceDriven(self):
+        self.assertFalse(self.w._isSourceDriven)
+        self.w.driverXsID = True
+        self.assertTrue(self.w._isSourceDriven)
+
+    def test_isGammaXSGenerationEnabled(self):
+        self.assertFalse(self.w._isGammaXSGenerationEnabled)
+
+    def test_getAllNuclidesByTemperatureInCNone(self):
+        nucsByTemp = self.w._getAllNuclidesByTemperatureInC(None)
+        keys0 = list(nucsByTemp.keys())
+        self.assertEqual(len(keys0), 1)
+        self.assertEqual(keys0[0], 450.0)
+        keys1 = nucsByTemp[keys0[0]]
+        self.assertGreater(len(keys1), 1)
+        names = [k.name for k in keys1]
+        self.assertIn("AM241", names)
+        self.assertIn("U238", names)
+
+    def test_getAllNuclidesByTemperatureInC(self):
+        self.w.explicitFissionProducts = False
+        c = self.r.core[0][0]
+        nucsByTemp = self.w._getAllNuclidesByTemperatureInC(c)
+        keys0 = list(nucsByTemp.keys())
+        self.assertEqual(len(keys0), 1)
+        self.assertEqual(keys0[0], 450.0)
+        keys1 = nucsByTemp[keys0[0]]
+        self.assertGreater(len(keys1), 1)
+        names = [k.name for k in keys1]
+        self.assertIn("AM241", names)
+        self.assertIn("U238", names)
+
+    def test_getAllNuclidesByTemperatureInCExplicitFissionProducts(self):
+        self.w.explicitFissionProducts = True
+        c = self.r.core[0][0]
+        nucsByTemp = self.w._getAllNuclidesByTemperatureInC(c)
+        keys0 = list(nucsByTemp.keys())
+        self.assertEqual(len(keys0), 1)
+        self.assertEqual(keys0[0], 450.0)
+        keys1 = nucsByTemp[keys0[0]]
+        self.assertGreater(len(keys1), 1)
+        names = [k.name for k in keys1]
+        self.assertIn("AM241", names)
+        self.assertIn("U238", names)
+
+    def test_getAvgNuclideTemperatureInC(self):
+        temp = self.w._getAvgNuclideTemperatureInC("U238")
+        self.assertAlmostEqual(temp, 450, delta=0.001)
+
+        temp = self.w._getAvgNuclideTemperatureInC("U235")
+        self.assertAlmostEqual(temp, 450, delta=0.001)
+
+    def test_getFuelTemperature(self):
+        temp = self.w._getFuelTemperature()
+        self.assertAlmostEqual(temp, 450, delta=0.001)
+
+    def test_getDetailedFissionProducts(self):
+        dfpDen = defaultdict(int)
+        dfpDen["U238"] = 1.2
+        dfpDen["U235"] = 2.3
+        dfpDen["AM241"] = 3.4
+        prods = self.w._getDetailedFissionProducts(dfpDen)
+        self.assertEqual(len(prods), 3)
+        self.assertIn("U238", prods)
+        self.assertIn("U235", prods)
+        self.assertIn("AM241", prods)
+
+    def test_getDetailedFissionProductsPass(self):
+        self.cs[CONF_FP_MODEL] = "noFissionProducts"
+
+        prods = self.w._getDetailedFissionProducts({})
+        self.assertEqual(len(prods), 0)
+
+    def test_getDetailedFPDensities(self):
+        self.w.modelFissionProducts = False
+        dens = self.w._getDetailedFPDensities()
+        self.assertEqual(len(dens), 0)
+
+        self.w.modelFissionProducts = True
+        with self.assertRaises(AttributeError):
+            dens = self.w._getDetailedFPDensities()
+
+    def test_isCriticalBucklingSearchActive(self):
+        isActive = self.w._isCriticalBucklingSearchActive
+        self.assertTrue(isActive)
+
+    def test_getDriverBlock(self):
+        b = self.w._getDriverBlock()
+        self.assertIsNone(b)
+
 
 if __name__ == "__main__":
     unittest.main()