Source code for wntr.sim.epanet
"""The EPANET simulator.
"""
from wntr.sim.core import WaterNetworkSimulator
from wntr.network.io import write_inpfile
import wntr.epanet.io
import warnings
import logging
logger = logging.getLogger(__name__)
try:
import wntr.epanet.toolkit
except ImportError as e:
print('{}'.format(e))
logger.critical('%s',e)
raise ImportError('Error importing epanet toolkit while running epanet simulator. '
'Make sure libepanet is installed and added to path.')
[docs]
class EpanetSimulator(WaterNetworkSimulator):
"""
Fast EPANET simulator class.
Use the EPANET DLL to run an INP file as-is, and read the results from the
binary output file. Multiple water quality simulations are still possible
using the WQ keyword in the run_sim function. Hydraulics will be stored and
saved to a file. This file will not be deleted by default, nor will any
binary files be deleted.
The reason this is considered a "fast" simulator is due to the fact that there
is no looping within Python. The "ENsolveH" and "ENsolveQ" toolkit
functions are used instead.
.. note::
WNTR now includes access to both the EPANET 2.0.12 and EPANET 2.2 toolkit libraries.
By default, version 2.2 will be used.
Parameters
----------
wn : WaterNetworkModel
Water network model
reader : wntr.epanet.io.BinFile derived object
Defaults to None, which will create a new wntr.epanet.io.BinFile object with
the results_types specified as an init option. Otherwise, a fully
result_types : dict
Defaults to None, or all results. Otherwise, is a keyword dictionary to pass to
the reader to specify what results should be saved.
.. seealso::
wntr.epanet.io.BinFile
"""
[docs]
def __init__(self, wn, reader=None, result_types=None):
WaterNetworkSimulator.__init__(self, wn)
self.reader = reader
self.prep_time_before_main_loop = 0.0
if self.reader is None:
self.reader = wntr.epanet.io.BinFile(result_types=result_types)
[docs]
def run_sim(self, file_prefix='temp', save_hyd=False, use_hyd=False, hydfile=None,
version=2.2, convergence_error=False):
"""
Run the EPANET simulator.
Runs the EPANET simulator through the compiled toolkit DLL. Can use/save hydraulics
to allow for separate WQ runs.
.. note::
By default, WNTR now uses the EPANET 2.2 toolkit as the engine for the EpanetSimulator.
To force usage of the older EPANET 2.0 toolkit, use the ``version`` command line option.
Note that if the demand_model option is set to PDD, then a warning will be issued, as
EPANET 2.0 does not support such analysis.
Parameters
----------
file_prefix : str
Default prefix is "temp". All files (.inp, .bin/.out, .hyd, .rpt) use this prefix
use_hyd : bool
Will load hydraulics from ``file_prefix + '.hyd'`` or from file specified in `hydfile_name`
save_hyd : bool
Will save hydraulics to ``file_prefix + '.hyd'`` or to file specified in `hydfile_name`
hydfile : str
Optionally specify a filename for the hydraulics file other than the `file_prefix`
version : float, {2.0, **2.2**}
Optionally change the version of the EPANET toolkit libraries. Valid choices are
either 2.2 (the default if no argument provided) or 2.0.
convergence_error: bool (optional)
If convergence_error is True, an error will be raised if the
simulation does not converge. If convergence_error is False, partial results are returned,
a warning will be issued, and results.error_code will be set to 0
if the simulation does not converge. Default = False.
"""
if isinstance(version, str):
version = float(version)
inpfile = file_prefix + '.inp'
write_inpfile(self._wn, inpfile, units=self._wn.options.hydraulic.inpfile_units, version=version)
enData = wntr.epanet.toolkit.ENepanet(version=version)
rptfile = file_prefix + '.rpt'
outfile = file_prefix + '.bin'
if hydfile is None:
hydfile = file_prefix + '.hyd'
enData.ENopen(inpfile, rptfile, outfile)
if use_hyd:
enData.ENusehydfile(hydfile)
logger.debug('Loaded hydraulics')
else:
enData.ENsolveH()
logger.debug('Solved hydraulics')
if save_hyd:
enData.ENsavehydfile(hydfile)
logger.debug('Saved hydraulics')
enData.ENsolveQ()
logger.debug('Solved quality')
enData.ENreport()
logger.debug('Ran quality')
enData.ENclose()
logger.debug('Completed run')
#os.sys.stderr.write('Finished Closing\n')
results = self.reader.read(outfile, convergence_error, self._wn.options.hydraulic.headloss=='D-W')
return results
