openmc.MaterialFilter,

openmc.CellFilter,

openmc.DistribcellFilter,

openmc.CellInstanceFilter,

# return the md5 hash of a file

h = hashlib.md5()

with path.open('rb') as file:

chunk = 0

while chunk != b'':

# read 32768 bytes at a time

chunk = file.read(32768)

h.update(chunk)

"""Get hash values for materials.xml and geometry.xml (or model.xml)"""

if model_path.is_file():

mat_xml_hash = hash_file(model_path)

geom_xml_hash = ""

elif (model_path / 'model.xml').exists():

mat_xml_hash = hash_file(model_path / 'model.xml')

geom_xml_hash = ""

else:

mat_xml_hash = hash_file(model_path / 'materials.xml')

geom_xml_hash = hash_file(model_path / 'geometry.xml')

finished = Signal(object, object, object)

error = Signal(str)

@Slot(object)

def generate_maps(self, work_item: PlotWorkItem):

try:

params = work_item.view_params

view_param = ViewParam(params["origin"], params["width"],

params["height"], params["h_res"])

view_param.h_res = params["h_res"]

view_param.v_res = params["v_res"]

view_param.basis = params["basis"]

view_param.level = params["level"]

view_param.color_overlaps = params["color_overlaps"]

ids_map = openmc.lib.id_map(view_param)

properties = openmc.lib.property_map(view_param)

self.finished.emit(work_item.view_params, ids_map, properties)

except Exception as exc:

plot_started = Signal()

plot_queued = Signal()

plot_finished = Signal(object, object, object, object)

plot_error = Signal(str)

plot_idle = Signal()

work_requested = Signal(object)

def __init__(self):

super().__init__()

self._thread = QThread()

self._worker = PlotWorker()

self._worker.moveToThread(self._thread)

self._thread.finished.connect(self._worker.deleteLater)

self.work_requested.connect(self._worker.generate_maps)

self._worker.finished.connect(self._on_worker_finished)

self._worker.error.connect(self._on_worker_error)

self._thread.start()

self._pending_request = None

self._in_flight_request = None

self._latest_view_params = None

@property

def latest_view_params(self):

return self._latest_view_params

@property

def is_busy(self):

return self._in_flight_request is not None

@property

def has_pending(self):

return self._pending_request is not None

def enqueue(self, view_snapshot, view_params):

request = PlotRequest(view_snapshot, view_params)

self._latest_view_params = view_params

if self._in_flight_request is None:

self._pending_request = request

self._start_next()

return

self._pending_request = request

self.plot_queued.emit()

def set_latest_view_params(self, view_params):

self._latest_view_params = view_params

def clear_pending(self):

self._pending_request = None

def wait_for_idle(self, timeout_ms: Optional[int] = None):

if not self.is_busy and not self.has_pending:

return True

loop = QEventLoop()

timer = None

if timeout_ms is not None:

timer = QTimer()

timer.setSingleShot(True)

timer.timeout.connect(loop.quit)

def on_idle():

loop.quit()

self.plot_idle.connect(on_idle)

if timer is not None:

timer.start(timeout_ms)

loop.exec()

self.plot_idle.disconnect(on_idle)

if timer is not None:

timer.stop()

return not self.is_busy and not self.has_pending

def shutdown(self):

self._pending_request = None

self._in_flight_request = None

if self._thread.isRunning():

self._thread.quit()

self._thread.wait()

def _start_next(self):

if self._pending_request is None:

return

self._in_flight_request = self._pending_request

self._pending_request = None

self.plot_started.emit()

work_item = PlotWorkItem(self._in_flight_request.view_params)

self.work_requested.emit(work_item)

@Slot(object, object, object)

def _on_worker_finished(self, view_params, ids_map, properties):

request = self._in_flight_request

self._in_flight_request = None

if request is not None:

self.plot_finished.emit(request.view_snapshot,

view_params, ids_map, properties)

if self._pending_request is not None:

self._start_next()

else:

self.plot_idle.emit()

@Slot(str)

def _on_worker_error(self, message):

self._in_flight_request = None

self.plot_error.emit(message)

if self._pending_request is not None:

self._start_next()

else:

"""Geometry and plot settings for OpenMC Plot Explorer model

def __init__(self, use_settings_pkl, model_path, default_res):

""" Initialize PlotModel class attributes """

# Retrieve OpenMC Cells/Materials

self.modelCells = openmc.lib.cells

self.modelMaterials = openmc.lib.materials

self.max_universe_levels = openmc.lib._coord_levels()

# Cell/Material ID by coordinates

self.ids = None

# Return values from id_map and property_map

self.ids_map = None

self.properties = None

self.map_view_params = None

self.version = __version__

# default statepoint value

self._statepoint = None

# default tally/filter info

self.appliedFilters = ()

self.appliedScores = ()

self.appliedNuclides = ()

self.previousViews = []

self.subsequentViews = []

self.defaultView = self.getDefaultView(default_res)

# Source site defaults

self.sourceSitesApplyTolerance = False

self.sourceSitesTolerance = 0.1 # cm

self.sourceSitesColor = (0, 0, 255)

self.sourceSitesVisible = True

self.sourceSites = None

if model_path.is_file():

settings_pkl = model_path.with_name('plot_settings.pkl')

else:

settings_pkl = model_path / 'plot_settings.pkl'

if use_settings_pkl and settings_pkl.is_file():

with settings_pkl.open('rb') as file:

try:

data = pickle.load(file)

except AttributeError:

msg_box = QMessageBox()

msg = "WARNING: previous plot settings are in an incompatible format. " +\

"They will be ignored."

msg_box.setText(msg)

msg_box.setIcon(QMessageBox.Warning)

msg_box.setStandardButtons(QMessageBox.Ok)

msg_box.exec()

self.currentView = copy.deepcopy(self.defaultView)

else:

restore_domains = False

# check GUI version

if data['version'] != self.version:

print("WARNING: previous plot settings are for a different "

"version of the GUI. They will be ignored.")

wrn_msg = "Existing version: {}, Current GUI version: {}"

print(wrn_msg.format(data['version'], self.version))

view = None

else:

view = data['currentView']

# get materials.xml and geometry.xml hashes to

# restore additional settings if possible

mat_xml_hash, geom_xml_hash = hash_model(model_path)

if mat_xml_hash == data['mat_xml_hash'] and \

geom_xml_hash == data['geom_xml_hash']:

restore_domains = True

# restore statepoint file

try:

self.statepoint = data['statepoint']

except OSError:

msg_box = QMessageBox()

msg = "Could not open statepoint file: \n\n {} \n"

msg_box.setText(msg.format(self.model.statepoint.filename))

msg_box.setIcon(QMessageBox.Warning)

msg_box.setStandardButtons(QMessageBox.Ok)

msg_box.exec()

self.statepoint = None

self.currentView = PlotView(restore_view=view,

restore_domains=restore_domains,

default_res=default_res)

else:

self.currentView = copy.deepcopy(self.defaultView)

self.activeView = copy.deepcopy(self.currentView)

self.plot_manager = PlotManager()

def openStatePoint(self, filename):

self.statepoint = StatePointModel(filename, open_file=True)

@property

def statepoint(self):

return self._statepoint

@statepoint.setter

def statepoint(self, statepoint):

if statepoint is None:

self._statepoint = None

elif isinstance(statepoint, StatePointModel):

self._statepoint = statepoint

elif isinstance(statepoint, str):

self._statepoint = StatePointModel(statepoint, open_file=True)

else:

raise TypeError("Invalid statepoint object")

if self._statepoint and not self._statepoint.is_open:

self._statepoint.open()

def getDefaultView(self, default_res):

""" Generates default PlotView instance for OpenMC geometry

lower_left, upper_right = openmc.lib.global_bounding_box()

# Check for valid bounding_box dimensions

if -np.inf not in lower_left[:2] and np.inf not in upper_right[:2]:

xcenter = (upper_right[0] + lower_left[0])/2

width = abs(upper_right[0] - lower_left[0]) * 1.005

ycenter = (upper_right[1] + lower_left[1])/2

height = abs(upper_right[1] - lower_left[1]) * 1.005

else:

xcenter, ycenter, width, height = (0.00, 0.00, 25, 25)

if lower_left[2] != -np.inf and upper_right[2] != np.inf:

zcenter = (upper_right[2] + lower_left[2])/2

else:

zcenter = 0.00

default = PlotView([xcenter, ycenter, zcenter], width, height,

default_res=default_res)

return default

def resetColors(self):

""" Reset colors to those generated in the default view """

self.activeView.cells = self.defaultView.cells

self.activeView.materials = self.defaultView.materials

def view_params_payload(self, view: "PlotView"):

vp = view.view_params

return {

"origin": tuple(vp.origin),

"width": float(vp.width),

"height": float(vp.height),

"h_res": int(vp.h_res),

"v_res": int(vp.v_res),

"basis": str(vp.basis),

"level": int(vp.level),

"color_overlaps": bool(vp.color_overlaps),

}

def can_reuse_maps(self, view: "PlotView"):

if self.ids_map is None or self.properties is None:

return False

return self.map_view_params == self.view_params_payload(view)

def makePlot(self, view: Optional["PlotView"] = None,

ids_map=None, properties=None):

""" Generate new plot image from active view settings

if view is None:

view = self.activeView

# update/call maps under 2 circumstances

# 1. this is the intial plot (ids_map/properties are None)

# 2. The active (desired) view differs from the current view parameters

if ids_map is None or properties is None:

if (self.currentView.view_params != view.view_params) or \

(self.ids_map is None) or (self.properties is None):

self.ids_map = openmc.lib.id_map(view.view_params)

self.properties = openmc.lib.property_map(view.view_params)

self.map_view_params = self.view_params_payload(view)

else:

self.ids_map = ids_map

self.properties = properties

self.map_view_params = self.view_params_payload(view)

# update current view

cv = self.currentView = copy.deepcopy(view)

# set model ids based on domain

if cv.colorby == 'cell':

self.ids = self.cell_ids

domain = cv.cells

source = self.modelCells

else:

self.ids = self.mat_ids

domain = cv.materials

source = self.modelMaterials

# generate colors if not present

for cell_id, cell in cv.cells.items():

if cell.color is None:

cv.cells.set_color(cell_id, random_rgb())

for mat_id, mat in cv.materials.items():

if mat.color is None:

cv.material.set_color(mat_id, random_rgb())

# construct image data

domain[_OVERLAP] = DomainView(_OVERLAP, "Overlap", cv.overlap_color)

domain[_NOT_FOUND] = DomainView(_NOT_FOUND, "Not Found", cv.domainBackground)

u, inv = np.unique(self.ids, return_inverse=True)

image = np.array([domain[id].color for id in u])[inv]

image.shape = (cv.v_res, cv.h_res, 3)

if cv.masking:

for id, dom in domain.items():

if dom.masked:

image[self.ids == int(id)] = cv.maskBackground

if cv.highlighting:

for id, dom in domain.items():

if dom.highlight:

image[self.ids == int(id)] = cv.highlightBackground

# set model image

self.image = image

# tally data

self.tally_data = None

self.properties[self.properties < 0.0] = np.nan

self.temperatures = self.properties[..., _PROPERTY_INDICES['temperature']]

self.densities = self.properties[..., _PROPERTY_INDICES['density']]

minmax = {}

for prop in _MODEL_PROPERTIES:

idx = _PROPERTY_INDICES[prop]

prop_data = self.properties[:, :, idx]

minmax[prop] = (np.min(np.nan_to_num(prop_data)),

np.max(np.nan_to_num(prop_data)))

cv.data_minmax = minmax

if self.activeView.view_params == cv.view_params:

self.activeView.data_minmax = minmax

def undo(self):

""" Revert to previous PlotView instance. Re-generate plot image """

if self.previousViews:

self.subsequentViews.append(copy.deepcopy(self.currentView))

self.activeView = self.previousViews.pop()

def redo(self):

""" Revert to subsequent PlotView instance. Re-generate plot image """

if self.subsequentViews:

self.storeCurrent()

self.activeView = self.subsequentViews.pop()

def getExternalSourceSites(self, n=100):

"""Plot source sites from a source file

if n == 0:

self.source_sites = None

return

sites = openmc.lib.sample_external_source(n)

self.sourceSites = np.array([s.r for s in sites[:n]], dtype=float)

def storeCurrent(self):

""" Add current view to previousViews list """

self.previousViews.append(copy.deepcopy(self.currentView))

def create_tally_image(self, view: Optional[PlotView] = None):

"""

if view is None:

view = self.currentView

tally_id = view.selectedTally

scores = self.appliedScores

nuclides = self.appliedNuclides

tally_selected = view.selectedTally is not None

tally_visible = view.tallyDataVisible

visible_selection = scores and nuclides

if not tally_selected or not tally_visible or not visible_selection:

return (None, None, None, None, None)

tally = self.statepoint.tallies[tally_id]

tally_value = _TALLY_VALUES[view.tallyValue]

# check score units

units = {_SCORE_UNITS.get(score, _REACTION_UNITS) for score in scores}

if len(units) != 1:

msg_box = QMessageBox()

unit_str = " ".join(units)

msg = "The scores selected have incompatible units:\n"

for unit in units:

msg += " - {}\n".format(unit)

msg_box.setText(msg)

msg_box.setIcon(QMessageBox.Information)

msg_box.setStandardButtons(QMessageBox.Ok)

msg_box.exec()

return (None, None, None, None, None)

units_out = list(units)[0]

contains_distribcell = tally.contains_filter(openmc.DistribcellFilter)

contains_cellinstance = tally.contains_filter(openmc.CellInstanceFilter)

if tally.contains_filter(openmc.MeshFilter):

# Check for volume normalization in order to change units

if view.tallyVolumeNorm:

units_out = _SCORE_UNITS_VOL.get(scores[0], _REACTION_UNITS_VOL)

if tally_value == 'rel_err':

# Get both the std. dev. data and mean data to create the

# relative error data

mean_data = self._create_tally_mesh_image(

tally, 'mean', scores, nuclides, view)[0]

std_dev_data = self._create_tally_mesh_image(

tally, 'std_dev', scores, nuclides, view)[0]

with np.errstate(divide='ignore', invalid='ignore'):

image_data = 100.0 * std_dev_data / mean_data

if np.isnan(image_data).all():

data_min, data_max = 0., 1.

else:

data_min = np.nanmin(image_data)

data_max = np.nanmax(image_data)

return image_data, None, data_min, data_max, '% error'

else:

image = self._create_tally_mesh_image(tally,

tally_value,

scores,

nuclides,

view)

return image + (units_out,)

elif contains_distribcell or contains_cellinstance:

if tally_value == 'rel_err':

mean_data = self._create_distribcell_image(

tally, 'mean', scores, nuclides, contains_cellinstance)

std_dev_data = self._create_distribcell_image(

tally, 'std_dev', scores, nuclides)

image_data = 100 * np.divide(

std_dev_data[0], mean_data[0],

out=np.zeros_like(mean_data[0]),

where=mean_data != 0

)

data_min = np.min(image_data)

data_max = np.max(image_data)

return image_data, None, data_min, data_max, '% error'

else:

image = self._create_distribcell_image(

tally, tally_value, scores, nuclides, contains_cellinstance)

return image + (units_out,)

else:

# same as above, get the std. dev. data

# and mean date to produce the relative error data

if tally_value == 'rel_err':

mean_data = self._create_tally_domain_image(tally,

'mean',

scores,

nuclides)

std_dev_data = self._create_tally_domain_image(tally,

'std_dev',

scores,

nuclides)

image_data = 100 * np.divide(std_dev_data[0],

mean_data[0],

out=np.zeros_like(mean_data[0]),

where=mean_data != 0)

# adjust for NaNs in bins without tallies

image_data = np.nan_to_num(image_data,

nan=0.0,

posinf=0.0,

neginf=0.0)

extents = mean_data[1]

data_min = np.min(image_data)

data_max = np.max(image_data)

return image_data, extents, data_min, data_max, '% error'

else:

image = self._create_tally_domain_image(tally,

tally_value,

scores,

nuclides)

return image + (units_out,)

def _create_tally_domain_image(self, tally, tally_value, scores, nuclides, view=None):

# data resources used throughout

if view is None:

view = self.currentView

data = tally.get_reshaped_data(tally_value)

data_out = np.full(self.ids.shape, -1.0)

def _do_op(array, tally_value, ax=0):

if tally_value == 'mean':

return np.sum(array, axis=ax)

elif tally_value == 'std_dev':

return np.sqrt(np.sum(array**2, axis=ax))

# data structure for tracking which spatial

# filter bins are enabled

spatial_filter_bins = defaultdict(list)

n_spatial_filters = 0

for tally_filter in tally.filters:

if tally_filter in self.appliedFilters:

selected_bins = self.appliedFilters[tally_filter]

if type(tally_filter) in _SPATIAL_FILTERS:

spatial_filter_bins[tally_filter] = selected_bins

n_spatial_filters += 1

else:

slc = [slice(None)] * len(data.shape)

slc[n_spatial_filters] = selected_bins

slc = tuple(slc)

data = _do_op(data[slc], tally_value, n_spatial_filters)

else:

data[:] = 0.0

data = _do_op(data, tally_value, n_spatial_filters)

# filter by selected scores

selected_scores = []

for idx, score in enumerate(tally.scores):

if score in scores:

selected_scores.append(idx)

data = _do_op(data[..., np.array(selected_scores)], tally_value, -1)

# filter by selected nuclides

selected_nuclides = []

for idx, nuclide in enumerate(tally.nuclides):

if nuclide in nuclides:

selected_nuclides.append(idx)

data = _do_op(data[..., np.array(selected_nuclides)], tally_value, -1)

# get data limits

data_min = np.min(data)

data_max = np.max(data)

# for all combinations of spatial bins, create a mask

# and set image data values

spatial_filters = list(spatial_filter_bins.keys())

spatial_bins = list(spatial_filter_bins.values())

for bin_indices in itertools.product(*spatial_bins):

# look up the tally value

tally_val = data[bin_indices]

if tally_val == 0.0:

continue

# generate a mask with the correct size

mask = np.full(self.ids.shape, True, dtype=bool)

for tally_filter, bin_idx in zip(spatial_filters, bin_indices):

bin = tally_filter.bins[bin_idx]

if isinstance(tally_filter, openmc.CellFilter):

mask &= self.cell_ids == bin

elif isinstance(tally_filter, openmc.MaterialFilter):

mask &= self.mat_ids == bin

elif isinstance(tally_filter, openmc.UniverseFilter):

# get the statepoint summary

univ_cells = self.statepoint.universes[bin].cells

for cell in univ_cells:

mask &= self.cell_ids == cell

# set image data values

data_out[mask] = tally_val

# mask out invalid values

image_data = np.ma.masked_where(data_out < 0.0, data_out)

return image_data, None, data_min, data_max

def _create_distribcell_image(self, tally, tally_value, scores, nuclides, cellinstance=False):

# Get flattened array of tally results

data = tally.get_values(scores=scores, nuclides=nuclides, value=tally_value)

data = data.flatten()

# Create an empty array of appropriate shape for image

image_data = np.full_like(self.ids, np.nan, dtype=float)

# Determine mapping of cell IDs to list of (instance, tally value).

if cellinstance:

f = tally.find_filter(openmc.CellInstanceFilter)

cell_id_to_inst_value = defaultdict(list)

for value, (cell_id, instance) in zip(data, f.bins):

cell_id_to_inst_value[cell_id].append((instance, value))

else:

f = tally.find_filter(openmc.DistribcellFilter)

cell_id_to_inst_value = {f.bins[0]: list(enumerate(data))}

for cell_id, value_list in cell_id_to_inst_value.items():

# Get mask for each relevant cell

cell_id_mask = (self.cell_ids == cell_id)

# For each cell, iterate over instances and corresponding tally

# values and set any matching pixels

for instance, value in value_list:

instance_mask = (self.instances == instance)

image_data[cell_id_mask & instance_mask] = value

data_min = np.min(data)

data_max = np.max(data)

image_data = np.ma.masked_where(image_data < 0.0, image_data)

return image_data, None, data_min, data_max

def cpp_mesh_ids(self):

return list(openmc.lib.meshes.keys())

def mesh_plot_bins(self, mesh_id, view: PlotView = None, translation: tuple[float, float, float] = None):

mesh = openmc.lib.meshes[mesh_id]

if view is None:

view = self.currentView

if translation is None:

origin = view.origin

else:

origin = (view.origin[0] - translation[0],

view.origin[1] - translation[1],

view.origin[2] - translation[2])

mesh_bins = mesh.get_plot_bins(

origin=origin,

width=(view.width, view.height),

basis=view.basis,

pixels=(view.h_res, view.v_res),

)

return mesh_bins

def _create_tally_mesh_image(

self, tally: openmc.Tally, tally_value: TallyValueType,

scores: Tuple[str], nuclides: Tuple[str], view: PlotView = None

):

# some variables used throughout

if view is None:

view = self.currentView

mesh_filter = tally.find_filter(openmc.MeshFilter)

mesh = mesh_filter.mesh

def _do_op(array, tally_value, ax=0):

if tally_value == 'mean':

return np.sum(array, axis=ax)

elif tally_value == 'std_dev':

return np.sqrt(np.sum(array**2, axis=ax))

# start with reshaped data

data = tally.get_reshaped_data(tally_value)

# move mesh axes to the end of the filters

filter_idx = [type(filter) for filter in tally.filters].index(openmc.MeshFilter)

data = np.moveaxis(data, filter_idx, -1)

# sum over the rest of the tally filters

for tally_filter in tally.filters:

if type(tally_filter) == openmc.MeshFilter:

continue

selected_bins = self.appliedFilters[tally_filter]

if selected_bins:

# sum filter data for the selected bins

data = data[np.array(selected_bins)].sum(axis=0)

else:

# if the filter is completely unselected,

# set all of its data to zero and remove the axis

data[:] = 0.0

data = _do_op(data, tally_value)

# filter by selected nuclides

if not nuclides:

data = 0.0

selected_nuclides = []

for idx, nuclide in enumerate(tally.nuclides):

if nuclide in nuclides:

selected_nuclides.append(idx)

data = _do_op(data[np.array(selected_nuclides)], tally_value)

# filter by selected scores

if not scores:

data = 0.0

selected_scores = []

for idx, score in enumerate(tally.scores):

if score in scores:

selected_scores.append(idx)

data = _do_op(data[np.array(selected_scores)], tally_value)

# Get mesh bins from openmc.lib

mesh_cpp = openmc.lib.meshes[mesh.id]

mesh_bins = self.mesh_plot_bins(mesh.id, view, mesh_filter.translation)

# Apply volume normalization

if view.tallyVolumeNorm:

data /= mesh_cpp.volumes

# set image data

image_data = np.full_like(self.ids, np.nan, dtype=float)

mask = (mesh_bins >= 0)

image_data[mask] = data[mesh_bins[mask]]

# get dataset's min/max

data_min = np.min(data)

data_max = np.max(data)

return image_data, None, data_min, data_max

@property

def cell_ids(self):

return self.ids_map[:, :, 0]

@property

def instances(self):

return self.ids_map[:, :, 1]

@property

def mat_ids(self):

"""Viewer settings that are needed for _PlotBase and are independent

def __init__(self, origin=(0, 0, 0), width=10, height=10, default_res=1000):

"""Initialize ViewParam attributes"""

super().__init__()