from typing import Any from .project import * from .s1_title import * from .s2_junctions import * from .s3_reservoirs import * from .s4_tanks import * from .s5_pipes import * from .s6_pumps import * from .s7_valves import * from .s9_demands import * from .s10_status import * from .s11_patterns import * from .s12_curves import * from .s16_emitters import * from .s21_times import * from .s23_options import * def read_inp(name: str, inp: str): if is_project_open(name): close_project(name) if have_project(name): delete_project(name) create_project(name) open_project(name) section = '' title : str = '' junctions : dict[str, dict[str, Any]] = {} reservoirs : dict[str, dict[str, Any]] = {} tanks : dict[str, dict[str, Any]] = {} pipes : dict[str, dict[str, Any]] = {} pumps : dict[str, dict[str, Any]] = {} valves : dict[str, dict[str, Any]] = {} demands : dict[str, list[dict[str, Any]]] = {} status : dict[str, dict[str, Any]] = {} patterns : dict[str, list[float]] = {} curves : dict[str, list[dict[str, float]]] = {} emitters : dict[str, float] = {} times : dict[str, str] = {} options : dict[str, str] = {} for line in open(inp): line = line.strip() if line.startswith(';'): continue if line.endswith(';'): line = line.removesuffix(';') if line.startswith('[TITLE'): section = 'title' continue if line.startswith('[JUNCTION'): section = JUNCTION continue if line.startswith('[RESERVOIR'): section = RESERVOIR continue if line.startswith('[TANK'): section = TANK continue if line.startswith('[PIPE'): section = PIPE continue if line.startswith('[PUMP'): section = PUMP continue if line.startswith('[VALVE'): section = VALVE continue if line.startswith('[DEMAND'): section = 'demand' continue if line.startswith('[STATUS'): section = 'status' continue if line.startswith('[PATTERN'): section = PATTERN continue if line.startswith('[CURVE'): section = CURVE continue if line.startswith('[EMITTER'): section = 'emitter' continue if line.startswith('[TIME'): section = 'time' continue if line.startswith('[OPTION'): section = 'option' continue if line.startswith('[COORDINATE'): section = 'coordinate' continue if line.startswith('['): section = '' continue tokens = [] if section == 'time' or section == 'option': tokens = line.upper().split() else: tokens = line.split() tokens_len = len(tokens) if tokens_len == 0: continue if section == 'title': if title == '': title += '\n' title += line continue elif section == JUNCTION: junction_demand = float(tokens[2]) if tokens_len >= 3 else None junction_pattern = tokens[3] if tokens_len == 4 else None junctions[tokens[0]] = {'id': tokens[0], 'elevation': tokens[1], 'demand': junction_demand, 'pattern': junction_pattern} continue elif section == RESERVOIR: reservoir_pattern = tokens[2] if tokens_len == 3 else None reservoirs[tokens[0]] = {'id': tokens[0], 'head': tokens[1], 'pattern': reservoir_pattern} continue elif section == TANK: tank_vol_curve = tokens[7] if tokens_len >= 8 else None tank_overflow = tokens[8].upper() if tokens_len == 9 else None tanks[tokens[0]] = {'id': tokens[0], 'elevation': tokens[1], 'init_level': tokens[2], 'min_level': tokens[3], 'max_level': tokens[4], 'diameter': tokens[5], 'min_vol': tokens[6], 'vol_curve': tank_vol_curve, 'overflow': tank_overflow} continue elif section == PIPE: # status is must-have, here fix input pipe_status = tokens[7].upper() if tokens_len == 8 else PIPE_STATUS_OPEN pipes[tokens[0]] = {'id': tokens[0], 'node1': tokens[1], 'node2': tokens[2], 'length': tokens[3], 'diameter': tokens[4], 'roughness': tokens[5], 'minor_loss': tokens[6], 'status': pipe_status} continue elif section == PUMP: pumps[tokens[0]] = {'id': tokens[0], 'node1': tokens[1], 'node2': tokens[2]} for i in range(3, tokens_len, 2): pumps[tokens[0]] |= { tokens[i].lower(): tokens[i + 1] } continue elif section == VALVE: valves[tokens[0]] = {'id': tokens[0], 'node1': tokens[1], 'node2': tokens[2], 'diameter': tokens[3], 'v_type': tokens[4], 'setting': tokens[5], 'minor_loss': tokens[6]} continue elif section == 'demand': demand_pattern = tokens[2] if tokens_len >= 3 else None demand_category = tokens[3] if tokens_len == 4 else None demands[tokens[0]].append({'demand': tokens[1], 'pattern': demand_pattern, 'category': demand_category}) continue elif section == 'status': if tokens[0] not in status: status[tokens[0]] = {} setting = None try: setting = float(tokens[1]) except: setting = None if setting != None: status[tokens[0]]['setting'] = setting else: status[tokens[0]]['status'] = tokens[1].upper() continue elif section == PATTERN: if tokens[0] not in patterns: patterns[tokens[0]] = [] for i in range(1, tokens_len): patterns[tokens[0]].append(float(tokens[i])) continue elif section == CURVE: if tokens[0] not in curves: curves[tokens[0]] = [] for i in range(1, tokens_len, 2): curves[tokens[0]].append({ 'x': float(tokens[i]), 'y': float(tokens[i + 1]) }) continue elif section == 'emitter': emitters[tokens[0]] = float(tokens[1]) continue elif section == 'time': if tokens_len == 2: times[tokens[0]] = tokens[1] elif tokens_len == 3: times[tokens[0] + ' ' + tokens[1]] = tokens[2] elif tokens_len == 4: times[tokens[0] + ' ' + tokens[1]] = tokens[2] + ' ' + tokens[3] continue elif section == 'option': if tokens[0] == 'HYDRAULICS' or tokens[0] == 'MAP': continue if tokens_len == 2: options[tokens[0]] = tokens[1] elif tokens_len == 3: if tokens[0] == 'UNBALANCED' or tokens[0] == 'QUALITY': options[tokens[0]] = tokens[1] + ' ' + tokens[2] else: options[tokens[0] + ' ' + tokens[1]] = tokens[2] continue elif section == 'coordinate': if tokens[0] in junctions: junctions[tokens[0]] |= {'x': tokens[1], 'y': tokens[2]} elif tokens[0] in reservoirs: reservoirs[tokens[0]] |= {'x': tokens[1], 'y': tokens[2]} elif tokens[0] in tanks: tanks[tokens[0]] |= {'x': tokens[1], 'y': tokens[2]} continue # title set_title(name, ChangeSet({ 'value': title })) # pattern for key, value in patterns.items(): set_pattern(name, ChangeSet({'id': key, 'factors': value})) # curve for key, value in curves.items(): set_curve(name, ChangeSet({'id': key, 'coords': value})) # junction for value in junctions.values(): if 'x' not in value: value['x'] = 0.0 if 'y' not in value: value['y'] = 0.0 add_junction(name, ChangeSet(value)) # reservoir for value in reservoirs.values(): if 'x' not in value: value['x'] = 0.0 if 'y' not in value: value['y'] = 0.0 add_reservoir(name, ChangeSet(value)) # tank for value in tanks.values(): if 'x' not in value: value['x'] = 0.0 if 'y' not in value: value['y'] = 0.0 add_tank(name, ChangeSet(value)) # pipe for value in pipes.values(): add_pipe(name, ChangeSet(value)) # pump for value in pumps.values(): add_pump(name, ChangeSet(value)) # valve for value in valves.values(): add_valve(name, ChangeSet(value)) # demand for key, value in demands.items(): set_demand(name, ChangeSet({'junction': key, 'demands': value})) # status for key, value in status.items(): set_status(name, ChangeSet({'link': key} | value)) # emitter for key, value in emitters.items(): set_emitter(name, ChangeSet({'junction': key, 'coefficient': value})) # time set_time(name, ChangeSet(times)) # option set_option(name, ChangeSet(options)) close_project(name)