• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

USEPA / WNTR / 19307136919

12 Nov 2025 06:03PM UTC coverage: 81.466% (-0.04%) from 81.505%
19307136919

push

github

web-flow
update version for rc3

1 of 1 new or added line in 1 file covered. (100.0%)

264 existing lines in 3 files now uncovered.

13107 of 16089 relevant lines covered (81.47%)

0.81 hits per line

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

70.63
/wntr/epanet/toolkit.py
1
"""
2
The wntr.epanet.toolkit module is a Python extension for the EPANET 
3
Programmers Toolkit DLLs.
4
"""
5
import ctypes
1✔
6
import logging
1✔
7
import os
1✔
8
import os.path
1✔
9
import platform
1✔
10
import sys
1✔
11
from ctypes import byref
1✔
12

13
if sys.version_info[0:2] <= (3, 11):
1✔
14
    from pkg_resources import resource_filename
1✔
15
else:
16
    from importlib.resources import files
1✔
17

18
from .exceptions import EN_ERROR_CODES, EpanetException
1✔
19
from .util import SizeLimits
1✔
20

21
logger = logging.getLogger(__name__)
1✔
22

23
epanet_toolkit = "wntr.epanet.toolkit"
1✔
24

25
if os.name in ["nt", "dos"]:
1✔
26
    libepanet = "libepanet/windows-x64/epanet22.dll"
×
27
elif sys.platform in ["darwin"]:
1✔
28
    if 'arm' in platform.platform().lower():
×
UNCOV
29
        libepanet = "libepanet/darwin-arm/libepanet2.dylib"
×
30
    else:
UNCOV
31
        libepanet = "libepanet/darwin-x64/libepanet22.dylib"
×
32
else:
33
    libepanet = "libepanet/linux-x64/libepanet22.so"
1✔
34

35

36
def ENgetwarning(code, sec=-1):
1✔
37
    if sec >= 0:
1✔
38
        hours = int(sec / 3600.0)
1✔
39
        sec -= hours * 3600
1✔
40
        mm = int(sec / 60.0)
1✔
41
        sec -= mm * 60
1✔
42
        header = "%3d:%.2d:%.2d" % (hours, mm, sec)
1✔
43
    else:
UNCOV
44
        header = "{}".format(code)
×
45
    if code < 100:
1✔
46
        msg = EN_ERROR_CODES.get(code, "Unknown warning %s")
1✔
47
    else:
UNCOV
48
        raise EpanetException(code)
×
49

50
    return msg % header
1✔
51

52

53
def runepanet(inpfile, rptfile=None, binfile=None):
1✔
54
    """Run an EPANET command-line simulation
55

56
    Parameters
57
    ----------
58
    inpfile : str
59
        The input file name
60
    """
61
    file_prefix, file_ext = os.path.splitext(inpfile)
1✔
62
    if rptfile is None:
1✔
UNCOV
63
        rptfile = file_prefix + ".rpt"
×
64
    if binfile is None:
1✔
UNCOV
65
        binfile = file_prefix + ".bin"
×
66

67
    enData = ENepanet()
1✔
68
    enData.ENopen(inpfile, rptfile, binfile)
1✔
69
    enData.ENsolveH()
1✔
70
    enData.ENsolveQ()
1✔
71
    try:
1✔
72
        enData.ENreport()
1✔
UNCOV
73
    except:
×
UNCOV
74
        pass
×
75
    enData.ENclose()
1✔
76

77

78
class ENepanet:
1✔
79
    """Wrapper class to load the EPANET DLL object, then perform operations on
80
    the EPANET object that is created when a file is loaded.
81

82
    This simulator is thread safe **only** for EPANET `version=2.2`.
83

84
    Parameters
85
    ----------
86
    inpfile : str
87
        Input file to use
88
    rptfile : str
89
        Output file to report to
90
    binfile : str
91
        Results file to generate
92
    version : float
93
        EPANET version to use (either 2.0 or 2.2)
94
    """
95

96
    def __init__(self, inpfile="", rptfile="", binfile="", version=2.2):
1✔
97
        self.ENlib = None
1✔
98
        self.errcode = 0
1✔
99
        self.errcodelist = []
1✔
100
        self.cur_time = 0
1✔
101

102
        self.Warnflag = False
1✔
103
        self.Errflag = False
1✔
104
        self.fileLoaded = False
1✔
105

106
        self.inpfile = inpfile
1✔
107
        self.rptfile = rptfile
1✔
108
        self.binfile = binfile
1✔
109

110
        try:
1✔
111
            if sys.version_info[0:2] <= (3, 11):
1✔
112
                if float(version) == 2.0:
1✔
113
                    libname = libepanet.replace('epanet22.','epanet20.')
1✔
114
                    if 'arm' in platform.platform():
1✔
UNCOV
115
                        raise NotImplementedError('ARM-based processors not supported for version 2.0 of EPANET. Please use version=2.2')
×
116
                else:
117
                    libname = libepanet
1✔
118
                libname = resource_filename(__name__, libname)
1✔
119
                if os.name in ["nt", "dos"]:
1✔
120
                    self.ENlib = ctypes.windll.LoadLibrary(libname)
×
121
                else:
122
                    self.ENlib = ctypes.cdll.LoadLibrary(libname)
1✔
123
            else:
124
                if float(version) == 2.0:
1✔
125
                    libname = libepanet.replace('epanet22.','epanet20.')
1✔
126
                    if 'arm' in platform.platform():
1✔
UNCOV
127
                        raise NotImplementedError('ARM-based processors not supported for version 2.0 of EPANET. Please use version=2.2')
×
128
                else:
129
                    libname = libepanet
1✔
130
                libname = files(__name__).joinpath(libname)
1✔
131
                if os.name in ["nt", "dos"]:
1✔
UNCOV
132
                    self.ENlib = ctypes.windll.LoadLibrary(libname)
×
133
                else:
134
                    self.ENlib = ctypes.cdll.LoadLibrary(libname)
1✔
UNCOV
135
        except:
×
UNCOV
136
            raise
×
137
        finally:
138
            if version >= 2.2:
1✔
139
                self._project = ctypes.c_uint64()
1✔
140
            else:
141
                self._project = None
1✔
142
        return
1✔
143

144
    def isOpen(self):
1✔
145
        """Checks to see if the file is open"""
146
        return self.fileLoaded
1✔
147

148
    def _error(self, *args):
1✔
149
        """Print the error text the corresponds to the error code returned"""
150
        if not self.errcode:
1✔
151
            return
1✔
152
        # errtxt = self.ENlib.ENgeterror(self.errcode)
153
        errtext = EN_ERROR_CODES.get(self.errcode, "unknown error")
1✔
154
        if "%" in errtext and len(args) == 1:
1✔
UNCOV
155
            errtext % args
×
156
        if self.errcode >= 100:
1✔
UNCOV
157
            self.Errflag = True
×
UNCOV
158
            logger.error("EPANET error {} - {}".format(self.errcode, errtext))
×
UNCOV
159
            raise EpanetException(self.errcode)
×
160
        else:
161
            self.Warnflag = True
1✔
162
            # warnings.warn(ENgetwarning(self.errcode))
163
            logger.warning("EPANET warning {} - {}".format(self.errcode, ENgetwarning(self.errcode, self.cur_time)))
1✔
164
            self.errcodelist.append(ENgetwarning(self.errcode, self.cur_time))
1✔
165
        return
1✔
166

167
    def ENopen(self, inpfile=None, rptfile=None, binfile=None):
1✔
168
        """
169
        Opens an EPANET input file and reads in network data
170

171
        Parameters
172
        ----------
173
        inpfile : str
174
            EPANET INP file (default to constructor value)
175
        rptfile : str
176
            Output file to create (default to constructor value)
177
        binfile : str
178
            Binary output file to create (default to constructor value)
179
        """
180
        if self._project is not None:
1✔
181
            if self.fileLoaded:
1✔
UNCOV
182
                self.EN_close(self._project)
×
183
            if self.fileLoaded:
1✔
UNCOV
184
                raise RuntimeError("File is loaded and cannot be closed")
×
185
            if inpfile is None:
1✔
186
                inpfile = self.inpfile
×
187
            if rptfile is None:
1✔
188
                rptfile = self.rptfile
×
189
            if binfile is None:
1✔
190
                binfile = self.binfile
1✔
191
            inpfile = inpfile.encode("latin-1")
1✔
192
            rptfile = rptfile.encode("latin-1")
1✔
193
            binfile = binfile.encode("latin-1")
1✔
194
            self.ENlib.EN_createproject(ctypes.byref(self._project))
1✔
195
            self.errcode = self.ENlib.EN_open(self._project, inpfile, rptfile, binfile)
1✔
196
            self._error()
1✔
197
            if self.errcode < 100:
1✔
198
                self.fileLoaded = True
1✔
199
            return
1✔
200
        else:
201
            if self.fileLoaded:
1✔
UNCOV
202
                self.ENclose()
×
203
            if self.fileLoaded:
1✔
UNCOV
204
                raise RuntimeError("File is loaded and cannot be closed")
×
205
            if inpfile is None:
1✔
UNCOV
206
                inpfile = self.inpfile
×
207
            if rptfile is None:
1✔
UNCOV
208
                rptfile = self.rptfile
×
209
            if binfile is None:
1✔
210
                binfile = self.binfile
1✔
211
            inpfile = inpfile.encode("latin-1")
1✔
212
            rptfile = rptfile.encode("latin-1")
1✔
213
            binfile = binfile.encode("latin-1")
1✔
214
            self.errcode = self.ENlib.ENopen(inpfile, rptfile, binfile)
1✔
215
            self._error()
1✔
216
            if self.errcode < 100:
1✔
217
                self.fileLoaded = True
1✔
218
            return
1✔
219

220
    def ENclose(self):
1✔
221
        """Frees all memory and files used by EPANET"""
222
        if self._project is not None:
1✔
223
            self.errcode = self.ENlib.EN_close(self._project)
1✔
224
            self.ENlib.EN_deleteproject(self._project)
1✔
225
            self._project = None
1✔
226
            self._project = ctypes.c_uint64()
1✔
227
        else:
228
            self.errcode = self.ENlib.ENclose()
1✔
229
        self._error()
1✔
230
        if self.errcode < 100:
1✔
231
            self.fileLoaded = False
1✔
232
        return
1✔
233

234
    def ENsolveH(self):
1✔
235
        """Solves for network hydraulics in all time periods"""
236
        if self._project is not None:
1✔
237
            self.errcode = self.ENlib.EN_solveH(self._project)
1✔
238
        else:
239
            self.errcode = self.ENlib.ENsolveH()
1✔
240
        self._error()
1✔
241
        return
1✔
242

243
    def ENsaveH(self):
1✔
244
        """Solves for network hydraulics in all time periods
245

246
        Must be called before ENreport() if no water quality simulation made.
247
        Should not be called if ENsolveQ() will be used.
248
        """
UNCOV
249
        if self._project is not None:
×
UNCOV
250
            self.errcode = self.ENlib.EN_saveH(self._project)
×
251
        else:
UNCOV
252
            self.errcode = self.ENlib.ENsaveH()
×
UNCOV
253
        self._error()
×
UNCOV
254
        return
×
255

256
    def ENopenH(self):
1✔
257
        """Sets up data structures for hydraulic analysis"""
258
        if self._project is not None:
1✔
259
            self.errcode = self.ENlib.EN_openH(self._project)
1✔
260
        else:
261
            self.errcode = self.ENlib.ENopenH()
1✔
262
        self._error()
1✔
263
        return
1✔
264

265
    def ENinitH(self, iFlag):
1✔
266
        """Initializes hydraulic analysis
267

268
        Parameters
269
        -----------
270
        iFlag : 2-digit flag
271
            2-digit flag where 1st (left) digit indicates
272
            if link flows should be re-initialized (1) or
273
            not (0) and 2nd digit indicates if hydraulic
274
            results should be saved to file (1) or not (0)
275
        """
276
        if self._project is not None:
1✔
277
            self.errcode = self.ENlib.EN_initH(self._project, iFlag)
1✔
278
        else:
279
            self.errcode = self.ENlib.ENinitH(iFlag)
1✔
280
        self._error()
1✔
281
        return
1✔
282

283
    def ENrunH(self):
1✔
284
        """Solves hydraulics for conditions at time t
285

286
        This function is used in a loop with ENnextH() to run
287
        an extended period hydraulic simulation.
288
        See ENsolveH() for an example.
289

290
        Returns
291
        --------
292
        int
293
            Current simulation time (seconds)
294
        """
295
        lT = ctypes.c_long()
1✔
296
        if self._project is not None:
1✔
297
            self.errcode = self.ENlib.EN_runH(self._project, byref(lT))
1✔
298
        else:
299
            self.errcode = self.ENlib.ENrunH(byref(lT))
1✔
300
        self._error()
1✔
301
        self.cur_time = lT.value
1✔
302
        return lT.value
1✔
303

304
    def ENnextH(self):
1✔
305
        """Determines time until next hydraulic event
306

307
        This function is used in a loop with ENrunH() to run
308
        an extended period hydraulic simulation.
309
        See ENsolveH() for an example.
310

311
        Returns
312
        ---------
313
        int
314
            Time (seconds) until next hydraulic event (0 marks end of simulation period)
315
        """
316
        lTstep = ctypes.c_long()
1✔
317
        if self._project is not None:
1✔
318
            self.errcode = self.ENlib.EN_nextH(self._project, byref(lTstep))
1✔
319
        else:
320
            self.errcode = self.ENlib.ENnextH(byref(lTstep))
1✔
321
        self._error()
1✔
322
        return lTstep.value
1✔
323

324
    def ENcloseH(self):
1✔
325
        """Frees data allocated by hydraulics solver"""
326
        if self._project is not None:
1✔
327
            self.errcode = self.ENlib.EN_closeH(self._project)
1✔
328
        else:
329
            self.errcode = self.ENlib.ENcloseH()
1✔
330
        self._error()
1✔
331
        return
1✔
332

333
    def ENsavehydfile(self, filename):
1✔
334
        """Copies binary hydraulics file to disk
335

336
        Parameters
337
        -------------
338
        filename : str
339
            Name of hydraulics file to output
340
        """
341
        if self._project is not None:
1✔
342
            self.errcode = self.ENlib.EN_savehydfile(self._project, filename.encode("latin-1"))
1✔
343
        else:
344
            self.errcode = self.ENlib.ENsavehydfile(filename.encode("latin-1"))
×
345
        self._error()
1✔
346
        return
1✔
347

348
    def ENusehydfile(self, filename):
1✔
349
        """Opens previously saved binary hydraulics file
350

351
        Parameters
352
        -------------
353
        filename : str
354
            Name of hydraulics file to use
355
        """
UNCOV
356
        if self._project is not None:
×
UNCOV
357
            self.errcode = self.ENlib.EN_usehydfile(self._project, filename.encode("latin-1"))
×
358
        else:
359
            self.errcode = self.ENlib.ENusehydfile(filename.encode("latin-1"))
×
UNCOV
360
        self._error()
×
361
        return
×
362

363
    def ENsolveQ(self):
1✔
364
        """Solves for network water quality in all time periods"""
365
        if self._project is not None:
1✔
366
            self.errcode = self.ENlib.EN_solveQ(self._project)
1✔
367
        else:
368
            self.errcode = self.ENlib.ENsolveQ()
1✔
369
        self._error()
1✔
370
        return
1✔
371

372
    def ENopenQ(self):
1✔
373
        """Sets up data structures for water quality analysis"""
374
        if self._project is not None:
×
UNCOV
375
            self.errcode = self.ENlib.EN_openQ(self._project)
×
376
        else:
377
            self.errcode = self.ENlib.ENopenQ()
×
378
        self._error()
×
UNCOV
379
        return
×
380

381
    def ENinitQ(self, iSaveflag):
1✔
382
        """Initializes water quality analysis
383

384
        Parameters
385
        -------------
386
        iSaveflag : int
387
             EN_SAVE (1) if results saved to file, EN_NOSAVE (0) if not
388
        """
UNCOV
389
        if self._project is not None:
×
UNCOV
390
            self.errcode = self.ENlib.EN_initQ(self._project, iSaveflag)
×
391
        else:
392
            self.errcode = self.ENlib.ENinitQ(iSaveflag)
×
393
        self._error()
×
394
        return
×
395

396
    def ENrunQ(self):
1✔
397
        """Retrieves hydraulic and water quality results at time t
398

399
        This function is used in a loop with ENnextQ() to run
400
        an extended period water quality simulation. See ENsolveQ() for
401
        an example.
402

403
        Returns
404
        -------
405
        int
406
            Current simulation time (seconds)
407
        """
UNCOV
408
        lT = ctypes.c_long()
×
UNCOV
409
        if self._project is not None:
×
UNCOV
410
            self.errcode = self.ENlib.EN_runQ(self._project, byref(lT))
×
411
        else:
412
            self.errcode = self.ENlib.ENrunQ(byref(lT))
×
413
        self._error()
×
414
        return lT.value
×
415

416
    def ENnextQ(self):
1✔
417
        """Advances water quality simulation to next hydraulic event
418

419
        This function is used in a loop with ENrunQ() to run
420
        an extended period water quality simulation. See ENsolveQ() for
421
        an example.
422

423
        Returns
424
        --------
425
        int
426
            Time (seconds) until next hydraulic event (0 marks end of simulation period)
427
        """
UNCOV
428
        lTstep = ctypes.c_long()
×
UNCOV
429
        if self._project is not None:
×
UNCOV
430
            self.errcode = self.ENlib.EN_nextQ(self._project, byref(lTstep))
×
431
        else:
UNCOV
432
            self.errcode = self.ENlib.ENnextQ(byref(lTstep))
×
UNCOV
433
        self._error()
×
UNCOV
434
        return lTstep.value
×
435

436
    def ENcloseQ(self):
1✔
437
        """Frees data allocated by water quality solver"""
UNCOV
438
        if self._project is not None:
×
UNCOV
439
            self.errcode = self.ENlib.EN_closeQ(self._project)
×
440
        else:
UNCOV
441
            self.errcode = self.ENlib.ENcloseQ()
×
UNCOV
442
        self._error()
×
UNCOV
443
        return
×
444

445
    def ENreport(self):
1✔
446
        """Writes report to report file"""
447
        if self._project is not None:
1✔
448
            self.errcode = self.ENlib.EN_report(self._project)
1✔
449
        else:
450
            self.errcode = self.ENlib.ENreport()
1✔
451
        self._error()
1✔
452
        return
1✔
453

454
    def ENgetcount(self, iCode):
1✔
455
        """Retrieves the number of components of a given type in the network
456

457
        Parameters
458
        -------------
459
        iCode : int
460
            Component code (see toolkit.optComponentCounts)
461

462
        Returns
463
        ---------
464
        int
465
            Number of components in network
466

467
        """
468
        iCount = ctypes.c_int()
1✔
469
        if self._project is not None:
1✔
470
            self.errcode = self.ENlib.EN_getcount(self._project, iCode, byref(iCount))
1✔
471
        else:
472
            self.errcode = self.ENlib.ENgetcount(iCode, byref(iCount))
1✔
473
        self._error()
1✔
474
        return iCount.value
1✔
475

476
    def ENgetflowunits(self):
1✔
477
        """Retrieves flow units code
478

479
        Returns
480
        -----------
481
        Code of flow units in use (see toolkit.optFlowUnits)
482

483
        """
484
        iCode = ctypes.c_int()
1✔
485
        if self._project is not None:
1✔
486
            self.errcode = self.ENlib.EN_getflowunits(self._project, byref(iCode))
1✔
487
        else:
488
            self.errcode = self.ENlib.ENgetflowunits(byref(iCode))
1✔
489
        self._error()
1✔
490
        return iCode.value
1✔
491

492
    def ENgetnodeid(self, iIndex):
1✔
493
        """Gets the ID name of a node given its index.
494

495
        Parameters
496
        ----------
497
        iIndex : int
498
            a node's index (starting from 1).
499

500
        Returns
501
        -------
502
        str
503
            the node name
504
        """
UNCOV
505
        fValue = ctypes.create_string_buffer(SizeLimits.EN_MAX_ID.value)
×
UNCOV
506
        if self._project is not None:
×
UNCOV
507
            self.errcode = self.ENlib.EN_getnodeid(self._project, iIndex, byref(fValue))
×
508
        else:
UNCOV
509
            self.errcode = self.ENlib.ENgetnodeid(iIndex, byref(fValue))
×
UNCOV
510
        self._error()
×
UNCOV
511
        return str(fValue.value, "UTF-8")
×
512

513
    def ENgetnodeindex(self, sId):
1✔
514
        """Retrieves index of a node with specific ID
515

516
        Parameters
517
        -------------
518
        sId : int
519
            Node ID
520

521
        Returns
522
        ---------
523
        Index of node in list of nodes
524

525
        """
526
        iIndex = ctypes.c_int()
1✔
527
        if self._project is not None:
1✔
528
            self.errcode = self.ENlib.EN_getnodeindex(self._project, sId.encode("latin-1"), byref(iIndex))
1✔
529
        else:
530
            self.errcode = self.ENlib.ENgetnodeindex(sId.encode("latin-1"), byref(iIndex))
1✔
531
        self._error()
1✔
532
        return iIndex.value
1✔
533

534
    def ENgetnodetype(self, iIndex):
1✔
535
        """Retrieves a node's type given its index.
536

537
        Parameters
538
        ----------
539
        iIndex : int
540
            The index of the node
541

542
        Returns
543
        -------
544
        int
545
            the node type as an integer
546
        """
UNCOV
547
        fValue = ctypes.c_int()
×
UNCOV
548
        if self._project is not None:
×
UNCOV
549
            self.errcode = self.ENlib.EN_getnodetype(self._project, iIndex, byref(fValue))
×
550
        else:
UNCOV
551
            self.errcode = self.ENlib.ENgetnodetype(iIndex, byref(fValue))
×
UNCOV
552
        self._error()
×
UNCOV
553
        return fValue.value
×
554

555
    def ENgetnodevalue(self, iIndex, iCode):
1✔
556
        """Retrieves parameter value for a node
557

558
        Parameters
559
        -------------
560
        iIndex: int
561
            Node index
562
        iCode : int
563
            Node parameter code (see toolkit.optNodeParams)
564

565
        Returns
566
        ---------
567
        Value of node's parameter
568

569
        """
570
        fValue = ctypes.c_float()
1✔
571
        if self._project is not None:
1✔
572
            fValue = ctypes.c_double()
1✔
573
            self.errcode = self.ENlib.EN_getnodevalue(self._project, iIndex, iCode, byref(fValue))
1✔
574
        else:
575
            self.errcode = self.ENlib.ENgetnodevalue(iIndex, iCode, byref(fValue))
1✔
576
        self._error()
1✔
577
        return fValue.value
1✔
578

579
    def ENgetlinktype(self, iIndex):
1✔
580
        """Retrieves a link's type given its index.
581

582
        Parameters
583
        ----------
584
        iIndex : int
585
            The index of the link
586

587
        Returns
588
        -------
589
        int
590
            the link type as an integer
591
        """
UNCOV
592
        fValue = ctypes.c_int()
×
UNCOV
593
        if self._project is not None:
×
UNCOV
594
            self.errcode = self.ENlib.EN_getlinktype(self._project, iIndex, byref(fValue))
×
595
        else:
UNCOV
596
            self.errcode = self.ENlib.EN_getlinktype(iIndex, byref(fValue))
×
UNCOV
597
        self._error()
×
UNCOV
598
        return fValue.value
×
599

600
    def ENgetlinkindex(self, sId):
1✔
601
        """Retrieves index of a link with specific ID
602

603
        Parameters
604
        -------------
605
        sId : int
606
            Link ID
607

608
        Returns
609
        ---------
610
        Index of link in list of links
611

612
        """
613
        iIndex = ctypes.c_int()
1✔
614
        if self._project is not None:
1✔
615
            self.errcode = self.ENlib.EN_getlinkindex(self._project, sId.encode("latin-1"), byref(iIndex))
1✔
616
        else:
617
            self.errcode = self.ENlib.ENgetlinkindex(sId.encode("latin-1"), byref(iIndex))
1✔
618
        self._error()
1✔
619
        return iIndex.value
1✔
620

621
    def ENgetlinkvalue(self, iIndex, iCode):
1✔
622
        """Retrieves parameter value for a link
623

624
        Parameters
625
        -------------
626
        iIndex : int
627
            Link index
628
        iCode : int
629
            Link parameter code (see toolkit.optLinkParams)
630

631
        Returns
632
        ---------
633
        Value of link's parameter
634

635
        """
636
        fValue = ctypes.c_float()
1✔
637
        if self._project is not None:
1✔
638
            fValue = ctypes.c_double()
1✔
639
            self.errcode = self.ENlib.EN_getlinkvalue(self._project, iIndex, iCode, byref(fValue))
1✔
640
        else:
641
            self.errcode = self.ENlib.ENgetlinkvalue(iIndex, iCode, byref(fValue))
1✔
642
        self._error()
1✔
643
        return fValue.value
1✔
644

645
    def ENsetlinkvalue(self, iIndex, iCode, fValue):
1✔
646
        """Set the value on a link
647

648
        Parameters
649
        ----------
650
        iIndex : int
651
            the link index
652
        iCode : int
653
            the parameter enum integer
654
        fValue : float
655
            the value to set on the link
656
        """
657
        if self._project is not None:
1✔
658
            self.errcode = self.ENlib.EN_setlinkvalue(
1✔
659
                self._project, ctypes.c_int(iIndex), ctypes.c_int(iCode), ctypes.c_double(fValue)
660
            )
661
        else:
662
            self.errcode = self.ENlib.ENsetlinkvalue(ctypes.c_int(iIndex), ctypes.c_int(iCode), ctypes.c_float(fValue))
1✔
663
        self._error()
1✔
664

665
    def ENsetnodevalue(self, iIndex, iCode, fValue):
1✔
666
        """
667
        Set the value on a node
668

669
        Parameters
670
        ----------
671
        iIndex : int
672
            the node index
673
        iCode : int
674
            the parameter enum integer
675
        fValue : float
676
            the value to set on the node
677
        """
678
        if self._project is not None:
1✔
679
            self.errcode = self.ENlib.EN_setnodevalue(
1✔
680
                self._project, ctypes.c_int(iIndex), ctypes.c_int(iCode), ctypes.c_double(fValue)
681
            )
682
        else:
683
            self.errcode = self.ENlib.ENsetnodevalue(ctypes.c_int(iIndex), ctypes.c_int(iCode), ctypes.c_float(fValue))
1✔
684
        self._error()
1✔
685

686
    def ENsettimeparam(self, eParam, lValue):
1✔
687
        """Set a time parameter value
688

689
        Parameters
690
        ----------
691
        eParam : int
692
            the time parameter to set
693
        lValue : long
694
            the value to set, in seconds
695
        """
696
        if self._project is not None:
1✔
697
            self.errcode = self.ENlib.EN_settimeparam(self._project, ctypes.c_int(eParam), ctypes.c_long(lValue))
1✔
698
        else:
699
            self.errcode = self.ENlib.ENsettimeparam(ctypes.c_int(eParam), ctypes.c_long(lValue))
1✔
700
        self._error()
1✔
701

702
    def ENgettimeparam(self, eParam):
1✔
703
        """Get a time parameter value
704

705
        Parameters
706
        ----------
707
        eParam : int
708
            the time parameter to get
709

710
        Returns
711
        -------
712
        long
713
            the value of the time parameter, in seconds
714
        """
715
        lValue = ctypes.c_long()
1✔
716
        if self._project is not None:
1✔
717
            self.errcode = self.ENlib.EN_gettimeparam(self._project, ctypes.c_int(eParam), byref(lValue))
1✔
718
        else:
719
            self.errcode = self.ENlib.ENgettimeparam(ctypes.c_int(eParam), byref(lValue))
1✔
720
        self._error()
1✔
721
        return lValue.value
1✔
722

723
    def ENaddcontrol(self, iType: int, iLinkIndex: int, dSetting: float, iNodeIndex: int, dLevel: float) -> int:
1✔
724
        """Add a new simple control
725

726
        Parameters
727
        ----------
728
        iType : int
729
            the type of control
730
        iLinkIndex : int
731
            the index of the link
732
        dSetting : float
733
            the new link setting value
734
        iNodeIndex : int
735
            Set to 0 for time of day or timer
736
        dLevel : float
737
            the level to compare against
738

739
        Returns
740
        -------
741
        int
742
            the new control number
743
        """
UNCOV
744
        lValue = ctypes.c_int()
×
UNCOV
745
        if self._project is not None:
×
UNCOV
746
            self.errcode = self.ENlib.EN_addcontrol(
×
747
                self._project,
748
                ctypes.c_int(iType),
749
                ctypes.c_int(iLinkIndex),
750
                ctypes.c_double(dSetting),
751
                ctypes.c_int(iNodeIndex),
752
                ctypes.c_double(dLevel),
753
                byref(lValue),
754
            )
755
        else:
UNCOV
756
            self.errcode = self.ENlib.ENaddcontrol(
×
757
                ctypes.c_int(iType),
758
                ctypes.c_int(iLinkIndex),
759
                ctypes.c_double(dSetting),
760
                ctypes.c_int(iNodeIndex),
761
                ctypes.c_double(dLevel),
762
                byref(lValue),
763
            )
764
        self._error()
×
765
        return lValue.value
×
766

767
    def ENgetcontrol(self, iIndex: int):
1✔
768
        """Get values defined by a control.
769

770
        Parameters
771
        ----------
772
        iIndex : int
773
            the control number
774
        """
775
        iType = ctypes.c_int()
×
UNCOV
776
        iLinkIndex = ctypes.c_int()
×
UNCOV
777
        dSetting = ctypes.c_double()
×
778
        iNodeIndex = ctypes.c_int()
×
779
        dLevel = ctypes.c_double()
×
UNCOV
780
        if self._project is not None:
×
UNCOV
781
            self.errcode = self.ENlib.EN_getcontrol(
×
782
                self._project,
783
                ctypes.c_int(iIndex),
784
                byref(iType),
785
                byref(iLinkIndex),
786
                byref(dSetting),
787
                byref(iNodeIndex),
788
                byref(dLevel),
789
            )
790
        else:
UNCOV
791
            self.errcode = self.ENlib.ENgetcontrol(
×
792
                ctypes.c_int(iIndex), byref(iType), byref(iLinkIndex), byref(dSetting), byref(iNodeIndex), byref(dLevel)
793
            )
UNCOV
794
        self._error()
×
UNCOV
795
        return dict(
×
796
            index=iIndex,
797
            type=iType.value,
798
            linkindex=iLinkIndex.value,
799
            setting=dSetting.value,
800
            nodeindex=iNodeIndex.value,
801
            level=dLevel.value,
802
        )
803

804
    def ENsetcontrol(self, iIndex: int, iType: int, iLinkIndex: int, dSetting: float, iNodeIndex: int, dLevel: float):
1✔
805
        """Change values on a simple control
806

807
        Parameters
808
        ----------
809
        iIndex : int
810
            the control index
811
        iType : int
812
            the type of control comparison
813
        iLinkIndex : int
814
            the link being changed
815
        dSetting : float
816
            the setting to change to
817
        iNodeIndex : int
818
            the node being compared against, Set to 0 for time of day or timer
819
        dLevel : float
820
            the level being checked
821

822
        Warning
823
        -------
824
        There is an error in EPANET 2.2 that sets the `dLevel` parameter to 0.0 on Macs
825
        regardless of the value the user passes in. This means that to use this toolkit
826
        functionality on a Mac, the user must delete and create a new control to change
827
        the level.
828

829
        """
UNCOV
830
        if self._project is not None:
×
UNCOV
831
            try:
×
UNCOV
832
                self.errcode = self.ENlib.EN_setcontrol(
×
833
                    self._project,
834
                    ctypes.c_int(iIndex),
835
                    ctypes.c_int(iType),
836
                    ctypes.c_int(iLinkIndex),
837
                    ctypes.c_double(dSetting),
838
                    ctypes.c_int(iNodeIndex),
839
                    ctypes.c_double(dLevel),
840
                )
UNCOV
841
            except:
×
UNCOV
842
                self.errcode = self.ENlib.EN_setcontrol(
×
843
                    self._project,
844
                    ctypes.c_int(iIndex),
845
                    ctypes.c_int(iType),
846
                    ctypes.c_int(iLinkIndex),
847
                    ctypes.c_double(dSetting),
848
                    ctypes.c_int(iNodeIndex),
849
                    ctypes.c_float(dLevel),
850
                )
851
        else:
UNCOV
852
            self.errcode = self.ENlib.ENsetcontrol(
×
853
                ctypes.c_int(iIndex),
854
                ctypes.c_int(iType),
855
                ctypes.c_int(iLinkIndex),
856
                ctypes.c_double(dSetting),
857
                ctypes.c_int(iNodeIndex),
858
                ctypes.c_double(dLevel),
859
            )
UNCOV
860
        self._error()
×
861

862
    def ENdeletecontrol(self, iControlIndex):
1✔
863
        """Delete a control.
864

865
        Parameters
866
        ----------
867
        iControlIndex : int
868
            the simple control to delete
869
        """
UNCOV
870
        if self._project is not None:
×
UNCOV
871
            self.errcode = self.ENlib.EN_deletecontrol(self._project, ctypes.c_int(iControlIndex))
×
872
        else:
UNCOV
873
            self.errcode = self.ENlib.ENdeletecontrol(ctypes.c_int(iControlIndex))
×
UNCOV
874
        self._error()
×
875

876
    def ENsaveinpfile(self, inpfile):
1✔
877
        """Saves EPANET input file
878

879
        Parameters
880
        -------------
881
        inpfile : str
882
                    EPANET INP output file
883

884
        """
885

886
        inpfile = inpfile.encode("latin-1")
1✔
887
        if self._project is not None:
1✔
888
            self.errcode = self.ENlib.EN_saveinpfile(self._project, inpfile)
1✔
889
        else:
890
            self.errcode = self.ENlib.ENsaveinpfile(inpfile)
1✔
891
        self._error()
1✔
892

893
        return
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc