mirror of
https://github.com/openmc-dev/openmc.git
synced 2026-07-21 14:35:27 -04:00
346 lines
12 KiB
Python
346 lines
12 KiB
Python
from difflib import unified_diff
|
|
import filecmp
|
|
import glob
|
|
import hashlib
|
|
from optparse import OptionParser
|
|
import os
|
|
import shutil
|
|
import sys
|
|
|
|
import numpy as np
|
|
import openmc
|
|
from openmc.examples import pwr_core
|
|
|
|
from tests.regression_tests import config
|
|
|
|
|
|
class TestHarness(object):
|
|
"""General class for running OpenMC regression tests."""
|
|
|
|
def __init__(self, statepoint_name):
|
|
self._sp_name = statepoint_name
|
|
|
|
def main(self):
|
|
"""Accept commandline arguments and either run or update tests."""
|
|
if config['update']:
|
|
self.update_results()
|
|
else:
|
|
self.execute_test()
|
|
|
|
def execute_test(self):
|
|
"""Run OpenMC with the appropriate arguments and check the outputs."""
|
|
try:
|
|
self._run_openmc()
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
self._write_results(results)
|
|
self._compare_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def update_results(self):
|
|
"""Update the results_true using the current version of OpenMC."""
|
|
try:
|
|
self._run_openmc()
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
self._write_results(results)
|
|
self._overwrite_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def _run_openmc(self):
|
|
if config['mpi']:
|
|
mpi_args = [config['mpiexec'], '-n', config['mpi_np']]
|
|
openmc.run(openmc_exec=config['exe'], mpi_args=mpi_args)
|
|
else:
|
|
openmc.run(openmc_exec=config['exe'])
|
|
|
|
def _test_output_created(self):
|
|
"""Make sure statepoint.* and tallies.out have been created."""
|
|
statepoint = glob.glob(self._sp_name)
|
|
assert len(statepoint) == 1, 'Either multiple or no statepoint files' \
|
|
' exist.'
|
|
assert statepoint[0].endswith('h5'), \
|
|
'Statepoint file is not a HDF5 file.'
|
|
if os.path.exists('tallies.xml'):
|
|
assert os.path.exists('tallies.out'), \
|
|
'Tally output file does not exist.'
|
|
|
|
def _get_results(self, hash_output=False):
|
|
"""Digest info in the statepoint and return as a string."""
|
|
# Read the statepoint file.
|
|
statepoint = glob.glob(self._sp_name)[0]
|
|
with openmc.StatePoint(statepoint) as sp:
|
|
# Write out k-combined.
|
|
outstr = 'k-combined:\n'
|
|
form = '{0:12.6E} {1:12.6E}\n'
|
|
outstr += form.format(sp.k_combined.n, sp.k_combined.s)
|
|
|
|
# Write out tally data.
|
|
for i, tally_ind in enumerate(sp.tallies):
|
|
tally = sp.tallies[tally_ind]
|
|
results = np.zeros((tally.sum.size * 2, ))
|
|
results[0::2] = tally.sum.ravel()
|
|
results[1::2] = tally.sum_sq.ravel()
|
|
results = ['{0:12.6E}'.format(x) for x in results]
|
|
|
|
outstr += 'tally {}:\n'.format(i + 1)
|
|
outstr += '\n'.join(results) + '\n'
|
|
|
|
# Hash the results if necessary.
|
|
if hash_output:
|
|
sha512 = hashlib.sha512()
|
|
sha512.update(outstr.encode('utf-8'))
|
|
outstr = sha512.hexdigest()
|
|
|
|
return outstr
|
|
|
|
def _write_results(self, results_string):
|
|
"""Write the results to an ASCII file."""
|
|
with open('results_test.dat', 'w') as fh:
|
|
fh.write(results_string)
|
|
|
|
def _overwrite_results(self):
|
|
"""Overwrite the results_true with the results_test."""
|
|
shutil.copyfile('results_test.dat', 'results_true.dat')
|
|
|
|
def _compare_results(self):
|
|
"""Make sure the current results agree with the _true standard."""
|
|
compare = filecmp.cmp('results_test.dat', 'results_true.dat')
|
|
if not compare:
|
|
os.rename('results_test.dat', 'results_error.dat')
|
|
assert compare, 'Results do not agree.'
|
|
|
|
def _cleanup(self):
|
|
"""Delete statepoints, tally, and test files."""
|
|
output = glob.glob('statepoint.*.h5')
|
|
output += ['tallies.out', 'results_test.dat', 'summary.h5']
|
|
output += glob.glob('volume_*.h5')
|
|
for f in output:
|
|
if os.path.exists(f):
|
|
os.remove(f)
|
|
|
|
|
|
class HashedTestHarness(TestHarness):
|
|
"""Specialized TestHarness that hashes the results."""
|
|
|
|
def _get_results(self):
|
|
"""Digest info in the statepoint and return as a string."""
|
|
return super()._get_results(True)
|
|
|
|
|
|
class CMFDTestHarness(TestHarness):
|
|
"""Specialized TestHarness for running OpenMC CMFD tests."""
|
|
|
|
def __init__(self, statepoint_name, cmfd_run):
|
|
super().__init__(statepoint_name)
|
|
self._create_cmfd_result_str(cmfd_run)
|
|
|
|
def _create_cmfd_result_str(self, cmfd_run):
|
|
"""Create CMFD result string from variables of CMFDRun instance"""
|
|
outstr = 'cmfd indices\n'
|
|
outstr += '\n'.join(['{:.6E}'.format(x) for x in cmfd_run.indices])
|
|
outstr += '\nk cmfd\n'
|
|
outstr += '\n'.join(['{:.6E}'.format(x) for x in cmfd_run.k_cmfd])
|
|
outstr += '\ncmfd entropy\n'
|
|
outstr += '\n'.join(['{:.6E}'.format(x) for x in cmfd_run.entropy])
|
|
outstr += '\ncmfd balance\n'
|
|
outstr += '\n'.join(['{:.5E}'.format(x) for x in cmfd_run.balance])
|
|
outstr += '\ncmfd dominance ratio\n'
|
|
outstr += '\n'.join(['{:.3E}'.format(x) for x in cmfd_run.dom])
|
|
outstr += '\ncmfd openmc source comparison\n'
|
|
outstr += '\n'.join(['{:.6E}'.format(x) for x in cmfd_run.src_cmp])
|
|
outstr += '\ncmfd source\n'
|
|
cmfdsrc = np.reshape(cmfd_run.cmfd_src, np.product(cmfd_run.indices),
|
|
order='F')
|
|
outstr += '\n'.join(['{:.6E}'.format(x) for x in cmfdsrc])
|
|
outstr += '\n'
|
|
self._cmfdrun_results = outstr
|
|
|
|
def execute_test(self):
|
|
"""Don't call _run_openmc as OpenMC will be called through C API for
|
|
CMFD tests, and write CMFD results that were passsed as argument
|
|
|
|
"""
|
|
try:
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
results += self._cmfdrun_results
|
|
self._write_results(results)
|
|
self._compare_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def update_results(self):
|
|
"""Don't call _run_openmc as OpenMC will be called through C API for
|
|
CMFD tests, and write CMFD results that were passsed as argument
|
|
|
|
"""
|
|
try:
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
results += self._cmfdrun_results
|
|
self._write_results(results)
|
|
self._overwrite_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def _cleanup(self):
|
|
"""Delete output files for numpy matrices and flux vectors."""
|
|
super()._cleanup()
|
|
output = ['loss.npz', 'loss.dat', 'prod.npz', 'prod.dat',
|
|
'fluxvec.npy', 'fluxvec.dat']
|
|
for f in output:
|
|
if os.path.exists(f):
|
|
os.remove(f)
|
|
|
|
|
|
class ParticleRestartTestHarness(TestHarness):
|
|
"""Specialized TestHarness for running OpenMC particle restart tests."""
|
|
|
|
def _run_openmc(self):
|
|
# Set arguments
|
|
args = {'openmc_exec': config['exe']}
|
|
if config['mpi']:
|
|
args['mpi_args'] = [config['mpiexec'], '-n', config['mpi_np']]
|
|
|
|
# Initial run
|
|
openmc.run(**args)
|
|
|
|
# Run particle restart
|
|
args.update({'restart_file': self._sp_name})
|
|
openmc.run(**args)
|
|
|
|
def _test_output_created(self):
|
|
"""Make sure the restart file has been created."""
|
|
particle = glob.glob(self._sp_name)
|
|
assert len(particle) == 1, 'Either multiple or no particle restart ' \
|
|
'files exist.'
|
|
assert particle[0].endswith('h5'), \
|
|
'Particle restart file is not a HDF5 file.'
|
|
|
|
def _get_results(self):
|
|
"""Digest info in the statepoint and return as a string."""
|
|
# Read the particle restart file.
|
|
particle = glob.glob(self._sp_name)[0]
|
|
p = openmc.Particle(particle)
|
|
|
|
# Write out the properties.
|
|
outstr = ''
|
|
outstr += 'current batch:\n'
|
|
outstr += "{0:12.6E}\n".format(p.current_batch)
|
|
outstr += 'current generation:\n'
|
|
outstr += "{0:12.6E}\n".format(p.current_generation)
|
|
outstr += 'particle id:\n'
|
|
outstr += "{0:12.6E}\n".format(p.id)
|
|
outstr += 'run mode:\n'
|
|
outstr += "{0}\n".format(p.run_mode)
|
|
outstr += 'particle weight:\n'
|
|
outstr += "{0:12.6E}\n".format(p.weight)
|
|
outstr += 'particle energy:\n'
|
|
outstr += "{0:12.6E}\n".format(p.energy)
|
|
outstr += 'particle xyz:\n'
|
|
outstr += "{0:12.6E} {1:12.6E} {2:12.6E}\n".format(p.xyz[0], p.xyz[1],
|
|
p.xyz[2])
|
|
outstr += 'particle uvw:\n'
|
|
outstr += "{0:12.6E} {1:12.6E} {2:12.6E}\n".format(p.uvw[0], p.uvw[1],
|
|
p.uvw[2])
|
|
|
|
return outstr
|
|
|
|
|
|
class PyAPITestHarness(TestHarness):
|
|
def __init__(self, statepoint_name, model=None):
|
|
super().__init__(statepoint_name)
|
|
if model is None:
|
|
self._model = pwr_core()
|
|
else:
|
|
self._model = model
|
|
self._model.plots = []
|
|
|
|
|
|
def main(self):
|
|
"""Accept commandline arguments and either run or update tests."""
|
|
if config['build_inputs']:
|
|
self._build_inputs()
|
|
elif config['update']:
|
|
self.update_results()
|
|
else:
|
|
self.execute_test()
|
|
|
|
def execute_test(self):
|
|
"""Build input XMLs, run OpenMC, and verify correct results."""
|
|
try:
|
|
self._build_inputs()
|
|
inputs = self._get_inputs()
|
|
self._write_inputs(inputs)
|
|
self._compare_inputs()
|
|
self._run_openmc()
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
self._write_results(results)
|
|
self._compare_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def update_results(self):
|
|
"""Update results_true.dat and inputs_true.dat"""
|
|
try:
|
|
self._build_inputs()
|
|
inputs = self._get_inputs()
|
|
self._write_inputs(inputs)
|
|
self._overwrite_inputs()
|
|
self._run_openmc()
|
|
self._test_output_created()
|
|
results = self._get_results()
|
|
self._write_results(results)
|
|
self._overwrite_results()
|
|
finally:
|
|
self._cleanup()
|
|
|
|
def _build_inputs(self):
|
|
"""Write input XML files."""
|
|
self._model.export_to_xml()
|
|
|
|
def _get_inputs(self):
|
|
"""Return a hash digest of the input XML files."""
|
|
xmls = ['geometry.xml', 'materials.xml', 'settings.xml',
|
|
'tallies.xml', 'plots.xml']
|
|
return ''.join([open(fname).read() for fname in xmls
|
|
if os.path.exists(fname)])
|
|
|
|
def _write_inputs(self, input_digest):
|
|
"""Write the digest of the input XMLs to an ASCII file."""
|
|
with open('inputs_test.dat', 'w') as fh:
|
|
fh.write(input_digest)
|
|
|
|
def _overwrite_inputs(self):
|
|
"""Overwrite inputs_true.dat with inputs_test.dat"""
|
|
shutil.copyfile('inputs_test.dat', 'inputs_true.dat')
|
|
|
|
def _compare_inputs(self):
|
|
"""Make sure the current inputs agree with the _true standard."""
|
|
compare = filecmp.cmp('inputs_test.dat', 'inputs_true.dat')
|
|
if not compare:
|
|
os.rename('inputs_test.dat', 'inputs_error.dat')
|
|
for line in unified_diff(open('inputs_true.dat', 'r').readlines(),
|
|
open('inputs_error.dat', 'r').readlines(),
|
|
'inputs_true.dat', 'inputs_error.dat'):
|
|
print(line, end='')
|
|
assert compare, 'Input files are broken.'
|
|
|
|
def _cleanup(self):
|
|
"""Delete XMLs, statepoints, tally, and test files."""
|
|
super()._cleanup()
|
|
output = ['materials.xml', 'geometry.xml', 'settings.xml',
|
|
'tallies.xml', 'plots.xml', 'inputs_test.dat']
|
|
for f in output:
|
|
if os.path.exists(f):
|
|
os.remove(f)
|
|
|
|
|
|
class HashedPyAPITestHarness(PyAPITestHarness):
|
|
def _get_results(self):
|
|
"""Digest info in the statepoint and return as a string."""
|
|
return super()._get_results(True)
|