12776996362

Committed 14 Jan 2025 09:49PM UTC coverage: 84.938% (+0.2%) from 84.729%

Build # 12776996362

Build Type

Pull #3133

github

Committed by

web-flow

Commit Message

Merge 0495246d9 into 549cc0973

Pull Request Pull Request #3133: Kinetics parameters using Iterated Fission Probability

Run Details

318 of 330 new or added lines in 10 files covered. (96.36%)

1658 existing lines in 66 files now uncovered.

50402 of 59340 relevant lines covered (84.94%)

33987813.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.53

/openmc/deplete/openmc_operator.py

"""OpenMC transport operator

This module implements functions shared by both OpenMC transport-coupled and
transport-independent transport operators.

"""

from abc import abstractmethod
from warnings import warn

import numpy as np

import openmc
from openmc.checkvalue import check_value
from openmc.exceptions import DataError
from openmc.mpi import comm
from .abc import TransportOperator, OperatorResult
from .atom_number import AtomNumber
from .reaction_rates import ReactionRates
from .pool import _distribute

__all__ = ["OpenMCOperator", "OperatorResult"]


class OpenMCOperator(TransportOperator):
    """Abstract class holding OpenMC-specific functions for running
    depletion calculations.

    Specific classes for running transport-coupled or transport-independent
    depletion calculations are implemented as subclasses of OpenMCOperator.

    Parameters
    ----------
    materials : openmc.Materials
        List of all materials in the model
    cross_sections : str or list of MicroXS
        Path to continuous energy cross section library, or list of objects
        containing cross sections.
    chain_file : str, optional
        Path to the depletion chain XML file. Defaults to
        openmc.config['chain_file'].
    prev_results : Results, optional
        Results from a previous depletion calculation. If this argument is
        specified, the depletion calculation will start from the latest state
        in the previous results.
    diff_burnable_mats : bool, optional
        Whether to differentiate burnable materials with multiple instances.
    fission_q : dict, optional
        Dictionary of nuclides and their fission Q values [eV].
    helper_kwargs : dict
        Keyword arguments for helper classes
    reduce_chain : bool, optional
        If True, use :meth:`openmc.deplete.Chain.reduce()` to reduce the
        depletion chain up to ``reduce_chain_level``.
    reduce_chain_level : int, optional
        Depth of the search when reducing the depletion chain. Only used
        if ``reduce_chain`` evaluates to true. The default value of
        ``None`` implies no limit on the depth.

    diff_volume_method : str
        Specifies how the volumes of the new materials should be found. Default
        is to 'divide equally' which divides the original material volume
        equally between the new materials, 'match cell' sets the volume of the
        material to volume of the cell they fill.

        .. versionadded:: 0.14.0

    Attributes
    ----------
    materials : openmc.Materials
        All materials present in the model
    cross_sections : str or list of MicroXS
        Path to continuous energy cross section library, or list of objects
        containing cross sections.
    output_dir : pathlib.Path
        Path to output directory to save results.
    round_number : bool
        Whether or not to round output to OpenMC to 8 digits.
        Useful in testing, as OpenMC is incredibly sensitive to exact values.
    number : openmc.deplete.AtomNumber
        Total number of atoms in simulation.
    nuclides_with_data : set of str
        A set listing all unique nuclides available from cross_sections.xml.
    chain : openmc.deplete.Chain
        The depletion chain information necessary to form matrices and tallies.
    reaction_rates : openmc.deplete.ReactionRates
        Reaction rates from the last operator step.
    burnable_mats : list of str
        All burnable material IDs
    heavy_metal : float
        Initial heavy metal inventory [g]
    local_mats : list of str
        All burnable material IDs being managed by a single process
    prev_res : Results or None
        Results from a previous depletion calculation. ``None`` if no
        results are to be used.

    """

    def __init__(
            self,
            materials=None,
            cross_sections=None,
            chain_file=None,
            prev_results=None,
            diff_burnable_mats=False,
            diff_volume_method='divide equally',
            fission_q=None,
            helper_kwargs=None,
            reduce_chain=False,
            reduce_chain_level=None):

        # If chain file was not specified, try to get it from global config
        if chain_file is None:
            chain_file = openmc.config.get('chain_file')
            if chain_file is None:
                raise DataError(
                    "No depletion chain specified and could not find depletion "
                    "chain in openmc.config['chain_file']"
                )

        super().__init__(chain_file, fission_q, prev_results)
        self.round_number = False
        self.materials = materials
        self.cross_sections = cross_sections

        check_value('diff volume method', diff_volume_method,
                    {'divide equally', 'match cell'})
        self.diff_volume_method = diff_volume_method

        # Reduce the chain to only those nuclides present
        if reduce_chain:
            init_nuclides = set()
            for material in self.materials:
                if not material.depletable:
                    continue
                for name, _dens_percent, _dens_type in material.nuclides:
                    init_nuclides.add(name)

            self.chain = self.chain.reduce(init_nuclides, reduce_chain_level)

        if diff_burnable_mats:
            self._differentiate_burnable_mats()
            self.materials = self.model.materials

        # Determine which nuclides have cross section data
        # This nuclides variables contains every nuclides
        # for which there is an entry in the micro_xs parameter
        self.nuclides_with_data = self._get_nuclides_with_data(
            self.cross_sections)

        # Select nuclides with data that are also in the chain
        self._burnable_nucs = [nuc.name for nuc in self.chain.nuclides
                               if nuc.name in self.nuclides_with_data]

        # Select nuclides without data that are also in the chain
        self._decay_nucs = [nuc.name for nuc in self.chain.nuclides
                            if nuc.name not in self.nuclides_with_data]

        self.burnable_mats, volumes, all_nuclides = self._get_burnable_mats()
        self.local_mats = _distribute(self.burnable_mats)

        self._mat_index_map = {
            lm: self.burnable_mats.index(lm) for lm in self.local_mats}

        if self.prev_res is not None:
            self._load_previous_results()

        # Extract number densities from the geometry / previous depletion run
        self._extract_number(self.local_mats,
                             volumes,
                             all_nuclides,
                             self.prev_res)

        # Create reaction rates array
        self.reaction_rates = ReactionRates(
            self.local_mats, self._burnable_nucs, self.chain.reactions)

        self._get_helper_classes(helper_kwargs)

    def _differentiate_burnable_mats(self):
        """Assign distribmats for each burnable material"""
        pass

    def _get_burnable_mats(self) -> tuple[list[str], dict[str, float], list[str]]:
        """Determine depletable materials, volumes, and nuclides

        Returns
        -------
        burnable_mats : list of str
            list of burnable material IDs
        volume : dict of str to float
            Volume of each material in [cm^3]
        nuclides : list of str
            Nuclides in order of how they'll appear in the simulation.

        """

        burnable_mats = set()
        model_nuclides = set()
        volume = {}

        self.heavy_metal = 0.0

        # Iterate once through the geometry to get dictionaries
        for mat in self.materials:
            for nuclide in mat.get_nuclides():
                if nuclide in self.nuclides_with_data or self._decay_nucs:
                    model_nuclides.add(nuclide)
                else:
                    msg = (f"Nuclide {nuclide} in material {mat.id} is not "
                           "present in the depletion chain and has no cross "
                           "section data.")
                    warn(msg)
            if mat.depletable:
                burnable_mats.add(str(mat.id))
                if mat.volume is None:
                    if mat.name is None:
                        msg = ("Volume not specified for depletable material "
                               f"with ID={mat.id}.")
                    else:
                        msg = ("Volume not specified for depletable material "
                               f"with ID={mat.id} Name={mat.name}.")
                    raise RuntimeError(msg)
                volume[str(mat.id)] = mat.volume
                self.heavy_metal += mat.fissionable_mass

        # Make sure there are burnable materials
        if not burnable_mats:
            raise RuntimeError(
                "No depletable materials were found in the model.")

        # Sort the sets
        burnable_mats = sorted(burnable_mats, key=int)
        model_nuclides = sorted(model_nuclides)

        # Construct a global nuclide dictionary, burned first
        nuclides = list(self.chain.nuclide_dict)
        for nuc in model_nuclides:
            if nuc not in nuclides:
                nuclides.append(nuc)
        return burnable_mats, volume, nuclides

    def _load_previous_results(self):
        """Load results from a previous depletion simulation"""
        pass

    @abstractmethod
    def _get_nuclides_with_data(self, cross_sections):
        """Find nuclides with cross section data

        Parameters
        ----------
        cross_sections : str or pandas.DataFrame
            Path to continuous energy cross section library, or object
            containing one-group cross-sections.

        Returns
        -------
        nuclides : set of str
            Set of nuclide names that have cross secton data

        """

    def _extract_number(self, local_mats, volume, all_nuclides, prev_res=None):
        """Construct AtomNumber using geometry

        Parameters
        ----------
        local_mats : list of str
            Material IDs to be managed by this process
        volume : dict of str to float
            Volumes for the above materials in [cm^3]
        all_nuclides : list of str
            Nuclides to be used in the simulation.
        prev_res : Results, optional
            Results from a previous depletion calculation

        """
        self.number = AtomNumber(local_mats, all_nuclides, volume, len(self.chain))

        # Now extract and store the number densities
        # From the geometry if no previous depletion results
        if prev_res is None:
            for mat in self.materials:
                if str(mat.id) in local_mats:
                    self._set_number_from_mat(mat)

        # Else from previous depletion results
        else:
            for mat in self.materials:
                if str(mat.id) in local_mats:
                    self._set_number_from_results(mat, prev_res)

    def _set_number_from_mat(self, mat):
        """Extracts material and number densities from openmc.Material

        Parameters
        ----------
        mat : openmc.Material
            The material to read from

        """
        mat_id = str(mat.id)

        for nuclide, atom_per_bcm in mat.get_nuclide_atom_densities().items():
            atom_per_cc = atom_per_bcm * 1.0e24
            self.number.set_atom_density(mat_id, nuclide, atom_per_cc)

    def _set_number_from_results(self, mat, prev_res):
        """Extracts material nuclides and number densities.

        If the nuclide concentration's evolution is tracked, the densities come
        from depletion results. Else, densities are extracted from the geometry
        in the summary.

        Parameters
        ----------
        mat : openmc.Material
            The material to read from
        prev_res : Results
            Results from a previous depletion calculation

        """
        mat_id = str(mat.id)

        # Get nuclide lists from geometry and depletion results
        depl_nuc = prev_res[-1].index_nuc
        geom_nuc_densities = mat.get_nuclide_atom_densities()

        # Merge lists of nuclides, with the same order for every calculation
        geom_nuc_densities.update(depl_nuc)

        for nuclide, atom_per_bcm in geom_nuc_densities.items():
            if nuclide in depl_nuc:
                concentration = prev_res.get_atoms(mat_id, nuclide)[1][-1]
                volume = prev_res[-1].volume[mat_id]
                atom_per_cc = concentration / volume
            else:
                atom_per_cc = atom_per_bcm * 1.0e24

            self.number.set_atom_density(mat_id, nuclide, atom_per_cc)

    @abstractmethod
    def _get_helper_classes(self, helper_kwargs):
        """Create the ``_rate_helper``, ``_normalization_helper``, and
        ``_yield_helper`` objects.

        Parameters
        ----------
        helper_kwargs : dict
            Keyword arguments for helper classes

        """

    def initial_condition(self, materials):
        """Performs final setup and returns initial condition.

        Parameters
        ----------
        materials : list of openmc.lib.Material
            list of materials

        Returns
        -------
        list of numpy.ndarray
            Total density for initial conditions.

        """

        self._rate_helper.generate_tallies(materials, self.chain.reactions)
        self._normalization_helper.prepare(
            self.chain.nuclides, self.reaction_rates.index_nuc)
        # Tell fission yield helper what materials this process is
        # responsible for
        self._yield_helper.generate_tallies(
            materials, tuple(sorted(self._mat_index_map.values())))

        # Return number density vector
        return list(self.number.get_mat_slice(np.s_[:]))

    def _update_materials_and_nuclides(self, vec):
        """Update the number density, material compositions, and nuclide
        lists in helper objects

        Parameters
        ----------
        vec : list of numpy.ndarray
            Total atoms.

        """

        # Update the number densities regardless of the source rate
        self.number.set_density(vec)
        self._update_materials()

        # Update tally nuclides data in preparation for transport solve
        nuclides = self._get_reaction_nuclides()
        self._rate_helper.nuclides = nuclides
        self._normalization_helper.nuclides = nuclides
        self._yield_helper.update_tally_nuclides(nuclides)

    @abstractmethod
    def _update_materials(self):
        """Updates material compositions in OpenMC on all processes."""

    def write_bos_data(self, step):
        """Document beginning of step data for a given step

        Called at the beginning of a depletion step and at
        the final point in the simulation.

        Parameters
        ----------
        step : int
            Current depletion step including restarts
        """
        # Since we aren't running a transport simulation, we simply pass
        pass

    def _get_reaction_nuclides(self):
        """Determine nuclides that should be tallied for reaction rates.

        This method returns a list of all nuclides that have cross section data
        and are listed in the depletion chain. Technically, we should count
        nuclides that may not appear in the depletion chain because we still
        need to get the fission reaction rate for these nuclides in order to
        normalize power, but that is left as a future exercise.

        Returns
        -------
        list of str
            Nuclides with reaction rates

        """
        nuc_set = set()

        # Create the set of all nuclides in the decay chain in materials marked
        # for burning in which the number density is greater than zero.
        for nuc in self.number.nuclides:
            if nuc in self.nuclides_with_data:
                nuc_set.add(nuc)

        # Communicate which nuclides have nonzeros to rank 0
        if comm.rank == 0:
            for i in range(1, comm.size):
                nuc_newset = comm.recv(source=i, tag=i)
                nuc_set |= nuc_newset
        else:
            comm.send(nuc_set, dest=0, tag=comm.rank)

        if comm.rank == 0:
            # Sort nuclides in the same order as self.number
            nuc_list = [nuc for nuc in self.number.nuclides
                        if nuc in nuc_set]
        else:
            nuc_list = None

        # Store list of nuclides on each process
        nuc_list = comm.bcast(nuc_list)
        return [nuc for nuc in nuc_list if nuc in self.chain]

    def _calculate_reaction_rates(self, source_rate):
        """Unpack tallies from OpenMC and return an operator result

        This method uses OpenMC's C API bindings to determine the k-effective
        value and reaction rates from the simulation. The reaction rates are
        normalized by a helper class depending on the method being used.

        Parameters
        ----------
        source_rate : float
            Power in [W] or source rate in [neutron/sec]

        Returns
        -------
        rates : openmc.deplete.ReactionRates
            Reaction rates for nuclides

        """
        rates = self.reaction_rates
        rates.fill(0.0)

        # Extract reaction nuclides
        rxn_nuclides = self._rate_helper.nuclides

        # Form fast map
        nuc_ind = [rates.index_nuc[nuc] for nuc in rxn_nuclides]
        rx_ind = [rates.index_rx[react] for react in self.chain.reactions]

        # Keep track of energy produced from all reactions in eV per source
        # particle
        self._normalization_helper.reset()
        self._yield_helper.unpack()

        # Store fission yield dictionaries
        fission_yields = []

        # Create arrays to store fission Q values, reaction rates, and nuclide
        # numbers, zeroed out in material iteration
        number = np.empty(rates.n_nuc)

        fission_ind = rates.index_rx.get("fission")

        # Reset the cached material reaction rates tallies
        self._rate_helper.reset_tally_means()

        # Extract results
        for i, mat in enumerate(self.local_mats):
            # Get tally index
            mat_index = self._mat_index_map[mat]

            # Zero out reaction rates and nuclide numbers
            number.fill(0.0)

            # Get new number densities
            for nuc, i_nuc_results in zip(rxn_nuclides, nuc_ind):
                number[i_nuc_results] = self.number[mat, nuc]

            # Get microscopic reaction rates in [(reactions/src)*b-cm/atom]. 2D
            # array with shape (nuclides, reactions).
            tally_rates = self._rate_helper.get_material_rates(
                mat_index, nuc_ind, rx_ind)

            # Compute fission yields for this material
            fission_yields.append(self._yield_helper.weighted_yields(i))

            # Accumulate energy from fission
            volume_b_cm = 1e24 * self.number.get_mat_volume(mat)
            if fission_ind is not None:
                atom_per_bcm = number / volume_b_cm
                fission_rates = tally_rates[:, fission_ind] * atom_per_bcm
                self._normalization_helper.update(fission_rates)

            # Divide by [b-cm] to get [(reactions/src)/atom]
            rates[i] = tally_rates / volume_b_cm

        # Scale reaction rates to obtain units of [(reactions/sec)/atom]
        rates *= self._normalization_helper.factor(source_rate)

        # Store new fission yields on the chain
        self.chain.fission_yields = fission_yields

        return rates

    def get_results_info(self):
        """Returns volume list, material lists, and nuc lists.

        Returns
        -------
        volume : dict of str float
            Volumes corresponding to materials in full_burn_dict
        nuc_list : list of str
            A list of all nuclide names. Used for sorting the simulation.
        burn_list : list of int
            A list of all material IDs to be burned.  Used for sorting the simulation.
        full_burn_list : list
            List of all burnable material IDs

        """
        nuc_list = self.number.burnable_nuclides
        burn_list = self.local_mats

        volume = {}
        for i, mat in enumerate(burn_list):
            volume[mat] = self.number.volume[i]

        # Combine volume dictionaries across processes
        volume_list = comm.allgather(volume)
        volume = {k: v for d in volume_list for k, v in d.items()}

        return volume, nuc_list, burn_list, self.burnable_mats

1	"""OpenMC transport operator
2
3	This module implements functions shared by both OpenMC transport-coupled and
4	transport-independent transport operators.
5
6	"""
7
8	from abc import abstractmethod	12✔
9	from warnings import warn	12✔
10
11	import numpy as np	12✔
12
13	import openmc	12✔
14	from openmc.checkvalue import check_value	12✔
15	from openmc.exceptions import DataError	12✔
16	from openmc.mpi import comm	12✔
17	from .abc import TransportOperator, OperatorResult	12✔
18	from .atom_number import AtomNumber	12✔
19	from .reaction_rates import ReactionRates	12✔
20	from .pool import _distribute	12✔
21
22	__all__ = ["OpenMCOperator", "OperatorResult"]	12✔
23
24
25	class OpenMCOperator(TransportOperator):	12✔
26	"""Abstract class holding OpenMC-specific functions for running
27	depletion calculations.
28
29	Specific classes for running transport-coupled or transport-independent
30	depletion calculations are implemented as subclasses of OpenMCOperator.
31
32	Parameters
33	----------
34	materials : openmc.Materials
35	List of all materials in the model
36	cross_sections : str or list of MicroXS
37	Path to continuous energy cross section library, or list of objects
38	containing cross sections.
39	chain_file : str, optional
40	Path to the depletion chain XML file. Defaults to
41	openmc.config['chain_file'].
42	prev_results : Results, optional
43	Results from a previous depletion calculation. If this argument is
44	specified, the depletion calculation will start from the latest state
45	in the previous results.
46	diff_burnable_mats : bool, optional
47	Whether to differentiate burnable materials with multiple instances.
48	fission_q : dict, optional
49	Dictionary of nuclides and their fission Q values [eV].
50	helper_kwargs : dict
51	Keyword arguments for helper classes
52	reduce_chain : bool, optional
53	If True, use :meth:`openmc.deplete.Chain.reduce()` to reduce the
54	depletion chain up to ``reduce_chain_level``.
55	reduce_chain_level : int, optional
56	Depth of the search when reducing the depletion chain. Only used
57	if ``reduce_chain`` evaluates to true. The default value of
58	``None`` implies no limit on the depth.
59
60	diff_volume_method : str
61	Specifies how the volumes of the new materials should be found. Default
62	is to 'divide equally' which divides the original material volume
63	equally between the new materials, 'match cell' sets the volume of the
64	material to volume of the cell they fill.
65
66	.. versionadded:: 0.14.0
67
68	Attributes
69	----------
70	materials : openmc.Materials
71	All materials present in the model
72	cross_sections : str or list of MicroXS
73	Path to continuous energy cross section library, or list of objects
74	containing cross sections.
75	output_dir : pathlib.Path
76	Path to output directory to save results.
77	round_number : bool
78	Whether or not to round output to OpenMC to 8 digits.
79	Useful in testing, as OpenMC is incredibly sensitive to exact values.
80	number : openmc.deplete.AtomNumber
81	Total number of atoms in simulation.
82	nuclides_with_data : set of str
83	A set listing all unique nuclides available from cross_sections.xml.
84	chain : openmc.deplete.Chain
85	The depletion chain information necessary to form matrices and tallies.
86	reaction_rates : openmc.deplete.ReactionRates
87	Reaction rates from the last operator step.
88	burnable_mats : list of str
89	All burnable material IDs
90	heavy_metal : float
91	Initial heavy metal inventory [g]
92	local_mats : list of str
93	All burnable material IDs being managed by a single process
94	prev_res : Results or None
95	Results from a previous depletion calculation. ``None`` if no
96	results are to be used.
97
98	"""
99
100	def __init__(	12✔
101	self,
102	materials=None,
103	cross_sections=None,
104	chain_file=None,
105	prev_results=None,
106	diff_burnable_mats=False,
107	diff_volume_method='divide equally',
108	fission_q=None,
109	helper_kwargs=None,
110	reduce_chain=False,
111	reduce_chain_level=None):
112
113	# If chain file was not specified, try to get it from global config
114	if chain_file is None:	12✔
115	chain_file = openmc.config.get('chain_file')	×
116	if chain_file is None:	×
117	raise DataError(	×
118	"No depletion chain specified and could not find depletion "
119	"chain in openmc.config['chain_file']"
120	)
121
122	super().__init__(chain_file, fission_q, prev_results)	12✔
123	self.round_number = False	12✔
124	self.materials = materials	12✔
125	self.cross_sections = cross_sections	12✔
126
127	check_value('diff volume method', diff_volume_method,	12✔
128	{'divide equally', 'match cell'})
129	self.diff_volume_method = diff_volume_method	12✔
130
131	# Reduce the chain to only those nuclides present
132	if reduce_chain:	12✔
133	init_nuclides = set()	×
134	for material in self.materials:	×
135	if not material.depletable:	×
136	continue	×
137	for name, _dens_percent, _dens_type in material.nuclides:	×
138	init_nuclides.add(name)	×
139
140	self.chain = self.chain.reduce(init_nuclides, reduce_chain_level)	×
141
142	if diff_burnable_mats:	12✔
143	self._differentiate_burnable_mats()	12✔
144	self.materials = self.model.materials	12✔
145
146	# Determine which nuclides have cross section data
147	# This nuclides variables contains every nuclides
148	# for which there is an entry in the micro_xs parameter
149	self.nuclides_with_data = self._get_nuclides_with_data(	12✔
150	self.cross_sections)
151
152	# Select nuclides with data that are also in the chain
153	self._burnable_nucs = [nuc.name for nuc in self.chain.nuclides	12✔
154	if nuc.name in self.nuclides_with_data]
155
156	# Select nuclides without data that are also in the chain
157	self._decay_nucs = [nuc.name for nuc in self.chain.nuclides	12✔
158	if nuc.name not in self.nuclides_with_data]
159
160	self.burnable_mats, volumes, all_nuclides = self._get_burnable_mats()	12✔
161	self.local_mats = _distribute(self.burnable_mats)	12✔
162
163	self._mat_index_map = {	12✔
164	lm: self.burnable_mats.index(lm) for lm in self.local_mats}
165
166	if self.prev_res is not None:	12✔
UNCOV 167	self._load_previous_results()	×
168
169	# Extract number densities from the geometry / previous depletion run
170	self._extract_number(self.local_mats,	12✔
171	volumes,
172	all_nuclides,
173	self.prev_res)
174
175	# Create reaction rates array
176	self.reaction_rates = ReactionRates(	12✔
177	self.local_mats, self._burnable_nucs, self.chain.reactions)
178
179	self._get_helper_classes(helper_kwargs)	12✔
180
181	def _differentiate_burnable_mats(self):	12✔
182	"""Assign distribmats for each burnable material"""
UNCOV 183	pass	×
184
185	def _get_burnable_mats(self) -> tuple[list[str], dict[str, float], list[str]]:	12✔
186	"""Determine depletable materials, volumes, and nuclides
187
188	Returns
189	-------
190	burnable_mats : list of str
191	list of burnable material IDs
192	volume : dict of str to float
193	Volume of each material in [cm^3]
194	nuclides : list of str
195	Nuclides in order of how they'll appear in the simulation.
196
197	"""
198
199	burnable_mats = set()	12✔
200	model_nuclides = set()	12✔
201	volume = {}	12✔
202
203	self.heavy_metal = 0.0	12✔
204
205	# Iterate once through the geometry to get dictionaries
206	for mat in self.materials:	12✔
207	for nuclide in mat.get_nuclides():	12✔
208	if nuclide in self.nuclides_with_data or self._decay_nucs:	12✔
209	model_nuclides.add(nuclide)	12✔
210	else:
UNCOV 211	msg = (f"Nuclide {nuclide} in material {mat.id} is not "	×
212	"present in the depletion chain and has no cross "
213	"section data.")
UNCOV 214	warn(msg)	×
215	if mat.depletable:	12✔
216	burnable_mats.add(str(mat.id))	12✔
217	if mat.volume is None:	12✔
UNCOV 218	if mat.name is None:	×
UNCOV 219	msg = ("Volume not specified for depletable material "	×
220	f"with ID={mat.id}.")
221	else:
UNCOV 222	msg = ("Volume not specified for depletable material "	×
223	f"with ID={mat.id} Name={mat.name}.")
224	raise RuntimeError(msg)	×
225	volume[str(mat.id)] = mat.volume	12✔
226	self.heavy_metal += mat.fissionable_mass	12✔
227
228	# Make sure there are burnable materials
229	if not burnable_mats:	12✔
UNCOV 230	raise RuntimeError(	×
231	"No depletable materials were found in the model.")
232
233	# Sort the sets
234	burnable_mats = sorted(burnable_mats, key=int)	12✔
235	model_nuclides = sorted(model_nuclides)	12✔
236
237	# Construct a global nuclide dictionary, burned first
238	nuclides = list(self.chain.nuclide_dict)	12✔
239	for nuc in model_nuclides:	12✔
240	if nuc not in nuclides:	12✔
241	nuclides.append(nuc)	12✔
242	return burnable_mats, volume, nuclides	12✔
243
244	def _load_previous_results(self):	12✔
245	"""Load results from a previous depletion simulation"""
UNCOV 246	pass	×
247
248	@abstractmethod	12✔
249	def _get_nuclides_with_data(self, cross_sections):	12✔
250	"""Find nuclides with cross section data
251
252	Parameters
253	----------
254	cross_sections : str or pandas.DataFrame
255	Path to continuous energy cross section library, or object
256	containing one-group cross-sections.
257
258	Returns
259	-------
260	nuclides : set of str
261	Set of nuclide names that have cross secton data
262
263	"""
264
265	def _extract_number(self, local_mats, volume, all_nuclides, prev_res=None):	12✔
266	"""Construct AtomNumber using geometry
267
268	Parameters
269	----------
270	local_mats : list of str
271	Material IDs to be managed by this process
272	volume : dict of str to float
273	Volumes for the above materials in [cm^3]
274	all_nuclides : list of str
275	Nuclides to be used in the simulation.
276	prev_res : Results, optional
277	Results from a previous depletion calculation
278
279	"""
280	self.number = AtomNumber(local_mats, all_nuclides, volume, len(self.chain))	12✔
281
282	# Now extract and store the number densities
283	# From the geometry if no previous depletion results
284	if prev_res is None:	12✔
285	for mat in self.materials:	12✔
286	if str(mat.id) in local_mats:	12✔
287	self._set_number_from_mat(mat)	12✔
288
289	# Else from previous depletion results
290	else:
UNCOV 291	for mat in self.materials:	×
UNCOV 292	if str(mat.id) in local_mats:	×
293	self._set_number_from_results(mat, prev_res)	×
294
295	def _set_number_from_mat(self, mat):	12✔
296	"""Extracts material and number densities from openmc.Material
297
298	Parameters
299	----------
300	mat : openmc.Material
301	The material to read from
302
303	"""
304	mat_id = str(mat.id)	12✔
305
306	for nuclide, atom_per_bcm in mat.get_nuclide_atom_densities().items():	12✔
307	atom_per_cc = atom_per_bcm * 1.0e24	12✔
308	self.number.set_atom_density(mat_id, nuclide, atom_per_cc)	12✔
309
310	def _set_number_from_results(self, mat, prev_res):	12✔
311	"""Extracts material nuclides and number densities.
312
313	If the nuclide concentration's evolution is tracked, the densities come
314	from depletion results. Else, densities are extracted from the geometry
315	in the summary.
316
317	Parameters
318	----------
319	mat : openmc.Material
320	The material to read from
321	prev_res : Results
322	Results from a previous depletion calculation
323
324	"""
UNCOV 325	mat_id = str(mat.id)	×
326
327	# Get nuclide lists from geometry and depletion results
UNCOV 328	depl_nuc = prev_res[-1].index_nuc	×
UNCOV 329	geom_nuc_densities = mat.get_nuclide_atom_densities()	×
330
331	# Merge lists of nuclides, with the same order for every calculation
UNCOV 332	geom_nuc_densities.update(depl_nuc)	×
333
334	for nuclide, atom_per_bcm in geom_nuc_densities.items():	×
UNCOV 335	if nuclide in depl_nuc:	×
336	concentration = prev_res.get_atoms(mat_id, nuclide)[1][-1]	×
337	volume = prev_res[-1].volume[mat_id]	×
338	atom_per_cc = concentration / volume	×
339	else:
340	atom_per_cc = atom_per_bcm * 1.0e24	×
341
342	self.number.set_atom_density(mat_id, nuclide, atom_per_cc)	×
343
344	@abstractmethod	12✔
345	def _get_helper_classes(self, helper_kwargs):	12✔
346	"""Create the ``_rate_helper``, ``_normalization_helper``, and
347	``_yield_helper`` objects.
348
349	Parameters
350	----------
351	helper_kwargs : dict
352	Keyword arguments for helper classes
353
354	"""
355
356	def initial_condition(self, materials):	12✔
357	"""Performs final setup and returns initial condition.
358
359	Parameters
360	----------
361	materials : list of openmc.lib.Material
362	list of materials
363
364	Returns
365	-------
366	list of numpy.ndarray
367	Total density for initial conditions.
368
369	"""
370
371	self._rate_helper.generate_tallies(materials, self.chain.reactions)	12✔
372	self._normalization_helper.prepare(	12✔
373	self.chain.nuclides, self.reaction_rates.index_nuc)
374	# Tell fission yield helper what materials this process is
375	# responsible for
376	self._yield_helper.generate_tallies(	12✔
377	materials, tuple(sorted(self._mat_index_map.values())))
378
379	# Return number density vector
380	return list(self.number.get_mat_slice(np.s_[:]))	12✔
381
382	def _update_materials_and_nuclides(self, vec):	12✔
383	"""Update the number density, material compositions, and nuclide
384	lists in helper objects
385
386	Parameters
387	----------
388	vec : list of numpy.ndarray
389	Total atoms.
390
391	"""
392
393	# Update the number densities regardless of the source rate
394	self.number.set_density(vec)	12✔
395	self._update_materials()	12✔
396
397	# Update tally nuclides data in preparation for transport solve
398	nuclides = self._get_reaction_nuclides()	12✔
399	self._rate_helper.nuclides = nuclides	12✔
400	self._normalization_helper.nuclides = nuclides	12✔
401	self._yield_helper.update_tally_nuclides(nuclides)	12✔
402
403	@abstractmethod	12✔
404	def _update_materials(self):	12✔
405	"""Updates material compositions in OpenMC on all processes."""
406
407	def write_bos_data(self, step):	12✔
408	"""Document beginning of step data for a given step
409
410	Called at the beginning of a depletion step and at
411	the final point in the simulation.
412
413	Parameters
414	----------
415	step : int
416	Current depletion step including restarts
417	"""
418	# Since we aren't running a transport simulation, we simply pass
419	pass	12✔
420
421	def _get_reaction_nuclides(self):	12✔
422	"""Determine nuclides that should be tallied for reaction rates.
423
424	This method returns a list of all nuclides that have cross section data
425	and are listed in the depletion chain. Technically, we should count
426	nuclides that may not appear in the depletion chain because we still
427	need to get the fission reaction rate for these nuclides in order to
428	normalize power, but that is left as a future exercise.
429
430	Returns
431	-------
432	list of str
433	Nuclides with reaction rates
434
435	"""
436	nuc_set = set()	12✔
437
438	# Create the set of all nuclides in the decay chain in materials marked
439	# for burning in which the number density is greater than zero.
440	for nuc in self.number.nuclides:	12✔
441	if nuc in self.nuclides_with_data:	12✔
442	nuc_set.add(nuc)	12✔
443
444	# Communicate which nuclides have nonzeros to rank 0
445	if comm.rank == 0:	12✔
446	for i in range(1, comm.size):	12✔
UNCOV 447	nuc_newset = comm.recv(source=i, tag=i)	×
UNCOV 448	nuc_set \|= nuc_newset	×
449	else:
UNCOV 450	comm.send(nuc_set, dest=0, tag=comm.rank)	×
451
452	if comm.rank == 0:	12✔
453	# Sort nuclides in the same order as self.number
454	nuc_list = [nuc for nuc in self.number.nuclides	12✔
455	if nuc in nuc_set]
456	else:
UNCOV 457	nuc_list = None	×
458
459	# Store list of nuclides on each process
460	nuc_list = comm.bcast(nuc_list)	12✔
461	return [nuc for nuc in nuc_list if nuc in self.chain]	12✔
462
463	def _calculate_reaction_rates(self, source_rate):	12✔
464	"""Unpack tallies from OpenMC and return an operator result
465
466	This method uses OpenMC's C API bindings to determine the k-effective
467	value and reaction rates from the simulation. The reaction rates are
468	normalized by a helper class depending on the method being used.
469
470	Parameters
471	----------
472	source_rate : float
473	Power in [W] or source rate in [neutron/sec]
474
475	Returns
476	-------
477	rates : openmc.deplete.ReactionRates
478	Reaction rates for nuclides
479
480	"""
481	rates = self.reaction_rates	12✔
482	rates.fill(0.0)	12✔
483
484	# Extract reaction nuclides
485	rxn_nuclides = self._rate_helper.nuclides	12✔
486
487	# Form fast map
488	nuc_ind = [rates.index_nuc[nuc] for nuc in rxn_nuclides]	12✔
489	rx_ind = [rates.index_rx[react] for react in self.chain.reactions]	12✔
490
491	# Keep track of energy produced from all reactions in eV per source
492	# particle
493	self._normalization_helper.reset()	12✔
494	self._yield_helper.unpack()	12✔
495
496	# Store fission yield dictionaries
497	fission_yields = []	12✔
498
499	# Create arrays to store fission Q values, reaction rates, and nuclide
500	# numbers, zeroed out in material iteration
501	number = np.empty(rates.n_nuc)	12✔
502
503	fission_ind = rates.index_rx.get("fission")	12✔
504
505	# Reset the cached material reaction rates tallies
506	self._rate_helper.reset_tally_means()	12✔
507
508	# Extract results
509	for i, mat in enumerate(self.local_mats):	12✔
510	# Get tally index
511	mat_index = self._mat_index_map[mat]	12✔
512
513	# Zero out reaction rates and nuclide numbers
514	number.fill(0.0)	12✔
515
516	# Get new number densities
517	for nuc, i_nuc_results in zip(rxn_nuclides, nuc_ind):	12✔
518	number[i_nuc_results] = self.number[mat, nuc]	12✔
519
520	# Get microscopic reaction rates in [(reactions/src)*b-cm/atom]. 2D
521	# array with shape (nuclides, reactions).
522	tally_rates = self._rate_helper.get_material_rates(	12✔
523	mat_index, nuc_ind, rx_ind)
524
525	# Compute fission yields for this material
526	fission_yields.append(self._yield_helper.weighted_yields(i))	12✔
527
528	# Accumulate energy from fission
529	volume_b_cm = 1e24 * self.number.get_mat_volume(mat)	12✔
530	if fission_ind is not None:	12✔
531	atom_per_bcm = number / volume_b_cm	12✔
532	fission_rates = tally_rates[:, fission_ind] * atom_per_bcm	12✔
533	self._normalization_helper.update(fission_rates)	12✔
534
535	# Divide by [b-cm] to get [(reactions/src)/atom]
536	rates[i] = tally_rates / volume_b_cm	12✔
537
538	# Scale reaction rates to obtain units of [(reactions/sec)/atom]
539	rates *= self._normalization_helper.factor(source_rate)	12✔
540
541	# Store new fission yields on the chain
542	self.chain.fission_yields = fission_yields	12✔
543
544	return rates	12✔
545
546	def get_results_info(self):	12✔
547	"""Returns volume list, material lists, and nuc lists.
548
549	Returns
550	-------
551	volume : dict of str float
552	Volumes corresponding to materials in full_burn_dict
553	nuc_list : list of str
554	A list of all nuclide names. Used for sorting the simulation.
555	burn_list : list of int
556	A list of all material IDs to be burned. Used for sorting the simulation.
557	full_burn_list : list
558	List of all burnable material IDs
559
560	"""
561	nuc_list = self.number.burnable_nuclides	12✔
562	burn_list = self.local_mats	12✔
563
564	volume = {}	12✔
565	for i, mat in enumerate(burn_list):	12✔
566	volume[mat] = self.number.volume[i]	12✔
567
568	# Combine volume dictionaries across processes
569	volume_list = comm.allgather(volume)	12✔
570	volume = {k: v for d in volume_list for k, v in d.items()}	12✔
571
572	return volume, nuc_list, burn_list, self.burnable_mats	12✔

openmc-dev / openmc / 12776996362

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous