Updates for pin-related functionality

armi/bookkeeping/db/tests/test_comparedb3.py

@@ -181,7 +181,7 @@ def test_compareDatabaseSim(self):
             dbs[1]._fullPath,
             timestepCompare=[(0, 0), (0, 1)],
         )
-        self.assertEqual(len(diffs.diffs), 477)
+        self.assertEqual(len(diffs.diffs), 501)
         # Cycle length is only diff (x3)
         self.assertEqual(diffs.nDiffs(), 3)
 

armi/bookkeeping/historyTracker.py

@@ -67,7 +67,7 @@ def getHistoryParams(self):
 See :ref:`detail-assems`.
 
 """
-from typing import Tuple
+from typing import TYPE_CHECKING
 import traceback
 
 from armi import interfaces
@@ -79,6 +79,10 @@ def getHistoryParams(self):
 
 ORDER = 2 * interfaces.STACK_ORDER.BEFORE + interfaces.STACK_ORDER.BOOKKEEPING
 
+if TYPE_CHECKING:
+    from armi.reactor.blocks import Block
+    from armi.reactor.assemblies import Assembly
+
 
 def describeInterfaces(cs):
     """Function for exposing interface(s) to other code."""
@@ -120,6 +124,8 @@ class HistoryTrackerInterface(interfaces.Interface):
 
     name = "history"
 
+    DETAILED_ASSEMBLY_FLAGS = [Flags.FUEL, Flags.CONTROL]
+
     def __init__(self, r, cs):
         """
         HistoryTracker that uses the database to look up parameter history rather than
@@ -146,7 +152,7 @@ def interactBOC(self, cycle=None):
         """Look for any new assemblies that are asked for and add them to tracking."""
         self.addDetailAssemsByAssemNums()
         if self.cs["detailAllAssems"]:
-            self.addAllFuelAssems()
+            self.addAllDetailedAssems()
 
     def interactEOL(self):
         """Generate the history reports."""
@@ -171,16 +177,16 @@ def addDetailAssembliesBOL(self):
                 self.addDetailAssembly(a)
 
         if self.cs["detailAllAssems"]:
-            self.addAllFuelAssems()
+            self.addAllDetailedAssems()
 
         # This also gets called at BOC but we still
         # do it here for operators that do not call BOC.
         self.addDetailAssemsByAssemNums()
 
-    def addAllFuelAssems(self):
-        """Add all fuel assems as detail assems."""
+    def addAllDetailedAssems(self):
+        """Add all assems who have the DETAILED_ASSEMBLY_FLAGS as detail assems."""
         for a in self.r.core:
-            if a.hasFlags(Flags.FUEL):
+            if a.hasFlags(self.DETAILED_ASSEMBLY_FLAGS):
                 self.addDetailAssembly(a)
 
     def addDetailAssemsByAssemNums(self):
@@ -236,13 +242,13 @@ def getTrackedParams(self):
                     trackedParams.add(newParam)
         return sorted(trackedParams)
 
-    def addDetailAssembly(self, a):
+    def addDetailAssembly(self, a: "Assembly"):
         """Track the name of assemblies that are flagged for detailed treatment."""
         aName = a.getName()
         if aName not in self.detailAssemblyNames:
             self.detailAssemblyNames.append(aName)
 
-    def getDetailAssemblies(self):
+    def getDetailAssemblies(self) -> list["Assembly"]:
         """Returns the assemblies that have been signaled as detail assemblies."""
         assems = []
         if not self.detailAssemblyNames:
@@ -259,7 +265,7 @@ def getDetailAssemblies(self):
                 )
         return assems
 
-    def getDetailBlocks(self):
+    def getDetailBlocks(self) -> list["Block"]:
         """Get all blocks in all detail assemblies."""
         return [block for a in self.getDetailAssemblies() for block in a]
 
@@ -281,7 +287,7 @@ def filterTimeIndices(self, timeIndices, boc=False, moc=False, eoc=False):
 
         return filtered
 
-    def writeAssemHistory(self, a, fName=""):
+    def writeAssemHistory(self, a: "Assembly", fName: str = ""):
         """Write the assembly history report to a text file."""
         fName = fName or self._getAssemHistoryFileName(a)
         dbi = self.getInterface("database")
@@ -373,7 +379,7 @@ def unloadBlockHistoryVals(self):
         """Remove all cached db reads."""
         self._preloadedBlockHistory = None
 
-    def getBlockHistoryVal(self, name: str, paramName: str, ts: Tuple[int, int]):
+    def getBlockHistoryVal(self, name: str, paramName: str, ts: tuple[int, int]):
         """
         Use the database interface to return the parameter values for the supplied block
         names, and timesteps.
@@ -422,28 +428,28 @@ def getBlockHistoryVal(self, name: str, paramName: str, ts: Tuple[int, int]):
                 raise
         return val
 
-    def _isCurrentTimeStep(self, ts: Tuple[int, int]):
+    def _isCurrentTimeStep(self, ts: tuple[int, int]) -> bool:
         """Return True if the timestep requested is the current time step."""
         return ts == (self.r.p.cycle, self.r.p.timeNode)
 
-    def _databaseHasDataForTimeStep(self, ts):
+    def _databaseHasDataForTimeStep(self, ts) -> bool:
         """Return True if the database has data for the requested time step."""
         dbi = self.getInterface("database")
         return ts in dbi.database.genTimeSteps()
 
-    def getTimeSteps(self, a=None):
-        r"""
-        Return list of time steps values (in years) that are available.
+    def getTimeSteps(self, a: "Assembly" = None) -> list[float]:
+        """
+        Given a fuel assembly, return list of time steps values (in years) that are available.
 
         Parameters
         ----------
-        a : Assembly object, optional
-            An assembly object designated a detail assem. If passed, only timesteps
+        a
+            A fuel assembly that has been designated a detail assem. If passed, only timesteps
             where this assembly is in the core will be tracked.
 
         Returns
         -------
-        timeSteps : list
+        timeSteps
             times in years that are available in the history
 
         See Also
@@ -462,15 +468,13 @@ def getTimeSteps(self, a=None):
         return timeInYears
 
     @staticmethod
-    def _getBlockInAssembly(a):
-        """Get a representative fuel block from an assembly."""
+    def _getBlockInAssembly(a: "Assembly") -> "Block":
+        """Get a representative fuel block from a fuel assembly."""
         b = a.getFirstBlock(Flags.FUEL)
         if not b:
-            # there is a problem, it doesn't look like we have a fueled assembly
-            # but that is all we track... what is it? Throw an error
-            runLog.warning("Assembly {} does not contain fuel".format(a))
+            runLog.error("Assembly {} does not contain fuel".format(a))
             for b in a:
-                runLog.warning("Block {}".format(b))
+                runLog.error("Block {}".format(b))
             raise RuntimeError(
                 "A tracked assembly does not contain fuel and has caused this error, see the details in stdout."
             )

armi/bookkeeping/tests/test_historyTracker.py

@@ -231,8 +231,8 @@ def test_historyReport(self):
 
         # test that detailAssemblyNames() is working
         self.assertEqual(len(history.detailAssemblyNames), 1)
-        history.addAllFuelAssems()
-        self.assertEqual(len(history.detailAssemblyNames), 51)
+        history.addAllDetailedAssems()
+        self.assertEqual(len(history.detailAssemblyNames), 54)
 
     def test_getBlockInAssembly(self):
         history = self.o.getInterface("history")

armi/physics/fuelCycle/tests/test_assemblyRotationAlgorithms.py

@@ -37,7 +37,6 @@ def test_buReducingAssemblyRotation(self):
 
         # apply dummy pin-level data to allow intelligent rotation
         for b in assem.getBlocks(Flags.FUEL):
-            b.breakFuelComponentsIntoIndividuals()
             b.initializePinLocations()
             b.p.percentBuMaxPinLocation = 10
             b.p.percentBuMax = 5

armi/reactor/blocks.py

@@ -29,7 +29,6 @@
 from armi import nuclideBases
 from armi import runLog
 from armi.bookkeeping import report
-from armi.nucDirectory import elements
 from armi.nuclearDataIO import xsCollections
 from armi.physics.neutronics import GAMMA
 from armi.physics.neutronics import NEUTRON
@@ -794,9 +793,16 @@ def completeInitialLoading(self, bolBlock=None):
         for child in self:
             hmMass = child.getHMMass() * sf
             massHmBOL += hmMass
-            # Components have a massHmBOL parameter but not every composite will
+            # Components have the following parameters but not every composite will
+            # massHmBOL, molesHmBOL, puFrac
             if isinstance(child, components.Component):
                 child.p.massHmBOL = hmMass
+                child.p.molesHmBOL = child.getHMMoles()
+                child.p.puFrac = (
+                    self.getPuMoles() / child.p.molesHmBOL
+                    if child.p.molesHmBOL > 0.0
+                    else 0.0
+                )
 
         self.p.massHmBOL = massHmBOL
 
@@ -1430,63 +1436,6 @@ def updateComponentDims(self):
             except NotImplementedError:
                 runLog.warning("{0} has no updatedDims method -- skipping".format(c))
 
-    def breakFuelComponentsIntoIndividuals(self):
-        """
-        Split block-level components (in fuel blocks) into pin-level components.
-
-        The fuel component will be broken up according to its multiplicity.
-
-        Order matters! The first pin component will be located at a particular (x, y), which
-        will be used in the fluxRecon module to determine the interpolated flux.
-
-        The fuel will become fuel001 through fuel169 if there are 169 pins.
-        """
-        fuels = self.getChildrenWithFlags(Flags.FUEL)
-        if len(fuels) != 1:
-            runLog.error(
-                "This block contains {0} fuel components: {1}".format(len(fuels), fuels)
-            )
-            raise RuntimeError(
-                "Cannot break {0} into multiple fuel components b/c there is not a single fuel"
-                " component.".format(self)
-            )
-
-        fuel = fuels[0]
-        fuelFlags = fuel.p.flags
-        nPins = self.getNumPins()
-        runLog.info(
-            "Creating {} individual {} components on {}".format(nPins, fuel, self)
-        )
-
-        # Handle all other components that may be linked to the fuel multiplicity
-        # by unlinking them and setting them directly.
-        # TODO: What about other (actual) dimensions? This is a limitation in that only fuel
-        # components are duplicated, and not the entire pin. It is also a reasonable assumption with
-        # current/historical usage of ARMI.
-        for comp, dim in self.getComponentsThatAreLinkedTo(fuel, "mult"):
-            comp.setDimension(dim, nPins)
-
-        # finish the first pin as a single pin
-        fuel.setDimension("mult", 1)
-        fuel.setName("fuel001")
-        fuel.p.pinNum = 1
-
-        # create all the new pin components and add them to the block with 'fuel001' names
-        for i in range(nPins - 1):
-            # wow, only use of a non-deepcopy
-            newC = copy.copy(fuel)
-            newC.setName("fuel{0:03d}".format(i + 2))  # start with 002.
-            newC.p.pinNum = i + 2
-            self.add(newC)
-
-        # update moles at BOL for each pin
-        self.p.molesHmBOLByPin = []
-        for pin in self.iterComponents(Flags.FUEL):
-            # Update the fuel component flags to be the same as before the split (i.e., DEPLETABLE)
-            pin.p.flags = fuelFlags
-            self.p.molesHmBOLByPin.append(pin.getHMMoles())
-            pin.p.massHmBOL /= nPins
-
     def getIntegratedMgFlux(self, adjoint=False, gamma=False):
         """
         Return the volume integrated multigroup neutron tracklength in [n-cm/s].
@@ -1737,18 +1686,6 @@ def getBoronMassEnrich(self):
             return 0.0
         return b10 / total
 
-    def getPuMoles(self):
-        """Returns total number of moles of Pu isotopes."""
-        nucNames = [nuc.name for nuc in elements.byZ[94].nuclides]
-        puN = sum(self.getNuclideNumberDensities(nucNames))
-
-        return (
-            puN
-            / units.MOLES_PER_CC_TO_ATOMS_PER_BARN_CM
-            * self.getVolume()
-            * self.getSymmetryFactor()
-        )
-
     def getUraniumMassEnrich(self):
         """Returns U-235 mass fraction assuming U-235 and U-238 only."""
         u5 = self.getMass("U235")

armi/reactor/components/component.py

@@ -21,6 +21,7 @@
 import re
 
 import numpy as np
+from typing import Optional
 
 from armi import materials
 from armi import runLog
@@ -31,6 +32,7 @@
 from armi.nucDirectory import nuclideBases
 from armi.reactor import composites
 from armi.reactor import flags
+from armi.reactor import grids
 from armi.reactor import parameters
 from armi.reactor.components import componentParameters
 from armi.utils import densityTools
@@ -1262,6 +1264,71 @@ def getIntegratedMgFlux(self, adjoint=False, gamma=False):
 
         return pinFluxes[self.p.pinNum - 1] * self.getVolume()
 
+    def getPinMgFluxes(
+        self, adjoint: Optional[bool] = False, gamma: Optional[bool] = False
+    ) -> np.ndarray:
+        """Retrieves the pin multigroup fluxes for the component.
+
+        Parameters
+        ----------
+        adjoint : bool, optional
+            Return adjoint flux instead of real
+        gamma : bool, optional
+            Whether to return the neutron flux or the gamma flux.
+
+        Returns
+        -------
+        np.ndarray
+            A ``(N, nGroup)`` array of pin multigroup fluxes, where ``N`` is the
+            equivalent to the multiplicity of the component (``self.p.mult``)
+            and ``nGroup`` is the number of energy groups of the flux.
+
+        Raises
+        ------
+        ValueError
+            If the location(s) of the component are not aligned with pin indices
+            from the block. This would happen if this component is not actually
+            a pin.
+        """
+        # Get the (i, j, k) location of all pins from the parent block
+        indicesAll = {
+            (loc.i, loc.j): i for i, loc in enumerate(self.parent.getPinLocations())
+        }
+
+        # Retrieve the indices of this component
+        if isinstance(self.spatialLocator, grids.MultiIndexLocation):
+            indices = [(loc.i, loc.j) for loc in self.spatialLocator]
+        else:
+            indices = [(self.spatialLocator.i, self.spatialLocator.j)]
+
+        # Map this component's indices to block's pin indices
+        indexMap = list(map(indicesAll.get, indices))
+        if None in indexMap:
+            msg = f"Failed to retrieve pin indices for component {self}."
+            runLog.error(msg)
+            raise ValueError(msg)
+
+        # Get the parameter name we are trying to retrieve
+        if gamma:
+            if adjoint:
+                raise ValueError("Adjoint gamma flux is currently unsupported.")
+            else:
+                param = "pinMgFluxesGamma"
+        else:
+            if adjoint:
+                param = "pinMgFluxesAdj"
+            else:
+                param = "pinMgFluxes"
+
+        # Return pin fluxes
+        try:
+            return self.parent.p[param][indexMap]
+        except Exception as ee:
+            msg = f"Failure getting {param} from {self} via parent {self.parent}"
+            runLog.error(msg)
+            runLog.error(ee)
+            raise ValueError(msg) from ee
+
     def density(self) -> float:
         """Returns the mass density of the object in g/cc."""
         density = composites.Composite.density(self)