• 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

36.25
/wntr/epanet/msx/toolkit.py
1
# coding: utf-8
2
"""
3
The wntr.epanet.msx.toolkit module is a Python extension for the EPANET-MSX
4
Programmers Toolkit DLLs.
5

6
.. note::
7
    
8
    Code in this section is based on code from "EPANET-MSX-Python-wrapper",
9
    licensed under the BSD license. See LICENSE.md for details.
10
"""
11
import ctypes
1✔
12
import logging
1✔
13
import os
1✔
14
import os.path
1✔
15
import platform
1✔
16
import sys
1✔
17
from typing import Union
1✔
18

19
if sys.version_info[0:2] <= (3, 11):
1✔
20
    from pkg_resources import resource_filename
1✔
21
else:
22
    from importlib.resources import files
1✔
23

24
from wntr.epanet.msx.enums import TkObjectType, TkSourceType
1✔
25

26
from ..toolkit import ENepanet
1✔
27
from .exceptions import (MSX_ERROR_CODES, EpanetMsxException, MSXKeyError,
1✔
28
                         MSXValueError)
29

30
logger = logging.getLogger(__name__)
1✔
31

32
epanet_toolkit = "wntr.epanet.toolkit"
1✔
33

34
class MSXepanet(ENepanet):
1✔
35
    def __init__(self, inpfile="", rptfile="", binfile="", msxfile=""):
1✔
36

37
        self.ENlib = None
1✔
38
        self.errcode = 0
1✔
39
        self.errcodelist = []
1✔
40
        self.cur_time = 0
1✔
41

42
        self.Warnflag = False
1✔
43
        self.Errflag = False
1✔
44
        self.fileLoaded = False
1✔
45

46
        self.inpfile = inpfile
1✔
47
        self.rptfile = rptfile
1✔
48
        self.binfile = binfile
1✔
49
        self.msxfile = msxfile
1✔
50

51
        try:
1✔
52
            if sys.version_info[0:2] <= (3, 11):
1✔
53

54
                if os.name in ["nt", "dos"]:
1✔
UNCOV
55
                    libepanet = resource_filename(__name__, "../libepanet/windows-x64/epanet2.dll")
×
UNCOV
56
                    libmsx = resource_filename(__name__, "../libepanet/windows-x64/epanetmsx.dll")
×
57
                elif sys.platform in ["darwin"]:
1✔
UNCOV
58
                    if 'arm' in platform.platform().lower():
×
UNCOV
59
                        libepanet = resource_filename(__name__, "../libepanet/darwin-arm/libepanet2.dylib")
×
UNCOV
60
                        libmsx = resource_filename(__name__, "../libepanet/darwin-arm/libepanetmsx.dylib")
×
61
                    else:
UNCOV
62
                        libepanet = resource_filename(__name__, "../libepanet/darwin-x64/libepanet2.dylib")
×
UNCOV
63
                        libmsx = resource_filename(__name__, "../libepanet/darwin-x64/libepanetmsx.dylib")
×
64
                else:
65
                    libepanet = resource_filename(__name__, "../libepanet/linux-x64/libepanet2.so")
1✔
66
                    libmsx = resource_filename(__name__, "../libepanet/linux-x64/libepanetmsx.so")
1✔
67

68
                dylib_dir = os.environ.get('DYLD_FALLBACK_LIBRARY_PATH','')
1✔
69
                if dylib_dir != '':
1✔
UNCOV
70
                    if 'arm' in platform.platform().lower():
×
UNCOV
71
                        dylib_dir = dylib_dir + ':' + resource_filename(__name__, "../libepanet/darwin-arm")
×
72
                    else:
73
                        dylib_dir = dylib_dir + ':' + resource_filename(__name__, "../libepanet/darwin-x64")
×
UNCOV
74
                    os.environ['DYLD_FALLBACK_LIBRARY_PATH'] = dylib_dir
×
75
            else:
76
                if os.name in ["nt", "dos"]:
1✔
77
                    libepanet = files(__name__).joinpath("../libepanet/windows-x64/epanet2.dll")
×
UNCOV
78
                    libmsx = files(__name__).joinpath("../libepanet/windows-x64/epanetmsx.dll")
×
79
                elif sys.platform in ["darwin"]:
1✔
UNCOV
80
                    if 'arm' in platform.platform().lower():
×
UNCOV
81
                        libepanet = files(__name__).joinpath("../libepanet/darwin-arm/libepanet2.dylib")
×
UNCOV
82
                        libmsx = files(__name__).joinpath("../libepanet/darwin-arm/libepanetmsx.dylib")
×
83
                    else:
UNCOV
84
                        libepanet = files(__name__).joinpath("../libepanet/darwin-x64/libepanet2.dylib")
×
UNCOV
85
                        libmsx = files(__name__).joinpath("../libepanet/darwin-x64/libepanetmsx.dylib")
×
86
                else:
87
                    libepanet = files(__name__).joinpath("../libepanet/linux-x64/libepanet2.so")
1✔
88
                    libmsx = files(__name__).joinpath("../libepanet/linux-x64/libepanetmsx.so")
1✔
89

90
                dylib_dir = os.environ.get('DYLD_FALLBACK_LIBRARY_PATH','')
1✔
91
                if dylib_dir != '':
1✔
92
                    if 'arm' in platform.platform().lower():
×
93
                        dylib_dir = dylib_dir + ':' + files(__name__).joinpath("../libepanet/darwin-arm")
×
94
                    else:
UNCOV
95
                        dylib_dir = dylib_dir + ':' + files(__name__).joinpath("../libepanet/darwin-x64")
×
UNCOV
96
                    os.environ['DYLD_FALLBACK_LIBRARY_PATH'] = dylib_dir
×
97
            if os.name in ["nt", "dos"]:
1✔
UNCOV
98
                self.ENlib = ctypes.windll.LoadLibrary(libmsx)
×
99
            else:
100
                self.ENlib = ctypes.cdll.LoadLibrary(libmsx)
1✔
UNCOV
101
        except:
×
UNCOV
102
            raise
×
103
        finally:
104
            self._project = None
1✔
105
        return
1✔
106

107
    def _error(self, *args):
1✔
108
        """Print the error text the corresponds to the error code returned"""
109
        if not self.errcode:
1✔
110
            return
1✔
111
        # errtxt = self.ENlib.ENgeterror(self.errcode)
UNCOV
112
        errtext =  MSX_ERROR_CODES.get(self.errcode, 'unknown error')
×
113
        if '%' in errtext and len(args) == 1:
×
UNCOV
114
            errtext % args
×
115
        if self.errcode >= 100:
×
UNCOV
116
            self.Errflag = True
×
117
            logger.error("EPANET error {} - {}".format(self.errcode, errtext))
×
UNCOV
118
            raise EpanetMsxException(self.errcode)
×
119
        return
×
120

121

122
    def ENopen(self, inpfile=None, rptfile=None, binfile=None):
1✔
123
        """
124
        Opens an EPANET input file and reads in network data
125

126
        Parameters
127
        ----------
128
        inpfile : str
129
            EPANET INP file (default to constructor value)
130
        rptfile : str
131
            Output file to create (default to constructor value)
132
        binfile : str
133
            Binary output file to create (default to constructor value)
134
        """
135
        if self.fileLoaded:
1✔
UNCOV
136
            self.ENclose()
×
137
        if self.fileLoaded:
1✔
UNCOV
138
            raise RuntimeError("File is loaded and cannot be closed")
×
139
        if inpfile is None:
1✔
UNCOV
140
            inpfile = self.inpfile
×
141
        if rptfile is None:
1✔
UNCOV
142
            rptfile = self.rptfile
×
143
        if binfile is None:
1✔
UNCOV
144
            binfile = self.binfile
×
145
        inpfile = inpfile.encode("latin-1")
1✔
146
        rptfile = rptfile.encode("latin-1")
1✔
147
        binfile = binfile.encode("latin-1")
1✔
148
        self.errcode = self.ENlib.MSXENopen(inpfile, rptfile, binfile)
1✔
149
        self._error()
1✔
150
        if self.errcode < 100:
1✔
151
            self.fileLoaded = True
1✔
152
        return
1✔
153

154
    def ENclose(self):
1✔
155
        """Frees all memory and files used by EPANET"""
156
        self.errcode = self.ENlib.MSXENclose()
1✔
157
        self._error()
1✔
158
        if self.errcode < 100:
1✔
159
            self.fileLoaded = False
1✔
160
        return
1✔
161

162
    # ----------running the simulation-----------------------------------------
163
    def MSXopen(self, msxfile):
1✔
164
        """Opens the MSX Toolkit to analyze a particular distribution system.
165

166
        Parameters
167
        ----------
168
        msxfile : str
169
            Name of the MSX input file
170
        """
171
        if msxfile is not None:
1✔
172
            msxfile = ctypes.c_char_p(msxfile.encode())
1✔
173
        ierr = self.ENlib.MSXopen(msxfile)
1✔
174
        if ierr != 0:
1✔
UNCOV
175
            raise EpanetMsxException(ierr, msxfile)
×
176

177
    def MSXclose(self):
1✔
178
        """Closes down the Toolkit system (including all files being processed)"""
179
        ierr = self.ENlib.MSXclose()
1✔
180
        if ierr != 0:
1✔
UNCOV
181
            raise EpanetMsxException(ierr)
×
182

183
    def MSXusehydfile(self, filename):
1✔
184
        """Uses the contents of the specified file as the current binary
185
        hydraulics file
186

187
        Parameters
188
        ----------
189
        filename : str
190
            Name of the hydraulics file to use
191
        """
192
        ierr = self.ENlib.MSXusehydfile(ctypes.c_char_p(filename.encode()))
1✔
193
        if ierr != 0:
1✔
UNCOV
194
            raise EpanetMsxException(ierr, filename)
×
195

196
    def MSXsolveH(self):
1✔
197
        """Runs a complete hydraulic simulation with results
198
        for all time periods written to the binary Hydraulics file."""
199
        ierr = self.ENlib.MSXsolveH()
1✔
200
        if ierr != 0:
1✔
201
            raise EpanetMsxException(ierr)
×
202

203
    def MSXinit(self, saveFlag=0):
1✔
204
        """Initializes the MSX system before solving for water quality results
205
        in step-wise fashion set saveFlag to 1 if water quality results should
206
        be saved to a scratch binary file, or to 0 is not saved to file"""
207
        saveFlag = int(saveFlag)
1✔
208
        ierr = self.ENlib.MSXinit(saveFlag)
1✔
209
        if ierr != 0:
1✔
UNCOV
210
            raise EpanetMsxException(ierr)
×
211

212
    def MSXsolveQ(self):
1✔
213
        """Solves for water quality over the entire simulation period and saves
214
        the results to an internal scratch file"""
215
        ierr = self.ENlib.MSXsolveQ()
1✔
216
        if ierr != 0:
1✔
217
            raise EpanetMsxException(ierr)
×
218

219
    def MSXstep(self):
1✔
220
        """Advances the water quality simulation one water quality time step.
221
        The time remaining in the overall simulation is returned as tleft, the
222
        current time as t."""
UNCOV
223
        t = ctypes.c_long()
×
UNCOV
224
        tleft = ctypes.c_long()
×
UNCOV
225
        ierr = self.ENlib.MSXstep(ctypes.byref(t), ctypes.byref(tleft))
×
UNCOV
226
        if ierr != 0:
×
UNCOV
227
            raise EpanetMsxException(ierr)
×
UNCOV
228
        out = [t.value, tleft.value]
×
UNCOV
229
        return out
×
230

231
    def MSXsaveoutfile(self, filename):
1✔
232
        """Saves water quality results computed for each node, link and
233
        reporting time period to a named binary file
234

235
        Parameters
236
        ----------
237
        filename : str
238
            Save a binary results file
239
        """
240
        ierr = self.ENlib.MSXsaveoutfile(ctypes.c_char_p(filename.encode()))
1✔
241
        if ierr != 0:
1✔
UNCOV
242
            raise EpanetMsxException(ierr)
×
243

244
    def MSXsavemsxfile(self, filename):
1✔
245
        """Saves the data associated with the current MSX project into a new
246
        MSX input file
247

248
        Parameters
249
        ----------
250
        filename : str
251
            Name of the MSX input file to create
252
        """
253
        ierr = self.ENlib.MSXsavemsxfile(ctypes.c_char_p(filename.encode()))
1✔
254
        if ierr != 0:
1✔
UNCOV
255
            raise EpanetMsxException(ierr, filename)
×
256

257
    def MSXreport(self):
1✔
258
        """Writes water quality simulations results as instructed by the MSX
259
        input file to a text file"""
260
        ierr = self.ENlib.MSXreport()
1✔
261
        if ierr != 0:
1✔
262
            raise EpanetMsxException(ierr)
×
263

264
    # ---------get parameters--------------------------------------------------
265
    def MSXgetindex(self, _type: Union[int, TkObjectType], name):
1✔
266
        """Gets the internal index of an MSX object given its name.
267

268
        Parameters
269
        ----------
270
        _type : int, str or ObjectType
271
            Type of object to get an index for
272
        name : str 
273
            Name of the object to get an index for
274

275
        Returns
276
        -------
277
        int
278
            Internal index
279
        
280
        Raises
281
        ------
282
        MSXKeyError
283
            If an invalid str is passed for _type
284
        MSXValueError
285
            If _type is not a valid MSX object type
286
        """
UNCOV
287
        try:
×
UNCOV
288
            _type = TkObjectType.get(_type)
×
289
        except KeyError:
×
290
            raise MSXKeyError(515, repr(_type))
×
291
        type_ind = int(_type)
×
292
        ind = ctypes.c_int()
×
293
        ierr = self.ENlib.MSXgetindex(type_ind, ctypes.c_char_p(name.encode()), ctypes.byref(ind))
×
294
        if ierr != 0:
×
295
            raise EpanetMsxException(ierr, repr(dict(_type=_type, name=name)))
×
296
        return ind.value
×
297

298
    def MSXgetIDlen(self, _type, index):
1✔
299
        """Get the number of characters in the ID name of an MSX object
300
        given its internal index number.
301

302
        Parameters
303
        ----------
304
        _type : int, str or ObjectType
305
            Type of object to get an index for
306
        index : int
307
            Index of the object to get the ID length for
308

309
        Returns
310
        -------
311
        int
312
            Length of the object ID
313
        """
UNCOV
314
        try:
×
315
            _type = TkObjectType.get(_type)
×
316
        except KeyError:
×
317
            raise MSXKeyError(515, repr(_type))
×
318
        type_ind = int(_type)
×
319
        len = ctypes.c_int()
×
320
        ierr = self.ENlib.MSXgetIDlen(type_ind, ctypes.c_int(index), ctypes.byref(len))
×
321
        if ierr != 0:
×
322
            raise EpanetMsxException(ierr, repr(dict(_type=_type, index=index)))
×
323
        return len.value
×
324

325
    def MSXgetID(self, _type, index):
1✔
326
        """Get the ID name of an object given its internal index number
327

328
        Parameters
329
        ----------
330
        _type : int, str or ObjectType
331
            Type of object to get an index for
332
        index : int
333
            Index of the object to get the ID for
334

335
        Returns
336
        -------
337
        str
338
            Object ID
339
        """
UNCOV
340
        try:
×
UNCOV
341
            _type = TkObjectType.get(_type)
×
UNCOV
342
        except KeyError:
×
UNCOV
343
            raise MSXKeyError(515, repr(_type))
×
UNCOV
344
        type_ind = int(_type)
×
UNCOV
345
        maxlen = 32
×
UNCOV
346
        id = ctypes.create_string_buffer(maxlen)
×
UNCOV
347
        ierr = self.ENlib.MSXgetID(type_ind, ctypes.c_int(index), ctypes.byref(id), ctypes.c_int(maxlen - 1))
×
UNCOV
348
        if ierr != 0:
×
UNCOV
349
            raise EpanetMsxException(ierr, repr(dict(_type=_type, index=index)))
×
350
        # the .decode() added my MF 6/3/21
UNCOV
351
        return id.value.decode()
×
352

353
    def MSXgetinitqual(self, _type, node_link_index, species_index):
1✔
354
        """Get the initial concentration of a particular chemical species
355
        assigned to a specific node or link of the pipe network
356

357
        Parameters
358
        ----------
359
        _type : str, int or ObjectType
360
            Type of object
361
        node_link_index : int
362
            Object index
363
        species_index : int
364
            Species index
365

366
        Returns
367
        -------
368
        float
369
            Initial quality value for that node or link
370

371
        Raises
372
        ------
373
        MSXKeyError
374
            Type passed in for ``_type`` is not valid
375
        MSXValueError
376
            Value for ``_type`` is not valid
377
        EpanetMsxException
378
            Any other error from the C-API
379
        """
UNCOV
380
        try:
×
UNCOV
381
            _type = TkObjectType.get(_type)
×
UNCOV
382
        except KeyError:
×
UNCOV
383
            raise MSXKeyError(515, repr(_type))
×
UNCOV
384
        if _type not in [TkObjectType.NODE, TkObjectType.LINK]:
×
UNCOV
385
            raise MSXValueError(515, repr(_type))
×
UNCOV
386
        type_ind = int(_type)
×
UNCOV
387
        iniqual = ctypes.c_double()
×
UNCOV
388
        ierr = self.ENlib.MSXgetinitqual(ctypes.c_int(type_ind), ctypes.c_int(node_link_index), ctypes.c_int(species_index), ctypes.byref(iniqual))
×
UNCOV
389
        if ierr != 0:
×
UNCOV
390
            raise EpanetMsxException(ierr, repr(dict(_type=_type, node_link_index=node_link_index, species_index=species_index)))
×
UNCOV
391
        return iniqual.value
×
392

393
    def MSXgetqual(self, _type, node_link_index, species_index):
1✔
394
        """Get a chemical species concentration at a given node or the
395
        average concentration along a link at the current simulation time step
396

397
        Parameters
398
        ----------
399
        _type : str, int or ObjectType
400
            Type of object
401
        node_link_index : int
402
            Object index
403
        species_index : int
404
            Species index
405

406
        Returns
407
        -------
408
        float
409
            Current quality value for that node or link
410

411
        Raises
412
        ------
413
        MSXKeyError
414
            Type passed in for ``_type`` is not valid
415
        MSXValueError
416
            Value for ``_type`` is not valid
417
        EpanetMsxException
418
            Any other error from the C-API
419
        """
UNCOV
420
        try:
×
UNCOV
421
            _type = TkObjectType.get(_type)
×
UNCOV
422
        except KeyError:
×
UNCOV
423
            raise MSXKeyError(515, repr(_type))
×
UNCOV
424
        if _type not in [TkObjectType.NODE, TkObjectType.LINK]:
×
UNCOV
425
            raise MSXValueError(515, repr(_type))
×
426
        type_ind = int(_type)
×
427
        qual = ctypes.c_double()
×
428
        ierr = self.ENlib.MSXgetqual(ctypes.c_int(type_ind), ctypes.c_int(node_link_index), ctypes.c_int(species_index), ctypes.byref(qual))
×
429
        if ierr != 0:
×
430
            raise EpanetMsxException(ierr, repr(dict(_type=_type, node_link_index=node_link_index, species_index=species_index)))
×
UNCOV
431
        return qual.value
×
432

433
    def MSXgetconstant(self, constant_index):
1✔
434
        """Get the value of a particular reaction constant
435

436
        Parameters
437
        ----------
438
        constant_index : int
439
            Index to the constant
440

441
        Returns
442
        -------
443
        float
444
            Value of the constant
445

446
        Raises
447
        ------
448
        EpanetMsxException
449
            Toolkit error occurred
450
        """
UNCOV
451
        const = ctypes.c_double()
×
UNCOV
452
        ierr = self.ENlib.MSXgetconstant(constant_index, ctypes.byref(const))
×
UNCOV
453
        if ierr != 0:
×
UNCOV
454
            raise EpanetMsxException(ierr, constant_index)
×
UNCOV
455
        return const.value
×
456

457
    def MSXgetparameter(self, _type, node_link_index, param_index):
1✔
458
        """Get the value of a particular reaction parameter for a given
459
        TANK or PIPE.
460

461
        Parameters
462
        ----------
463
        _type : int or str or Enum
464
            Get the type of the parameter
465
        node_link_index : int
466
            Link index
467
        param_index : int
468
            Parameter variable index
469

470
        Returns
471
        -------
472
        float
473
            Parameter value
474

475
        Raises
476
        ------
477
        MSXKeyError
478
            If there is no such _type
479
        MSXValueError
480
            If the _type is improper
481
        EpanetMsxException
482
            Any other error
483
        """
UNCOV
484
        try:
×
UNCOV
485
            _type = TkObjectType.get(_type)
×
UNCOV
486
        except KeyError:
×
UNCOV
487
            raise MSXKeyError(515, repr(_type))
×
UNCOV
488
        if _type not in [TkObjectType.NODE, TkObjectType.LINK]:
×
UNCOV
489
            raise MSXValueError(515, repr(_type))
×
UNCOV
490
        type_ind = int(_type)
×
491
        param = ctypes.c_double()
×
492
        ierr = self.ENlib.MSXgetparameter(ctypes.c_int(type_ind), ctypes.c_int(node_link_index), ctypes.c_int(param_index), ctypes.byref(param))
×
493
        if ierr != 0:
×
494
            raise EpanetMsxException(ierr, repr(dict(_type=_type, node_link_index=node_link_index, param_index=param_index)))
×
495
        return param.value
×
496

497
    def MSXgetsource(self, node_index, species_index):
1✔
498
        """Get information on any external source of a particular
499
        chemical species assigned to a specific node of the pipe network
500
        
501
        Parameters
502
        ----------
503
        node_index : int
504
            Node index
505
        species_index : int
506
            Species index
507
            
508
        Returns
509
        -------
510
        list
511
            [source type, level, and pattern] where level is the baseline 
512
            concentration (or mass flow rate) of the source and pattern the 
513
            index of the time pattern used to add variability to the source's 
514
            baseline level (0 if no pattern defined for the source)
515
        """
516
        level = ctypes.c_double()
×
517
        _type = ctypes.c_int()
×
UNCOV
518
        pat = ctypes.c_int()
×
UNCOV
519
        ierr = self.ENlib.MSXgetsource(ctypes.c_int(node_index), ctypes.c_int(species_index), ctypes.byref(_type), ctypes.byref(level), ctypes.byref(pat))
×
UNCOV
520
        if ierr != 0:
×
UNCOV
521
            raise EpanetMsxException(ierr, repr(dict(node_index=node_index, species_index=species_index)))
×
UNCOV
522
        src_out = [TkSourceType.get(_type.value), level.value, pat.value]
×
UNCOV
523
        return src_out
×
524

525
    def MSXgetpatternlen(self, pat):
1✔
526
        """Get the number of time periods within a SOURCE time pattern.
527

528
        Parameters
529
        ----------
530
        pat : int
531
            Pattern index
532

533
        Returns
534
        -------
535
        int
536
            Number of time periods in the pattern
537
        """
538
        len = ctypes.c_int()
×
UNCOV
539
        ierr = self.ENlib.MSXgetpatternlen(pat, ctypes.byref(len))
×
UNCOV
540
        if ierr != 0:
×
UNCOV
541
            raise EpanetMsxException(ierr)
×
UNCOV
542
        return len.value
×
543

544
    def MSXgetpatternvalue(self, pat, period):
1✔
545
        """Get the multiplier at a specific time period for a given
546
        SOURCE time pattern
547

548
        Parameters
549
        ----------
550
        pat : int
551
            Pattern index
552
        period : int
553
            1-indexed period of the pattern to retrieve
554

555
        Returns
556
        -------
557
            Multiplier
558
        """
559
        val = ctypes.c_double()
×
560
        ierr = self.ENlib.MSXgetpatternvalue(pat, period, ctypes.byref(val))
×
561
        if ierr != 0:
×
562
            raise EpanetMsxException(ierr)
×
563
        return val.value
×
564

565
    def MSXgetcount(self, _type):
1✔
566
        """Get the number of objects of a specified type.
567

568
        Parameters
569
        ----------
570
        _type : int or str or Enum
571
            Type of object to count
572

573
        Returns
574
        -------
575
        int
576
            Number of objects of specified type
577

578
        Raises
579
        ------
580
        MSXKeyError
581
            If the _type is invalid
582
        """
UNCOV
583
        try:
×
584
            _type = TkObjectType.get(_type)
×
585
        except KeyError:
×
586
            raise MSXKeyError(515, repr(_type))
×
587
        type_ind = int(_type)
×
588
        count = ctypes.c_int()
×
589
        ierr = self.ENlib.MSXgetcount(type_ind, ctypes.byref(count))
×
590
        if ierr != 0:
×
591
            raise EpanetMsxException(ierr)
×
592
        return count.value
×
593

594
    def MSXgetspecies(self, species_index):
1✔
595
        """Get the attributes of a chemical species given its internal
596
        index number.
597

598
        Parameters
599
        ----------
600
        species_index : int
601
            Species index to query (starting from 1 as listed in the MSX input
602
            file)
603

604
        Returns
605
        -------
606
        int, str, float, float
607
            Type, units, aTol, and rTol for the species
608
        """
UNCOV
609
        type_ind = ctypes.c_int()
×
UNCOV
610
        units = ctypes.create_string_buffer(15)
×
UNCOV
611
        aTol = ctypes.c_double()
×
UNCOV
612
        rTol = ctypes.c_double()
×
UNCOV
613
        ierr = self.ENlib.MSXgetspecies(species_index, ctypes.byref(type_ind), ctypes.byref(units), ctypes.byref(aTol), ctypes.byref(rTol))
×
UNCOV
614
        if ierr != 0:
×
615
            raise EpanetMsxException(ierr)
×
616
        spe_out = [type_ind.value, units.value, aTol.value, rTol.value]
×
617
        return spe_out
×
618

619
    def MSXgeterror(self, errcode, len=100):
1✔
620
        """Get the text for an error message given its error code
621

622
        Parameters
623
        ----------
624
        errcode : int
625
            Error code
626
        len : int, optional
627
            Length of the error message, by default 100 and minimum 80
628

629
        Returns
630
        -------
631
        str
632
            String decoded from the DLL
633

634
        Warning
635
        -------
636
        Getting string parameters in this way is not recommended, because it
637
        requires setting up string arrays that may or may not be the correct
638
        size. Use the wntr.epanet.msx.enums package to get error information.
639
        """
UNCOV
640
        errmsg = ctypes.create_string_buffer(len)
×
UNCOV
641
        self.ENlib.MSXgeterror(errcode, ctypes.byref(errmsg), len)
×
UNCOV
642
        return errmsg.value.decode()
×
643

644
    # --------------set parameters-----------------------------------
645

646
    def MSXsetconstant(self, ind, value):
1✔
647
        """Set a new value to a specific reaction constant
648

649
        Parameters
650
        ----------
651
        ind : int
652
            Index to the variable
653
        value : float
654
            Value to give the constant
655
        """
UNCOV
656
        ierr = self.ENlib.MSXsetconstant(ctypes.c_int(ind), ctypes.c_double(value))
×
657
        if ierr != 0:
×
658
            raise EpanetMsxException(ierr)
×
659

660
    def MSXsetparameter(self, _type, ind, param, value):
1✔
661
        """Set a value to a particular reaction parameter for a given TANK
662
        or PIPE
663

664
        Parameters
665
        ----------
666
        _type : int or str or enum
667
            Type of value to set
668
        ind : int
669
            Tank or pipe index
670
        param : int
671
            Parameter variable index
672
        value : float
673
            Value to be set
674

675
        Raises
676
        ------
677
        MSXKeyError
678
            If there is no such _type
679
        MSXValueError
680
            If the _type is invalid
681
        """
UNCOV
682
        try:
×
683
            _type = TkObjectType.get(_type)
×
684
        except KeyError:
×
685
            raise MSXKeyError(515, repr(_type))
×
686
        if _type not in [TkObjectType.NODE, TkObjectType.LINK]:
×
687
            raise MSXValueError(515, repr(_type))
×
688
        type_ind = int(_type)
×
689
        ierr = self.ENlib.MSXsetparameter(ctypes.c_int(type_ind), ctypes.c_int(ind), ctypes.c_int(param), ctypes.c_double(value))
×
690
        if ierr != 0:
×
691
            raise EpanetMsxException(ierr)
×
692

693
    def MSXsetinitqual(self, _type, ind, spe, value):
1✔
694
        """Set the initial concentration of a particular chemical species
695
        assigned to a specific node or link of the pipe network.
696

697
        Parameters
698
        ----------
699
        _type : int or str or enum
700
            Type of network element to set
701
        ind : int
702
            Index of the network element
703
        spe : int
704
            Index of the species
705
        value : float
706
            Initial quality value
707
        """
UNCOV
708
        try:
×
UNCOV
709
            _type = TkObjectType.get(_type)
×
UNCOV
710
        except KeyError:
×
711
            raise MSXKeyError(515, repr(_type))
×
712
        if _type not in [TkObjectType.NODE, TkObjectType.LINK]:
×
713
            raise MSXValueError(515, repr(_type))
×
714
        type_ind = int(_type)
×
715
        ierr = self.ENlib.MSXsetinitqual(ctypes.c_int(type_ind), ctypes.c_int(ind), ctypes.c_int(spe), ctypes.c_double(value))
×
716
        if ierr != 0:
×
717
            raise EpanetMsxException(ierr)
×
718

719
    def MSXsetsource(self, node, spe, _type, level, pat):
1✔
720
        """Set the attributes of an external source of a particular chemical
721
        species in a specific node of the pipe network
722

723
        Parameters
724
        ----------
725
        node : int
726
            Node index
727
        spe : int
728
            Species index
729
        _type : int or str or enum
730
            Type of source
731
        level : float
732
            Source quality value
733
        pat : int
734
            Pattern index
735
        """
736
        try:
×
737
            _type = TkSourceType.get(_type)
×
UNCOV
738
        except KeyError:
×
UNCOV
739
            raise MSXKeyError(515, repr(_type))
×
UNCOV
740
        type_ind = int(_type)
×
UNCOV
741
        ierr = self.ENlib.MSXsetsource(ctypes.c_int(node), ctypes.c_int(spe), ctypes.c_int(type_ind), ctypes.c_double(level), ctypes.c_int(pat))
×
UNCOV
742
        if ierr != 0:
×
UNCOV
743
            raise EpanetMsxException(ierr)
×
744

745
    def MSXsetpattern(self, pat, mult):
1✔
746
        """Set multipliers to a given MSX SOURCE time pattern
747

748
        Parameters
749
        ----------
750
        pat : int
751
            Pattern index
752
        mult : list-like
753
            Pattern multipliers
754
        """
UNCOV
755
        length = len(mult)
×
UNCOV
756
        cfactors_type = ctypes.c_double * length
×
UNCOV
757
        cfactors = cfactors_type()
×
UNCOV
758
        for i in range(length):
×
UNCOV
759
            cfactors[i] = float(mult[i])
×
UNCOV
760
        ierr = self.ENlib.MSXsetpattern(ctypes.c_int(pat), cfactors, ctypes.c_int(length))
×
UNCOV
761
        if ierr != 0:
×
UNCOV
762
            raise EpanetMsxException(ierr)
×
763

764
    def MSXsetpatternvalue(self, pat, period, value):
1✔
765
        """Set the multiplier factor for a specific period within a SOURCE time
766
        pattern.
767

768
        Parameters
769
        ----------
770
        pat : int
771
            Pattern index
772
        period : int
773
            1-indexed pattern time period index
774
        value : float
775
            Value to set at that time period
776
        """
UNCOV
777
        ierr = self.ENlib.MSXsetpatternvalue(ctypes.c_int(pat), ctypes.c_int(period), ctypes.c_double(value))
×
UNCOV
778
        if ierr != 0:
×
UNCOV
779
            raise EpanetMsxException(ierr)
×
780

781
    def MSXaddpattern(self, patternid):
1✔
782
        """Add a new, empty MSX source time pattern to an MSX project.
783

784
        Parameters
785
        ----------
786
        patternid : str
787
            Name of the new pattern
788
        """
UNCOV
789
        ierr = self.ENlib.MSXaddpattern(ctypes.c_char_p(patternid.encode()))
×
UNCOV
790
        if ierr != 0:
×
UNCOV
791
            raise EpanetMsxException(ierr)
×
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