from .database import * from .s23_options import OPTION_UNITS_CFS, OPTION_UNITS_GPM, OPTION_UNITS_MGD, OPTION_UNITS_IMGD, OPTION_UNITS_AFD, OPTION_UNITS_LPS, OPTION_UNITS_LPM, OPTION_UNITS_MLD, OPTION_UNITS_CMH, OPTION_UNITS_CMD from .s23_options import OPTION_PRESSURE_PSI, OPTION_PRESSURE_KPA, OPTION_PRESSURE_M from .s23_options import OPTION_HEADLOSS_HW, OPTION_HEADLOSS_DW, OPTION_HEADLOSS_CM from .s23_options import OPTION_UNBALANCED_STOP, OPTION_UNBALANCED_CONTINUE from .s23_options import OPTION_QUALITY_NONE, OPTION_QUALITY_CHEMICAL, OPTION_QUALITY_AGE, OPTION_QUALITY_TRACE from .s23_options import element_schema OPTION_V3_FLOW_UNITS_CFS = OPTION_UNITS_CFS OPTION_V3_FLOW_UNITS_GPM = OPTION_UNITS_GPM OPTION_V3_FLOW_UNITS_MGD = OPTION_UNITS_MGD OPTION_V3_FLOW_UNITS_IMGD = OPTION_UNITS_IMGD OPTION_V3_FLOW_UNITS_AFD = OPTION_UNITS_AFD OPTION_V3_FLOW_UNITS_LPS = OPTION_UNITS_LPS OPTION_V3_FLOW_UNITS_LPM = OPTION_UNITS_LPM OPTION_V3_FLOW_UNITS_MLD = OPTION_UNITS_MLD OPTION_V3_FLOW_UNITS_CMH = OPTION_UNITS_CMH OPTION_V3_FLOW_UNITS_CMD = OPTION_UNITS_CMD OPTION_V3_PRESSURE_UNITS_PSI = OPTION_PRESSURE_PSI OPTION_V3_PRESSURE_UNITS_KPA = OPTION_PRESSURE_KPA OPTION_V3_PRESSURE_UNITS_M = OPTION_PRESSURE_M OPTION_V3_HEADLOSS_MODEL_HW = OPTION_HEADLOSS_HW OPTION_V3_HEADLOSS_MODEL_DW = OPTION_HEADLOSS_DW OPTION_V3_HEADLOSS_MODEL_CM = OPTION_HEADLOSS_CM OPTION_V3_STEP_SIZING = 'FULL' OPTION_V3_STEP_SIZING = 'RELAXATION' OPTION_V3_STEP_SIZING = 'LINESEARCH' OPTION_V3_IF_UNBALANCED_STOP = OPTION_UNBALANCED_STOP OPTION_V3_IF_UNBALANCED_CONTINUE = OPTION_UNBALANCED_CONTINUE OPTION_V3_DEMAND_MODEL_FIXED = 'FIXED' OPTION_V3_DEMAND_MODEL_CONSTRAINED = 'CONSTRAINED' OPTION_V3_DEMAND_MODEL_POWER = 'POWER' OPTION_V3_DEMAND_MODEL_LOGISTIC = 'LOGISTIC' OPTION_V3_LEAKAGE_MODEL_NONE = 'NONE' OPTION_V3_LEAKAGE_MODEL_POWER = 'POWER' OPTION_V3_LEAKAGE_MODEL_FAVAD = 'FAVAD' OPTION_V3_QUALITY_MODEL_NONE = OPTION_QUALITY_NONE OPTION_V3_QUALITY_MODEL_CHEMICAL = OPTION_QUALITY_CHEMICAL OPTION_V3_QUALITY_MODEL_AGE = OPTION_QUALITY_AGE OPTION_V3_QUALITY_MODEL_TRACE = OPTION_QUALITY_TRACE OPTION_V3_QUALITY_UNITS_HRS = 'HRS' OPTION_V3_QUALITY_UNITS_PCNT = 'PCNT' OPTION_V3_QUALITY_UNITS_MGL = 'MG/L' OPTION_V3_QUALITY_UNITS_UGL = 'UG/L' def get_option_v3_schema(name: str) -> dict[str, dict[str, Any]]: return { 'UNITS' : element_schema, 'PRESSURE' : element_schema, 'HEADLOSS' : element_schema, 'QUALITY' : element_schema, 'UNBALANCED' : element_schema, 'PATTERN' : element_schema, 'DEMAND MODEL' : element_schema, 'DEMAND MULTIPLIER' : element_schema, 'EMITTER EXPONENT' : element_schema, 'VISCOSITY' : element_schema, 'DIFFUSIVITY' : element_schema, 'SPECIFIC GRAVITY' : element_schema, 'TRIALS' : element_schema, 'ACCURACY' : element_schema, 'HEADERROR' : element_schema, 'FLOWCHANGE' : element_schema, 'MINIMUM PRESSURE' : element_schema, 'REQUIRED PRESSURE' : element_schema, 'PRESSURE EXPONENT' : element_schema, 'TOLERANCE' : element_schema, 'HTOL' : element_schema, 'QTOL' : element_schema, 'RQTOL' : element_schema, 'CHECKFREQ' : element_schema, 'MAXCHECK' : element_schema, 'DAMPLIMIT' : element_schema } def get_option_v3(name: str) -> dict[str, Any]: ts = read_all(name, f"select * from options") d = {} for e in ts: d[e['key']] = str(e['value']) return d def set_option_v3_cmd(name: str, cs: ChangeSet) -> DbChangeSet: raw_old = get_option_v3(name) old = {} new = {} new_dict = cs.operations[0] schema = get_option_v3_schema(name) for key in schema.keys(): if key in new_dict: old[key] = str(raw_old[key]) new[key] = str(new_dict[key]) redo_cs = g_update_prefix | { 'type' : 'option' } redo_sql = '' for key, value in new.items(): if redo_sql != '': redo_sql += '\n' redo_sql += f"update options set value = '{value}' where key = '{key}';" redo_cs |= { key: value } undo_cs = g_update_prefix | { 'type' : 'option' } undo_sql = '' for key, value in old.items(): if undo_sql != '': undo_sql += '\n' undo_sql += f"update options set value = '{value}' where key = '{key}';" undo_cs |= { key: value } return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs]) def set_option_v3(name: str, cs: ChangeSet) -> ChangeSet: return execute_command(name, set_option_v3_cmd(name, cs)) #-------------------------------------------------------------- # [EPANET2][IN][OUT] # UNITS CFS/GPM/MGD/IMGD/AFD/LPS/LPM/MLD/CMH/CMD/SI # PRESSURE PSI/KPA/M # HEADLOSS H-W/D-W/C-M # QUALITY NONE/AGE/TRACE/CHEMICAL (TraceNode) # UNBALANCED STOP/CONTINUE {Niter} # PATTERN id # DEMAND MODEL DDA/PDA # DEMAND MULTIPLIER value # EMITTER EXPONENT value # VISCOSITY value # DIFFUSIVITY value # SPECIFIC GRAVITY value # TRIALS value # ACCURACY value# # HEADERROR value # FLOWCHANGE value # MINIMUM PRESSURE value # REQUIRED PRESSURE value # PRESSURE EXPONENT value# # TOLERANCE value # HTOL value # QTOL value # RQTOL value # CHECKFREQ value # MAXCHECK value # DAMPLIMIT value # ---- Unsupported Options ----- # HYDRAULICS USE/SAVE filename # MAP filename #-------------------------------------------------------------- def inp_in_option_v3(section: list[str]) -> ChangeSet: if len(section) > 0: cs = g_update_prefix | { 'type' : 'option' } for s in section: if s.startswith(';'): continue tokens = s.strip().split() if tokens[0].upper() == 'PATTERN': # can not upper id cs |= { 'PATTERN' : tokens[1] } elif tokens[0].upper() == 'QUALITY': # can not upper trace node value = tokens[1] if len(tokens) > 2: value += f' {tokens[2]}' cs |= { 'QUALITY' : value } else: line = s.upper().strip() for key in get_option_v3_schema('').keys(): if line.startswith(key): value = line.removeprefix(key).strip() cs |= { key : value } return ChangeSet(cs) return ChangeSet() def inp_out_option_v3(name: str) -> list[str]: lines = [] objs = read_all(name, f"select * from options") for obj in objs: key = obj['key'] value = obj['value'] lines.append(f'{key} {value}') return lines