334 lines
9.0 KiB
Python
334 lines
9.0 KiB
Python
from typing import Any
|
|
import os
|
|
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 .s8_tags import *
|
|
from .s9_demands import *
|
|
from .s10_status import *
|
|
from .s11_patterns import *
|
|
from .s12_curves import *
|
|
from .s13_controls import *
|
|
from .s14_rules import *
|
|
from .s15_energy import *
|
|
from .s16_emitters import *
|
|
from .s17_quality import *
|
|
from .s18_sources import *
|
|
from .s19_reactions import *
|
|
from .s20_mixing import *
|
|
from .s21_times import *
|
|
from .s22_report import *
|
|
from .s23_options import *
|
|
from .s24_coordinates import *
|
|
from .s25_vertices import *
|
|
from .s26_labels import *
|
|
from .s27_backdrop import *
|
|
from .s28_end import *
|
|
from .command import *
|
|
|
|
section_name = ['TITLE', 'JUNCTIONS', 'RESERVOIRS', 'TANKS', 'PIPES',
|
|
'PUMPS', 'VALVES', 'TAGS', 'DEMANDS', 'STATUS',
|
|
'PATTERNS', 'CURVES', 'CONTROLS', 'RULES', 'ENERGY',
|
|
'EMITTERS', 'QUALITY', 'SOURCES', 'REACTIONS', 'MIXING',
|
|
'TIMES', 'REPORT', 'OPTIONS', 'COORDINATES', 'VERTICES',
|
|
'LABELS', 'BACKDROP', 'END']
|
|
|
|
def _parse_inp(inp: str) -> dict[str, list[str]]:
|
|
file: dict[str, list[str]] = {}
|
|
for s in section_name:
|
|
file[s] = []
|
|
|
|
section = ''
|
|
|
|
for line in open(inp):
|
|
line = line.strip()
|
|
if line == '':
|
|
# skip empty line for control and rule
|
|
if section == 'CONTROLS' or section == 'RULES':
|
|
pass
|
|
else:
|
|
section = ''
|
|
continue
|
|
|
|
if line.startswith('['):
|
|
is_section = False
|
|
for s in section_name:
|
|
if line.startswith(f'[{s}'):
|
|
section = s
|
|
is_section = True
|
|
break
|
|
if is_section:
|
|
continue
|
|
|
|
if section != '':
|
|
file[section].append(line)
|
|
|
|
return file
|
|
|
|
|
|
def _read_inp(inp: str) -> ChangeSet:
|
|
file = _parse_inp(inp)
|
|
|
|
file_cs: dict[str, ChangeSet] = {}
|
|
for s in section_name:
|
|
file_cs[s] = ChangeSet()
|
|
|
|
for name, section in file.items():
|
|
if name == 'TITLE':
|
|
file_cs[name].merge(inp_in_title(section))
|
|
|
|
elif name == 'JUNCTIONS': # + coords
|
|
file_cs[name].merge(inp_in_junction(section))
|
|
|
|
elif name == 'RESERVOIRS': # + coords
|
|
file_cs[name].merge(inp_in_reservoir(section))
|
|
|
|
elif name == 'TANKS': # + coords
|
|
file_cs[name].merge(inp_in_tank(section))
|
|
|
|
elif name == 'PIPES':
|
|
file_cs[name].merge(inp_in_pipe(section))
|
|
|
|
elif name == 'PUMPS':
|
|
file_cs[name].merge(inp_in_pump(section))
|
|
|
|
elif name == 'VALVES':
|
|
file_cs[name].merge(inp_in_valve(section))
|
|
|
|
elif name == 'TAGS':
|
|
file_cs[name].merge(inp_in_tag(section))
|
|
|
|
elif name == 'DEMANDS':
|
|
file_cs[name].merge(inp_in_demand(section))
|
|
|
|
elif name == 'STATUS':
|
|
file_cs[name].merge(inp_in_status(section))
|
|
|
|
elif name == 'PATTERNS':
|
|
file_cs[name].merge(inp_in_pattern(section))
|
|
|
|
elif name == 'CURVES':
|
|
file_cs[name].merge(inp_in_curve(section))
|
|
|
|
elif name == 'CONTROLS':
|
|
file_cs[name].merge(inp_in_control(section))
|
|
|
|
elif name == 'RULES':
|
|
file_cs[name].merge(inp_in_rule(section))
|
|
|
|
elif name == 'ENERGY':
|
|
file_cs[name].merge(inp_in_energy(section))
|
|
|
|
elif name == 'EMITTERS':
|
|
file_cs[name].merge(inp_in_emitter(section))
|
|
|
|
elif name == 'QUALITY':
|
|
file_cs[name].merge(inp_in_quality(section))
|
|
|
|
elif name == 'SOURCES':
|
|
file_cs[name].merge(inp_in_source(section))
|
|
|
|
elif name == 'REACTIONS':
|
|
file_cs[name].merge(inp_in_reaction(section))
|
|
|
|
elif name == 'MIXING':
|
|
file_cs[name].merge(inp_in_mixing(section))
|
|
|
|
elif name == 'TIMES':
|
|
file_cs[name].merge(inp_in_time(section))
|
|
|
|
elif name == 'REPORT':
|
|
pass # ignore now, for simulation, always report all
|
|
|
|
elif name == 'OPTIONS':
|
|
file_cs[name].merge(inp_in_option(section))
|
|
|
|
elif name == 'COORDINATES':
|
|
coords = inp_in_coord(section)
|
|
for s in ['JUNCTIONS', 'RESERVOIRS', 'TANKS']:
|
|
for node in file_cs[s].operations:
|
|
if node['id'] in coords:
|
|
coord = coords[node['id']]
|
|
node |= { 'x' : coord['x'], 'y' : coord['y'] }
|
|
else:
|
|
print(f"WARNING: [{s}] {node['id']} has no coordinate, set it at origin!")
|
|
node |= { 'x' : 0.0, 'y' : 0.0 }
|
|
|
|
elif name == 'VERTICES':
|
|
file_cs[name].merge(inp_in_vertex(section))
|
|
|
|
elif name == 'LABELS':
|
|
file_cs[name].merge(inp_in_label(section))
|
|
|
|
elif name == 'BACKDROP':
|
|
file_cs[name].merge(inp_in_backdrop(section))
|
|
|
|
elif name == 'END':
|
|
pass # :)
|
|
|
|
cs = ChangeSet()
|
|
priorities = [
|
|
'PATTERNS',
|
|
'CURVES',
|
|
'JUNCTIONS',
|
|
'RESERVOIRS',
|
|
'TANKS',
|
|
'COORDINATES',
|
|
'PIPES',
|
|
'PUMPS',
|
|
'VALVES',
|
|
'DEMANDS',
|
|
'STATUS',
|
|
'OPTIONS',
|
|
'TIMES',
|
|
'EMITTERS',
|
|
'QUALITY',
|
|
'SOURCES',
|
|
'REACTIONS',
|
|
'MIXING',
|
|
'ENERGY',
|
|
'REPORT',
|
|
'VERTICES',
|
|
'CONTROLS',
|
|
'RULES',
|
|
'TITLE',
|
|
'TAGS',
|
|
'LABELS',
|
|
'BACKDROP',
|
|
'END',
|
|
]
|
|
for s in priorities:
|
|
cs.merge(file_cs[s])
|
|
|
|
return cs
|
|
|
|
|
|
def read_inp(project: str, inp: str):
|
|
if is_project_open(project):
|
|
close_project(project)
|
|
|
|
if have_project(project):
|
|
delete_project(project)
|
|
|
|
create_project(project)
|
|
open_project(project)
|
|
|
|
execute_batch_command(project, _read_inp(inp))
|
|
|
|
|
|
def dump_inp(project: str, inp: str):
|
|
if not have_project(project):
|
|
return
|
|
|
|
if not is_project_open(project):
|
|
open_project(project)
|
|
|
|
dir = os.getcwd()
|
|
path = os.path.join(dir, inp)
|
|
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
|
|
file = open(path, mode='w')
|
|
|
|
for name in section_name:
|
|
if name == 'TITLE':
|
|
file.write(f'[{name}]\n')
|
|
else:
|
|
file.write(f'\n[{name}]\n')
|
|
|
|
if name == 'TITLE':
|
|
file.write('\n'.join(inp_out_title(project)))
|
|
|
|
elif name == 'JUNCTIONS': # + coords
|
|
file.write('\n'.join(inp_out_junction(project)))
|
|
|
|
elif name == 'RESERVOIRS': # + coords
|
|
file.write('\n'.join(inp_out_reservoir(project)))
|
|
|
|
elif name == 'TANKS': # + coords
|
|
file.write('\n'.join(inp_out_tank(project)))
|
|
|
|
elif name == 'PIPES':
|
|
file.write('\n'.join(inp_out_pipe(project)))
|
|
|
|
elif name == 'PUMPS':
|
|
file.write('\n'.join(inp_out_pump(project)))
|
|
|
|
elif name == 'VALVES':
|
|
file.write('\n'.join(inp_out_valve(project)))
|
|
|
|
elif name == 'TAGS':
|
|
file.write('\n'.join(inp_out_tag(project)))
|
|
|
|
elif name == 'DEMANDS':
|
|
file.write('\n'.join(inp_out_demand(project)))
|
|
|
|
elif name == 'STATUS':
|
|
file.write('\n'.join(inp_out_status(project)))
|
|
|
|
elif name == 'PATTERNS':
|
|
file.write('\n'.join(inp_out_pattern(project)))
|
|
|
|
elif name == 'CURVES':
|
|
file.write('\n'.join(inp_out_curve(project)))
|
|
|
|
elif name == 'CONTROLS':
|
|
file.write('\n'.join(inp_out_control(project)))
|
|
|
|
elif name == 'RULES':
|
|
file.write('\n'.join(inp_out_rule(project)))
|
|
|
|
elif name == 'ENERGY':
|
|
file.write('\n'.join(inp_out_energy(project)))
|
|
|
|
elif name == 'EMITTERS':
|
|
file.write('\n'.join(inp_out_emitter(project)))
|
|
|
|
elif name == 'QUALITY':
|
|
file.write('\n'.join(inp_out_quality(project)))
|
|
|
|
elif name == 'SOURCES':
|
|
file.write('\n'.join(inp_out_source(project)))
|
|
|
|
elif name == 'REACTIONS':
|
|
file.write('\n'.join(inp_out_reaction(project)))
|
|
|
|
elif name == 'MIXING':
|
|
file.write('\n'.join(inp_out_mixing(project)))
|
|
|
|
elif name == 'TIMES':
|
|
file.write('\n'.join(inp_out_time(project)))
|
|
|
|
elif name == 'REPORT':
|
|
file.write('\n'.join(inp_out_report(project)))
|
|
|
|
elif name == 'OPTIONS':
|
|
file.write('\n'.join(inp_out_option(project)))
|
|
|
|
elif name == 'COORDINATES':
|
|
file.write('\n'.join(inp_out_coord(project)))
|
|
|
|
elif name == 'VERTICES':
|
|
file.write('\n'.join(inp_out_vertex(project)))
|
|
|
|
elif name == 'LABELS':
|
|
file.write('\n'.join(inp_out_label(project)))
|
|
|
|
elif name == 'BACKDROP':
|
|
file.write('\n'.join(inp_out_backdrop(project)))
|
|
|
|
elif name == 'END':
|
|
pass # :)
|
|
|
|
file.write('\n')
|
|
|
|
file.close()
|
|
|
|
close_project(project)
|