Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'pdd_poly1_coeffs_a'):
m.pdd_poly1_coeffs_a = aml.ParamDict()
m.pdd_poly1_coeffs_b = aml.ParamDict()
m.pdd_poly1_coeffs_c = aml.ParamDict()
m.pdd_poly1_coeffs_d = aml.ParamDict()
m.pdd_poly2_coeffs_a = aml.ParamDict()
m.pdd_poly2_coeffs_b = aml.ParamDict()
m.pdd_poly2_coeffs_c = aml.ParamDict()
m.pdd_poly2_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list
for node_name in index_over:
node = wn.get_node(node_name)
pmin = node.minimum_pressure
pnom = node.nominal_pressure
x1 = pmin
f1 = 0.0
x2 = pmin + m.pdd_smoothing_delta
f2 = ((x2 - pmin)/(pnom-pmin))**0.5
df1 = m.pdd_slope
df2 = 0.5*((x2-pmin)/(pnom-pmin))**(-0.5)*1.0/(pnom-pmin)
a1, b1, c1, d1 = cubic_spline(x1, x2, f1, f2, df1, df2)
x1 = pnom-m.pdd_smoothing_delta
def build(cls, m, wn, updater, index_over=None):
"""
Add a tcv resistance parameter to the model
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of tcv names
"""
if not hasattr(m, 'tcv_resistance'):
m.tcv_resistance = aml.ParamDict()
if index_over is None:
index_over = wn.tcv_name_list
for link_name in index_over:
link = wn.get_link(link_name)
value = 8.0 * link.setting / (9.81 * math.pi**2 * link.diameter**4)
if link_name in m.tcv_resistance:
m.tcv_resistance[link_name].value = value
else:
m.tcv_resistance[link_name] = aml.Param(value)
updater.add(link, 'setting', tcv_resistance_param.update)
updater.add(link, 'diameter', tcv_resistance_param.update)
def build(cls, m, wn, updater, index_over=None):
"""
Add parameters to the model for leak smoothing polynomial coefficients
Parameters
----------
m: wntr.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'leak_poly_coeffs_a'):
m.leak_poly_coeffs_a = aml.ParamDict()
m.leak_poly_coeffs_b = aml.ParamDict()
m.leak_poly_coeffs_c = aml.ParamDict()
m.leak_poly_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list + wn.tank_name_list
for node_name in index_over:
node = wn.get_node(node_name)
x1 = 0.0
f1 = 0.0
x2 = x1 + m.leak_delta
f2 = node.leak_discharge_coeff*node.leak_area*(2.0*9.81*x2)**0.5
df1 = m.leak_slope
df2 = 0.5*node.leak_discharge_coeff*node.leak_area*(2.0*9.81)**0.5*x2**(-0.5)
a, b, c, d = cubic_spline(x1, x2, f1, f2, df1, df2)
if node_name in m.leak_poly_coeffs_a:
def source_head_param(m, wn):
"""
Add a head param to the model
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
"""
if not hasattr(m, 'source_head'):
m.source_head = aml.ParamDict()
for node_name, node in wn.tanks():
m.source_head[node_name] = aml.Param(node.head)
for node_name, node in wn.reservoirs():
m.source_head[node_name] = aml.Param(node.head_timeseries.at(wn.sim_time))
else:
for node_name, node in wn.tanks():
m.source_head[node_name].value = node.head
for node_name, node in wn.reservoirs():
m.source_head[node_name].value = node.head_timeseries.at(wn.sim_time)
def build(cls, m, wn, updater, index_over=None):
"""
Add a nominal pressure parameter to the model
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'pnom'):
m.pnom = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list
for node_name in index_over:
node = wn.get_node(node_name)
if node_name in m.pnom:
m.pnom[node_name].value = node.nominal_pressure
else:
m.pnom[node_name] = aml.Param(node.nominal_pressure)
updater.add(node, 'nominal_pressure', pnom_param.update)
def build(cls, m, wn, updater, index_over=None):
"""
Add parameters to the model for pdd smoothing polynomial coefficients
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'pdd_poly1_coeffs_a'):
m.pdd_poly1_coeffs_a = aml.ParamDict()
m.pdd_poly1_coeffs_b = aml.ParamDict()
m.pdd_poly1_coeffs_c = aml.ParamDict()
m.pdd_poly1_coeffs_d = aml.ParamDict()
m.pdd_poly2_coeffs_a = aml.ParamDict()
m.pdd_poly2_coeffs_b = aml.ParamDict()
m.pdd_poly2_coeffs_c = aml.ParamDict()
m.pdd_poly2_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list
for node_name in index_over:
node = wn.get_node(node_name)
pmin = node.minimum_pressure
pnom = node.nominal_pressure
x1 = pmin
f1 = 0.0
"""
Add parameters to the model for leak smoothing polynomial coefficients
Parameters
----------
m: wntr.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'leak_poly_coeffs_a'):
m.leak_poly_coeffs_a = aml.ParamDict()
m.leak_poly_coeffs_b = aml.ParamDict()
m.leak_poly_coeffs_c = aml.ParamDict()
m.leak_poly_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list + wn.tank_name_list
for node_name in index_over:
node = wn.get_node(node_name)
x1 = 0.0
f1 = 0.0
x2 = x1 + m.leak_delta
f2 = node.leak_discharge_coeff*node.leak_area*(2.0*9.81*x2)**0.5
df1 = m.leak_slope
df2 = 0.5*node.leak_discharge_coeff*node.leak_area*(2.0*9.81)**0.5*x2**(-0.5)
a, b, c, d = cubic_spline(x1, x2, f1, f2, df1, df2)
if node_name in m.leak_poly_coeffs_a:
m.leak_poly_coeffs_a[node_name].value = a
m.leak_poly_coeffs_b[node_name].value = b
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'pdd_poly1_coeffs_a'):
m.pdd_poly1_coeffs_a = aml.ParamDict()
m.pdd_poly1_coeffs_b = aml.ParamDict()
m.pdd_poly1_coeffs_c = aml.ParamDict()
m.pdd_poly1_coeffs_d = aml.ParamDict()
m.pdd_poly2_coeffs_a = aml.ParamDict()
m.pdd_poly2_coeffs_b = aml.ParamDict()
m.pdd_poly2_coeffs_c = aml.ParamDict()
m.pdd_poly2_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list
for node_name in index_over:
node = wn.get_node(node_name)
pmin = node.minimum_pressure
pnom = node.nominal_pressure
x1 = pmin
f1 = 0.0
x2 = pmin + m.pdd_smoothing_delta
f2 = ((x2 - pmin)/(pnom-pmin))**0.5
df1 = m.pdd_slope
df2 = 0.5*((x2-pmin)/(pnom-pmin))**(-0.5)*1.0/(pnom-pmin)
a1, b1, c1, d1 = cubic_spline(x1, x2, f1, f2, df1, df2)
def build(cls, m, wn, updater, index_over=None):
"""
Add a leak discharge coefficient parameter to the model
Parameters
----------
m: wntr.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction/tank names
"""
if not hasattr(m, 'leak_area'):
m.leak_area = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list + wn.tank_name_list
for node_name in index_over:
node = wn.get_node(node_name)
if node_name in m.leak_area:
m.leak_area[node_name].value = node.leak_area
else:
m.leak_area[node_name] = aml.Param(node.leak_area)
updater.add(node, 'leak_area', leak_area_param.update)
Add parameters to the model for pdd smoothing polynomial coefficients
Parameters
----------
m: wntr.aml.aml.aml.Model
wn: wntr.network.model.WaterNetworkModel
updater: ModelUpdater
index_over: list of str
list of junction names
"""
if not hasattr(m, 'pdd_poly1_coeffs_a'):
m.pdd_poly1_coeffs_a = aml.ParamDict()
m.pdd_poly1_coeffs_b = aml.ParamDict()
m.pdd_poly1_coeffs_c = aml.ParamDict()
m.pdd_poly1_coeffs_d = aml.ParamDict()
m.pdd_poly2_coeffs_a = aml.ParamDict()
m.pdd_poly2_coeffs_b = aml.ParamDict()
m.pdd_poly2_coeffs_c = aml.ParamDict()
m.pdd_poly2_coeffs_d = aml.ParamDict()
if index_over is None:
index_over = wn.junction_name_list
for node_name in index_over:
node = wn.get_node(node_name)
pmin = node.minimum_pressure
pnom = node.nominal_pressure
x1 = pmin
f1 = 0.0
x2 = pmin + m.pdd_smoothing_delta
f2 = ((x2 - pmin)/(pnom-pmin))**0.5
df1 = m.pdd_slope