# Copyright 2019 TerraPower, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
r"""
This module contains the code necessary to represent parameter definitions.
``ParameterDefinition``\ s are the metadata that describe specific parameters, and aid in enforcing
certain rules upon the parameters themselves and the parameter collections that contain them.
This module also describes the ``ParameterDefinitionCollection`` class, which serves as a
specialized container to manage related parameter definitions.
See Also
--------
armi.reactor.parameters
"""
from typing import Any, Dict, Optional, Sequence, Tuple, Type
import enum
import functools
import re
import numpy
from armi import runLog
from armi.reactor.flags import Flags
from armi.reactor.parameters.exceptions import ParameterError, ParameterDefinitionError
# bitwise masks for high-speed operations on the `assigned` attribute
# see: https://web.archive.org/web/20120225043338/http://www.vipan.com/htdocs/bitwisehelp.html
# Note that the various operations are responsible for clearing the flags on the events.
# These should be interpreted as:
# The Parameter or ParameterCollection has been modified SINCE_<time-description>
# In order for that to happen, the flags need to be cleared when the <time-description> begins.
SINCE_INITIALIZATION = 1
SINCE_LAST_DISTRIBUTE_STATE = 4
SINCE_LAST_GEOMETRY_TRANSFORMATION = 8
SINCE_BACKUP = 16
SINCE_ANYTHING = (
SINCE_LAST_DISTRIBUTE_STATE
| SINCE_INITIALIZATION
| SINCE_LAST_GEOMETRY_TRANSFORMATION
| SINCE_BACKUP
)
NEVER = 32
[docs]class Category:
"""
A "namespace" for storing parameter categories.
Notes
-----
* `cumulative` parameters are accumulated over many time steps
* `pinQuantities` parameters are defined on the pin level within a block
* `multiGroupQuantities` parameters have group dependence (often a 1D numpy array)
* `fluxQuantities` parameters are related to neutron or gamma flux
* `neutronics` parameters are calculated in a neutronics global flux solve
* `gamma` parameters are calculated in a fixed-source gamma solve
* `detailedAxialExpansion` parameters are marked as such so that they are mapped from the
uniform mesh back to the non-uniform mesh
* `reactivity coefficients` parameters are related to reactivity coefficient or kinetics
parameters for kinetics solutions
* `thermal hydraulics` parameters come from a thermal hydraulics physics plugin (e.g., flow
rates, temperatures, etc.)
"""
depletion = "depletion"
cumulative = "cumulative"
cumulativeOverCycle = "cumulative over cycle"
assignInBlueprints = "assign in blueprints"
retainOnReplacement = "retain on replacement"
pinQuantities = "pinQuantities"
fluxQuantities = "fluxQuantities"
multiGroupQuantities = "multi-group quantities"
neutronics = "neutronics"
gamma = "gamma"
detailedAxialExpansion = "detailedAxialExpansion"
reactivityCoefficients = "reactivity coefficients"
thermalHydraulics = "thermal hydraulics"
[docs]class ParamLocation(enum.Flag):
"""Represents the point on which a parameter is physically meaningful."""
TOP = 1
CENTROID = 2
BOTTOM = 4
AVERAGE = 10 # 2 + 8
MAX = 16
CORNERS = 32
EDGES = 64
VOLUME_INTEGRATED = 128
CHILDREN = 256 # on some child of a composite, like a pin
[docs]class NoDefault:
"""Class used to allow distinction between not setting a default and setting a default of
``None``.
"""
def __init__(self):
raise NotImplementedError("You cannot create an instance of NoDefault")
class _Undefined:
"""Class used to identify a parameter property as being in the undefined state."""
def __init__(self):
raise NotImplementedError("You cannot create an instance of _Undefined.")
[docs]class Serializer:
r"""
Abstract class describing serialize/deserialize operations for Parameter data.
Parameters need to be stored to and read from database files. This currently requires that the
Parameter data be converted to a numpy array of a datatype supported by the ``h5py`` package.
Some parameters may contain data that are not trivially representable in numpy/HDF5, and need
special treatment. Subclassing ``Serializer`` and setting it as a ``Parameter``\ s
``serializer`` allows for special operations to be performed on the parameter values as they are
stored to the database or read back in.
The ``Database3`` already knows how to handle certain cases where the data are not
straightforward to get into a numpy array, such as when:
- There are ``None``\ s.
- The dimensions of the values stored on each object are inconsistent (e.g.,
"jagged" arrays)
So, in these cases, a Serializer is not needed. Serializers are necessary for when the actual
data need to be converted to a native data type (e.g., int, float, etc). For example, we use a
Serializer to handle writing ``Flags`` to the Database, as they tend to be too big to fit into a
system-native integer.
.. important::
Defining a Serializer for a Parameter in part defines the underlying representation of the
data within a database file; the data stored in a database are sensitive to the code that
wrote them. Changing the method that a Serializer uses to pack or unpack data may break
compatibility with old database files. Therefore, Serializers should be diligent about
signaling changes by updating their version. It is also good practice, whenever possible,
to support reading old versions so that database files written by old versions can still be
read.
.. impl:: Users can define custom parameter serializers.
:id: I_ARMI_PARAM_SERIALIZE
:implements: R_ARMI_PARAM_SERIALIZE
Important physical parameters are stored in every ARMI object. These parameters represent
the plant's state during execution of the model. Currently, this requires that the
parameters be serializable to a numpy array of a datatype supported by the ``h5py`` package
so that the data can be written to, and subsequently read from, an HDF5 file.
This class allows for these parameters to be serialized in a custom manner by providing
interfaces for packing and unpacking parameter data. The user or downstream plugin is able
to specify how data is serialized if that data is not naturally serializable.
See Also
--------
armi.bookkeeping.db.database3.packSpecialData
armi.bookkeeping.db.database3.unpackSpecialData
armi.reactor.flags.FlagSerializer
"""
# This will accompany the packed data as an attribute when written, and will be provided to the
# unpack() method when reading. If the underlying format of the data changes, make sure to
# change this.
version: Optional[str] = None
[docs] @staticmethod
def pack(data: Sequence[any]) -> Tuple[numpy.ndarray, Dict[str, any]]:
"""
Given unpacked data, return packed data and a dictionary of attributes needed to unpack it.
This should perform the fundamental packing operation, returning the packed data and any
metadata ("attributes") that would be necessary to unpack the data. The class's version is
always stored, so no need to provide it as an attribute.
See Also
--------
armi.reactor.flags.FlagSerializer.pack
"""
raise NotImplementedError()
[docs] @classmethod
def unpack(
cls, data: numpy.ndarray, version: Any, attrs: Dict[str, any]
) -> Sequence[any]:
"""Given packed data and attributes, return the unpacked data."""
raise NotImplementedError()
[docs]def isNumpyArray(paramStr):
"""Helper meta-function to create a method that sets a Parameter value to a NumPy array.
Parameters
----------
paramStr : str
Name of the Parameter we want to set.
Returns
-------
function
A setter method on the Parameter class to force the value to be a NumPy array.
"""
def setParameter(selfObj, value):
if value is None or isinstance(value, numpy.ndarray):
setattr(selfObj, "_p_" + paramStr, value)
else:
setattr(selfObj, "_p_" + paramStr, numpy.array(value))
return setParameter
[docs]@functools.total_ordering
class Parameter:
"""Metadata about a specific parameter."""
_validName = re.compile("^[a-zA-Z0-9_]+$")
# Using slots because Parameters are pretty static and mostly POD. __slots__ make this official,
# and offer some performance benefits in memory (not too important; there aren't that many
# instances of Parameter to begin with) and attribute access time (more important, since we need
# to go through Parameter objects to get to a specific parameter's value in a
# ParameterCollection)
__slots__ = (
"name",
"fieldName",
"collectionType",
"location",
"saveToDB",
"serializer",
"units",
"default",
"_getter",
"_setter",
"description",
"categories",
"assigned",
"_backup",
)
def __init__(
self,
name,
units,
description,
location,
saveToDB,
default,
setter,
categories,
serializer: Optional[Type[Serializer]] = None,
):
assert self._validName.match(name), "{} is not a valid param name".format(name)
# nonsensical to have a serializer with no intention of saving to DB
assert not (serializer is not None and not saveToDB)
assert serializer is None or saveToDB
# TODO: This warning is temporary. At some point, it will become an AssertionError.
if not len(description):
runLog.warning(
f"DeprecationWarning: Parameter {name} defined without description.",
single=True,
)
self.collectionType = _Undefined
self.name = name
self.fieldName = "_p_" + name
self.location = location
self.saveToDB = saveToDB
self.serializer = serializer
self.description = description
self.units = units
self.default = default
self.categories = categories
self.assigned = NEVER
self._backup = None
if self.default is not NoDefault:
def paramGetter(p_self):
return getattr(p_self, self.fieldName, self.default)
else:
def paramGetter(p_self):
value = getattr(p_self, self.fieldName)
if value is NoDefault:
raise ParameterError(
"Cannot get value for parameter `{}` in `{}` as no default has been "
"defined, and no value has been assigned.".format(
self.name, type(p_self)
)
)
return value
self._getter = paramGetter
self._setter = None # actually, it gets assigned with this:
self.setter(setter)
def __repr__(self):
return "<ParamDef name:{} collectionType:{} units:{} assigned:{}>".format(
self.name, self.collectionType.__name__, self.units, self.assigned
)
def __eq__(self, other):
"""Name defines equality."""
return self.name == other.name
def __ne__(self, other):
return not (self == other)
def __lt__(self, other):
"""Sort alphabetically by name."""
return self.name < other.name
def __hash__(self):
return hash(self.name) + id(self)
def __setstate__(self, state):
self._backup = state[0] # a tuple of 1 element.
def __set__(self, obj, val):
"""This is a property setter, see Python documentation for "descriptor"."""
self._setter(obj, val)
def __get__(self, obj, cls=None):
"""This is a property getter, see Python documentation for "descriptor".
Notes
-----
We do not check to see if ``cls != None``. This is an optimization choice, that someone may
deem unnecessary. As a result, unlike Python's ``property`` class, a subclass cannot
override the getter method.
"""
return self._getter(obj)
[docs] def setter(self, setter):
"""Decorator method for assigning setter.
.. impl:: Provide a way to signal if a parameter needs updating across processes.
:id: I_ARMI_PARAM_PARALLEL
:implements: R_ARMI_PARAM_PARALLEL
Parameters need to be handled properly during parallel code execution. This includes
notifying processes if a parameter has been updated by another process. This method
allows for setting a parameter's value as well as an attribute that signals whether this
parameter has been updated. Future processes will be able to query this attribute so
that the parameter's status is properly communicated.
Notes
-----
Unlike the traditional Python ``property`` class, this does not return a new instance of a
``Parameter``; therefore it cannot be reassigned in the same way that a Python ``property``
can be.
Examples
--------
>>> class MyParameterCollection(parameters.ParameterCollection):
... mass = parameters.Parameter(...)
... @mass.setter
... def mass(self, value):
... if value < 0:
... raise ValueError("Negative mass is not possible, consider a diet.")
... self._p_speed = value
"""
if setter is NoDefault:
def paramSetter(p_self, value):
self.assigned = SINCE_ANYTHING
p_self.assigned = SINCE_ANYTHING
setattr(p_self, self.fieldName, value)
elif setter is None:
def paramSetter(p_self, value):
raise ParameterError(
"Cannot set value for parameter `{}` on {} to `{}`, it has a restricted setter.".format(
self.name, p_self, value
)
)
elif callable(setter):
def paramSetter(p_self, value):
self.assigned = SINCE_ANYTHING
p_self.assigned = SINCE_ANYTHING
setter(p_self, value)
else:
raise ParameterDefinitionError(
"The setter for parameter `{}` must be callable. Setter attribute: {}".format(
self.name, setter
)
)
self._setter = paramSetter
return self
[docs] def backUp(self):
"""Back up the assigned state."""
self._backup = (self._backup, self.assigned)
[docs] def restoreBackup(self, paramsToApply):
"""Restore the backed up state."""
if self in paramsToApply:
# retain self.assigned if self in a category
self._backup, _assigned = self._backup
else:
self._backup, self.assigned = self._backup
[docs] def atLocation(self, loc):
"""True if parameter is defined at location."""
return self.location and self.location & loc
[docs]class ParameterDefinitionCollection:
"""
A very specialized container for managing parameter definitions.
Notes
-----
``_representedTypes`` is used to detect if this ``ParameterDefinitionCollection`` contains
definitions for only one type. If the collection only exists for 1 type, the lookup
(``__getitem__``) can short circuit O(n) logic for O(1) dictionary lookup.
"""
# Slots are not being used here as an attempt at optimization. Rather, they serve to add some
# needed rigidity to the parameter system.
__slots__ = ("_paramDefs", "_paramDefDict", "_representedTypes", "_locked")
def __init__(self):
self._paramDefs = list()
self._paramDefDict = dict()
self._representedTypes = set()
self._locked = False
def __iter__(self):
return iter(self._paramDefs)
def __len__(self):
return len(self._paramDefs)
def __getitem__(self, name):
"""Get a parameter by name.
Notes
-----
This method might break if the collection is for multiple composite types, and there exists
a parameter with the same name in multiple types.
"""
# O(1) lookup if there is only 1 type, could still raise a KeyError
if len(self._representedTypes) == 1:
return self._paramDefDict[name, next(iter(self._representedTypes))]
# "matches" only checks for the same name, while the add method checks both name and
# collectionType
matches = [pd for pd in self if pd.name == name]
if len(matches) != 1:
raise KeyError(
"Too {} parameters with the name `{}`. Matches:\n{}".format(
"many" if len(matches) > 1 else "few",
name,
"\n".join(str(pd) for pd in matches),
)
)
return matches[0]
[docs] def add(self, paramDef):
"""Add a :py:class:`Parameter` to this collection."""
assert not self._locked, "This ParameterDefinitionCollection has been locked."
self._paramDefs.append(paramDef)
self._paramDefDict[paramDef.name, paramDef.collectionType] = paramDef
self._representedTypes.add(paramDef.collectionType)
def _filter(self, filterFunc):
pdc = ParameterDefinitionCollection()
pdc.extend(filter(filterFunc, self._paramDefs))
return pdc
[docs] def items(self):
return self._paramDefDict.items()
[docs] def extend(self, other):
"""Grow a parameter definition collection by another parameter definition collection."""
assert (
not self._locked
), "This ParameterDefinitionCollection ({}) has been locked.".format(
self._representedTypes
)
assert self is not other
if other is None:
raise ValueError(
f"Cannot extend {self} with `None`. Ensure return value of parameter definitions "
"returns something."
)
for pd in other:
self.add(pd)
[docs] def inCategory(self, categoryName):
"""
Create a :py:class:`ParameterDefinitionCollection` that contains definitions that are in a
specific category.
"""
return self._filter(lambda pd: categoryName in pd.categories)
[docs] def atLocation(self, paramLoc):
"""
Make a param definition collection with all defs defined at a specific location.
Parameters can be defined at various locations within their container based on
:py:class:`ParamLocation`. This allows selection by those values.
"""
return self._filter(lambda pd: pd.atLocation(paramLoc))
[docs] def since(self, mask):
"""
Create a :py:class:`ParameterDefinitionCollection` that contains definitions that have been
modified since a specific set of actions.
"""
return self._filter(lambda pd: pd.assigned & mask)
[docs] def unchanged_since(self, mask):
"""
Create a :py:class:`ParameterDefinitionCollection` that contains definitions that have not
been modified since a specific set of actions. This is the complementary set of the
collection returned by `since`.
"""
return self._filter(lambda pd: not (pd.assigned & mask))
[docs] def forType(self, compositeType):
"""
Create a :py:class:`ParameterDefinitionCollection` that contains definitions for a
specific composite type.
"""
return self._filter(
lambda pd: issubclass(compositeType.paramCollectionType, pd.collectionType)
)
[docs] def resetAssignmentFlag(self, mask):
"""
Clear the `assigned` flag for a certain operation on all parameters.
These flags will get set by the param definition setters if they get changed again.
Notes
-----
See http://www.vipan.com/htdocs/bitwisehelp.html to understand the bitwise operations
"""
for pd in self._paramDefs:
pd.assigned &= ~mask
[docs] def setAssignmentFlag(self, mask):
for pd in self._paramDefs:
pd.assigned |= mask
[docs] def byNameAndType(self, name, compositeType):
"""Get a :py:class:`Parameter` by compositeType and name."""
return self._paramDefDict[name, compositeType.paramCollectionType]
[docs] def byNameAndCollectionType(self, name, collectionType):
"""Get a :py:class:`Parameter` by collectionType and name."""
return self._paramDefDict[name, collectionType]
@property
def categories(self):
"""Get the categories of all the :py:class:`~Parameter` instances within this collection."""
categories = set()
for paramDef in self:
categories |= paramDef.categories
return categories
@property
def names(self):
return [pd.name for pd in self]
[docs] def lock(self):
self._locked = True
@property
def locked(self):
return self._locked
[docs] def toWriteToDB(self, assignedMask: Optional[int] = None):
"""
Get a list of acceptable parameters to store to the database for a level of the data model.
.. impl:: Filter parameters to write to DB.
:id: I_ARMI_PARAM_DB
:implements: R_ARMI_PARAM_DB
This method is called when writing the parameters to the database file. It queries the
parameter's ``saveToDB`` attribute to ensure that this parameter is desired for saving
to the database file. It returns a list of parameters that should be included in the
database write operation.
Parameters
----------
assignedMask : int
A bitmask to down-filter which params to use based on how "stale" they are.
"""
mask = assignedMask or SINCE_ANYTHING
return [p for p in self if p.saveToDB and p.assigned & mask]
[docs] def createBuilder(self, *args, **kwargs):
"""
Create an associated object that can create definitions into this collection.
Using the returned ParameterBuilder will add all defined parameters to this
ParameterDefinitionCollection, using the passed arguments as defaults. Arguments should be
valid arguments to ``ParameterBuilder.__init__()``
"""
paramBuilder = ParameterBuilder(*args, **kwargs)
paramBuilder.associateParameterDefinitionCollection(self)
return paramBuilder
[docs]class ParameterBuilder:
"""Factory for creating Parameter and parameter properties."""
def __init__(
self,
location=ParamLocation.AVERAGE,
default=NoDefault,
categories=None,
saveToDB=True,
):
"""Create a :py:class:`ParameterBuilder`."""
self._entered = False
self._defaultLocation = location
self._defaultCategories = set(categories or []) # make sure it is always a set
self._defaultValue = default
self._assertDefaultIsProperType(default)
self._saveToDB = saveToDB
self._paramDefs = None
def __enter__(self):
self._entered = True
return self
def __exit__(self, exc_type, exc_value, tracebac):
if exc_type is not None:
# allow exceptions to be raised normally, to prevent confusing stack traces
return
self._entered = False
@staticmethod
def _assertDefaultIsProperType(default):
if default in (NoDefault, None) or isinstance(
default, (int, str, float, bool, Flags)
):
return
raise AssertionError(
"Cannot specify a default mutable type ({}) value to a parameter; all instances would "
"share the same list.".format(type(default))
)
[docs] def associateParameterDefinitionCollection(self, paramDefs):
"""
Associate this parameter factory with a specific ParameterDefinitionCollection.
Subsequent calls to defParam will automatically add the created ParameterDefinitions to this
ParameterDefinitionCollection. This results in a cleaner syntax when defining many
ParameterDefinitions.
"""
self._paramDefs = paramDefs
[docs] def defParam(
self,
name,
units,
description,
location=None,
saveToDB=NoDefault,
default=NoDefault,
setter=NoDefault,
categories=None,
serializer: Optional[Type[Serializer]] = None,
):
r"""Create a parameter as a property (with get/set) on a class.
Parameters
----------
name: str
the official name of the parameter
units: str
string representation of the units
description: str
a brief, but precise-as-possible description of what the parameter is used
for.
location: str
string representation of the location the attribute is applicable to, such
as average, max, etc.
saveToDB: bool
indicator as to whether the parameter should be written to the database. The
actual default is defined by the :py:class:`ParameterBuilder`, and is
:code:`True`.
default: immutable type
a default value for this parameter which must be an immutable type. If the
type is mutable, e.g. a list, dict, an exception should be raised, or
unknown behavior.
setter: None or callable
If ``None``, there is no direct way to set the parameter. If some other
callable method, (which may have the same name as the property!) then the
setter method is used instead.
categories: List of str
A list of categories to which this Parameter should belong. Categories are
typically used to engage special treatment for certain Parameters.
serializer: Optional subclass of Serializer
A class describing how the parameter data should be stored to the database.
This is usually only needed in exceptional cases where it is difficult to
store a parameter in a numpy array.
Notes
-----
It is not possible to initialize the parameter on the class this method would be used on,
because there is no instance (i.e. self) when this method is run. However, this method could
access a globally available set of definitions, if one existed.
"""
self._assertDefaultIsProperType(default)
if location is None and self._defaultLocation is None:
raise ParameterDefinitionError(
"The default location is not specified for {}; "
"a parameter-specific location is required.".format(self)
)
paramDef = Parameter(
name,
units=units,
description=description,
location=location or self._defaultLocation,
saveToDB=saveToDB if saveToDB is not NoDefault else self._saveToDB,
default=default if default is not NoDefault else self._defaultValue,
setter=setter,
categories=set(categories or []).union(self._defaultCategories),
serializer=serializer,
)
if self._paramDefs is not None:
self._paramDefs.add(paramDef)
return paramDef
# Container for all parameter definition collections that have been bound to an ArmiObject or
# subclass. These are added from the applyParameters() method on the ParameterCollection class.
ALL_DEFINITIONS = ParameterDefinitionCollection()