Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
elif self in [HydParam.RoughnessCoeff] and darcy_weisbach:
if flow_units.is_traditional:
data = data / (1000.0*0.3048) # 1e-3 ft from m
elif flow_units.is_metric:
data = data / 0.001 # mm from m
elif self in [HydParam.TankDiameter, HydParam.Elevation, HydParam.HydraulicHead,
HydParam.Length]:#, HydParam.HeadLoss]:
if flow_units.is_traditional:
data = data / 0.3048 # ft from m
elif self in [HydParam.Velocity]:
if flow_units.is_traditional:
data = data / 0.3048 # ft/s from m/s
elif self in [HydParam.Energy]:
data = data / 3600000.0 # kW*hr from J
elif self in [HydParam.Power]:
if flow_units.is_traditional:
data = data / 745.699872 # hp from W (Nm/s)
elif flow_units.is_metric:
data = data / 1000.0 # kW from W (Nm/s)
elif self in [HydParam.Pressure]:
if flow_units.is_traditional:
# data = data / 0.703249614902
data = data / (0.3048/0.4333) # psi from mH2O, i.e. psi / (m/ft / psi/ft) )
elif self in [HydParam.Volume]:
if flow_units.is_traditional:
data = data / np.power(0.3048, 3) # ft3 from m3
lnames = list(wn._pumps.keys())
lnames.sort()
for pump_name in lnames:
pump = wn._pumps[pump_name]
E = {'name': pump_name,
'node1': pump.start_node,
'node2': pump.end_node,
'ptype': pump.info_type,
'params': '',
'speed_keyword': 'SPEED',
'speed': pump.speed_timeseries.base_value,
'com': ';'}
if pump.info_type == 'HEAD':
E['params'] = pump.curve.name
elif pump.info_type == 'POWER':
E['params'] = str(from_si(self.flow_units, pump.power, HydParam.Power))
else:
raise RuntimeError('Only head or power info is supported of pumps.')
tmp_entry = _PUMP_ENTRY
if pump.speed_timeseries.pattern is not None:
tmp_entry = (tmp_entry.rstrip('\n').rstrip('}').rstrip('com:>3s').rstrip(' {') +
' {pattern_keyword:10s} {pattern:20s} {com:>3s}\n')
E['pattern_keyword'] = 'PATTERN'
E['pattern'] = pump.speed_timeseries.pattern.name
f.write(tmp_entry.format(**E).encode('ascii'))
f.write('\n'.encode('ascii'))
if flow_units.is_traditional:
data = data / 0.7032 # flowunit/psi0.5 to flowunit/m0.5
elif self in [HydParam.PipeDiameter]:
if flow_units.is_traditional:
data = data * 0.0254 # in to m
elif flow_units.is_metric:
data = data * 0.001 # mm to m
elif self in [HydParam.RoughnessCoeff] and darcy_weisbach:
if flow_units.is_traditional:
data = data * (1000.0*0.3048) # 1e-3 ft to m
elif flow_units.is_metric:
data = data * 0.001 # mm to m
elif self in [HydParam.TankDiameter, HydParam.Elevation, HydParam.HydraulicHead,
HydParam.Length]:#, HydParam.HeadLoss]:
if flow_units.is_traditional:
data = data * 0.3048 # ft to m
elif self in [HydParam.Velocity]:
if flow_units.is_traditional:
data = data * 0.3048 # ft/s to m/s
elif self in [HydParam.Energy]:
data = data * 3600000.0 # kW*hr to J
elif self in [HydParam.Power]:
if flow_units.is_traditional:
data = data * 745.699872 # hp to W (Nm/s)
elif flow_units.is_metric:
data = data * 1000.0 # kW to W (Nm/s)
f.write(';PUMP: {}\n'.format(curve_name).encode('ascii'))
for point in curve.points:
x = from_si(self.flow_units, point[0], HydParam.Flow)
y = from_si(self.flow_units, point[1], HydParam.HydraulicHead)
f.write(_CURVE_ENTRY.format(name=curve_name, x=x, y=y, com=';').encode('ascii'))
elif curve.curve_type == 'EFFICIENCY':
f.write(';EFFICIENCY: {}\n'.format(curve_name).encode('ascii'))
for point in curve.points:
x = from_si(self.flow_units, point[0], HydParam.Flow)
y = point[1]
f.write(_CURVE_ENTRY.format(name=curve_name, x=x, y=y, com=';').encode('ascii'))
elif curve.curve_type == 'HEADLOSS':
f.write(';HEADLOSS: {}\n'.format(curve_name).encode('ascii'))
for point in curve.points:
x = from_si(self.flow_units, point[0], HydParam.Flow)
y = from_si(self.flow_units, point[1], HydParam.HeadLoss)
f.write(_CURVE_ENTRY.format(name=curve_name, x=x, y=y, com=';').encode('ascii'))
else:
f.write(';UNKNOWN: {}\n'.format(curve_name).encode('ascii'))
for point in curve.points:
x = point[0]
y = point[1]
f.write(_CURVE_ENTRY.format(name=curve_name, x=x, y=y, com=';').encode('ascii'))
f.write('\n'.encode('ascii'))
f.write('\n'.encode('ascii'))
the values to save, in the node or link order specified earlier in the file
"""
if result_type in [ResultType.quality, ResultType.linkquality]:
if self.quality_type is QualType.Chem:
values = QualParam.Concentration._to_si(self.flow_units, values, mass_units=self.mass_units)
elif self.quality_type is QualType.Age:
values = QualParam.WaterAge._to_si(self.flow_units, values)
elif result_type == ResultType.demand:
values = HydParam.Demand._to_si(self.flow_units, values)
elif result_type == ResultType.flowrate:
values = HydParam.Flow._to_si(self.flow_units, values)
elif result_type == ResultType.head:
values = HydParam.HydraulicHead._to_si(self.flow_units, values)
elif result_type == ResultType.pressure:
values = HydParam.Pressure._to_si(self.flow_units, values)
elif result_type == ResultType.velocity:
values = HydParam.Velocity._to_si(self.flow_units, values)
if result_type in self.items:
if result_type.is_node:
self.results.node[result_type.name].iloc[period] = values
else:
self.results.link[result_type.name].iloc[period] = values
if wn.options.energy.global_price is not None:
f.write('GLOBAL PRICE {:.4f}\n'.format(to_si(self.flow_units, wn.options.energy.global_price, HydParam.Energy)).encode('ascii'))
if wn.options.energy.global_pattern is not None:
f.write('GLOBAL PATTERN {:s}\n'.format(wn.options.energy.global_pattern).encode('ascii'))
if wn.options.energy.global_efficiency is not None:
f.write('GLOBAL EFFIC {:.4f}\n'.format(wn.options.energy.global_efficiency).encode('ascii'))
if wn.options.energy.demand_charge is not None:
f.write('DEMAND CHARGE {:.4f}\n'.format(wn.options.energy.demand_charge).encode('ascii'))
lnames = list(wn._pumps.keys())
lnames.sort()
for pump_name in lnames:
pump = wn._pumps[pump_name]
if pump.efficiency is not None:
f.write('PUMP {:10s} EFFIC {:s}\n'.format(pump_name, pump.efficiency.name).encode('ascii'))
if pump.energy_price is not None:
f.write('PUMP {:10s} PRICE {:s}\n'.format(pump_name, to_si(self.flow_units, pump.energy_price, HydParam.Energy)).encode('ascii'))
if pump.energy_pattern is not None:
f.write('PUMP {:10s} PATTERN {:s}\n'.format(pump_name, pump.energy_pattern).encode('ascii'))
f.write('\n'.encode('ascii'))
def _write_tanks(self, f, wn):
f.write('[TANKS]\n'.encode('ascii'))
f.write(_TANK_LABEL.format(';ID', 'Elevation', 'Init Level', 'Min Level', 'Max Level',
'Diameter', 'Min Volume', 'Volume Curve').encode('ascii'))
nnames = list(wn._tanks.keys())
nnames.sort()
for tank_name in nnames:
tank = wn._tanks[tank_name]
E = {'name': tank_name,
'elev': from_si(self.flow_units, tank.elevation, HydParam.Elevation),
'initlev': from_si(self.flow_units, tank.init_level, HydParam.HydraulicHead),
'minlev': from_si(self.flow_units, tank.min_level, HydParam.HydraulicHead),
'maxlev': from_si(self.flow_units, tank.max_level, HydParam.HydraulicHead),
'diam': from_si(self.flow_units, tank.diameter, HydParam.TankDiameter),
'minvol': from_si(self.flow_units, tank.min_vol, HydParam.Volume),
'curve': '',
'com': ';'}
if tank.vol_curve is not None:
E['curve'] = tank.vol_curve.name
f.write(_TANK_ENTRY.format(**E).encode('ascii'))
f.write('\n'.encode('ascii'))
data = np.array(data)
# Do onversions
if self in [HydParam.Demand, HydParam.Flow, HydParam.EmitterCoeff]:
data = data / flow_units.factor
if self is HydParam.EmitterCoeff:
if flow_units.is_traditional:
data = data / 0.7032 # flowunit/psi0.5 from flowunit/m0.5
elif self in [HydParam.PipeDiameter]:
if flow_units.is_traditional:
data = data / 0.0254 # in from m
elif flow_units.is_metric:
data = data / 0.001 # mm from m
elif self in [HydParam.RoughnessCoeff] and darcy_weisbach:
if flow_units.is_traditional:
data = data / (1000.0*0.3048) # 1e-3 ft from m
elif flow_units.is_metric:
data = data / 0.001 # mm from m
elif self in [HydParam.TankDiameter, HydParam.Elevation, HydParam.HydraulicHead,
HydParam.Length]:#, HydParam.HeadLoss]:
if flow_units.is_traditional:
data = data / 0.3048 # ft from m
elif self in [HydParam.Velocity]:
if flow_units.is_traditional:
data = data / 0.3048 # ft/s from m/s
elif self in [HydParam.Energy]:
data = data / 3600000.0 # kW*hr from J
"""
# Convert to array for unit conversion
data_type = type(data)
if data_type is pd.core.frame.DataFrame:
data_index = data.index
data_columns = data.columns
data = data.values
elif data_type is dict:
data_keys = data.keys()
data = np.array(list(data.values()))
elif data_type is list:
data = np.array(data)
# Do conversions
if self in [HydParam.Demand, HydParam.Flow, HydParam.EmitterCoeff]:
data = data * flow_units.factor
if self is HydParam.EmitterCoeff:
if flow_units.is_traditional:
data = data / 0.7032 # flowunit/psi0.5 to flowunit/m0.5
elif self in [HydParam.PipeDiameter]:
if flow_units.is_traditional:
data = data * 0.0254 # in to m
elif flow_units.is_metric:
data = data * 0.001 # mm to m
elif self in [HydParam.RoughnessCoeff] and darcy_weisbach:
if flow_units.is_traditional:
data = data * (1000.0*0.3048) # 1e-3 ft to m
elif flow_units.is_metric:
data = data * 0.001 # mm to m