import datetime import os from .project import * from .database import ChangeSet, write from .sections import * from .s1_title import inp_in_title from .s2_junctions import inp_in_junction from .s3_reservoirs import inp_in_reservoir from .s4_tanks import inp_in_tank from .s5_pipes import inp_in_pipe from .s6_pumps import inp_in_pump from .s7_valves import inp_in_valve from .s8_tags import inp_in_tag from .s9_demands import inp_in_demand from .s10_status import inp_in_status from .s11_patterns import pattern_v3_types, inp_in_pattern from .s12_curves import curve_types, inp_in_curve from .s13_controls import inp_in_control from .s14_rules import inp_in_rule from .s15_energy import inp_in_energy from .s16_emitters import inp_in_emitter from .s17_quality import inp_in_quality from .s18_sources import inp_in_source from .s19_reactions import inp_in_reaction from .s20_mixing import inp_in_mixing from .s21_times import inp_in_time from .s22_report import inp_in_report from .s23_options import inp_in_option from .s23_options_v3 import inp_in_option_v3 from .s24_coordinates import inp_in_coord from .s25_vertices import inp_in_vertex from .s26_labels import inp_in_label from .s27_backdrop import inp_in_backdrop _S = 'S' _L = 'L' def _inp_in_option(section: list[str], version: str = '3') -> str: return inp_in_option_v3(section) if version == '3' else inp_in_option(section) _handler = { TITLE : (_S, inp_in_title), JUNCTIONS : (_L, inp_in_junction), # line, demand_outside RESERVOIRS : (_L, inp_in_reservoir), TANKS : (_L, inp_in_tank), PIPES : (_L, inp_in_pipe), PUMPS : (_L, inp_in_pump), VALVES : (_L, inp_in_valve), TAGS : (_L, inp_in_tag), DEMANDS : (_L, inp_in_demand), STATUS : (_L, inp_in_status), PATTERNS : (_L, inp_in_pattern), # line, fixed CURVES : (_L, inp_in_curve), CONTROLS : (_L, inp_in_control), RULES : (_L, inp_in_rule), ENERGY : (_L, inp_in_energy), EMITTERS : (_L, inp_in_emitter), QUALITY : (_L, inp_in_quality), SOURCES : (_L, inp_in_source), REACTIONS : (_L, inp_in_reaction), MIXING : (_L, inp_in_mixing), TIMES : (_S, inp_in_time), REPORT : (_S, inp_in_report), OPTIONS : (_S, _inp_in_option), # line, version COORDINATES : (_L, inp_in_coord), VERTICES : (_L, inp_in_vertex), LABELS : (_L, inp_in_label), BACKDROP : (_S, inp_in_backdrop), #END : 'END', } _level_1 = [ TITLE, PATTERNS, CURVES, CONTROLS, RULES, TIMES, REPORT, OPTIONS, BACKDROP, ] _level_2 = [ JUNCTIONS, RESERVOIRS, TANKS, ] _level_3 = [ PIPES, PUMPS, VALVES, DEMANDS, EMITTERS, QUALITY, SOURCES, MIXING, COORDINATES, LABELS, ] _level_4 = [ TAGS, STATUS, ENERGY, REACTIONS, VERTICES, ] class SQLBatch: def __init__(self, project: str, count: int = 100) -> None: self.batch: list[str] = [] self.project = project self.count = count def add(self, sql: str) -> None: self.batch.append(sql) if len(self.batch) == self.count: self.flush() def flush(self) -> None: write(self.project, ''.join(self.batch)) self.batch.clear() def _print_time(desc: str) -> datetime.datetime: now = datetime.datetime.now() time = now.strftime('%Y-%m-%d %H:%M:%S') print(f"{time}: {desc}") return now def _get_file_offset(inp: str) -> tuple[dict[str, list[int]], bool]: offset: dict[str, list[int]] = {} current = '' demand_outside = False with open(inp) as f: while True: line = f.readline() if not line: break line = line.strip() if line.startswith('['): for s in section_name: if line.startswith(f'[{s}'): if s not in offset: offset[s] = [] offset[s].append(f.tell()) current = s break elif line != '' and line.startswith(';') == False: if current == DEMANDS: demand_outside = True return (offset, demand_outside) def parse_file(project: str, inp: str, version: str = '3') -> None: start = _print_time(f'Start reading file "{inp}"...') _print_time("First scan...") offset, demand_outside = _get_file_offset(inp) levels = _level_1 + _level_2 + _level_3 + _level_4 # parse the whole section rather than line sections : dict[str, list[str]]= {} for [s, t] in _handler.items(): if t[0] == _S: sections[s] = [] variable_patterns = [] current_pattern = None current_curve = None curve_type_desc_line = None sql_batch = SQLBatch(project) _print_time("Second scan...") with open(inp) as f: for s in levels: if s not in offset: continue if s == DEMANDS and demand_outside == False: continue _print_time(f"[{s}]") is_s = _handler[s][0] == _S handler = _handler[s][1] for ptr in offset[s]: f.seek(ptr) while True: line = f.readline() if not line: break line = line.strip() if line.startswith('['): break elif line == '': continue if is_s: sections[s].append(line) else: if line.startswith(';'): if version != '3': #v2 line = line.removeprefix(';') if s == PATTERNS: # ;desc pass elif s == CURVES: # ;type: desc curve_type_desc_line = line continue if s == PATTERNS: tokens = line.split() if tokens[1].upper() in pattern_v3_types: #v3 sql_batch.add(f"insert into _pattern (id) values ('{tokens[0]}');") current_pattern = tokens[0] if tokens[1].upper() == 'VARIABLE': variable_patterns.append(tokens[0]) continue if current_pattern != tokens[0]: sql_batch.add(f"insert into _pattern (id) values ('{tokens[0]}');") current_pattern = tokens[0] elif s == CURVES: tokens = line.split() if tokens[1].upper() in curve_types: #v3 sql_batch.add(f"insert into _curve (id, type) values ('{tokens[0]}', '{tokens[1].upper()}');") current_curve = tokens[0] continue if current_curve != tokens[0]: type = curve_types[0] if curve_type_desc_line != None: type = curve_type_desc_line.split(':')[0].strip() sql_batch.add(f"insert into _curve (id, type) values ('{tokens[0]}', '{type}');") current_curve = tokens[0] curve_type_desc_line = None if s == JUNCTIONS: sql_batch.add(handler(line, demand_outside)) elif s == PATTERNS: sql_batch.add(handler(line, current_pattern not in variable_patterns)) elif s == OPTIONS: sql_batch.add(handler(line, version)) else: sql_batch.add(handler(line)) f.seek(0) if is_s: sql_batch.add(handler(sections[s])) sql_batch.flush() end = _print_time(f'End reading file "{inp}"') print(f"Total (in second): {(end-start).seconds}(s)") def read_inp(project: str, inp: str, version: str = '3') -> bool: if version != '3' and version != '2': version = '2' if is_project_open(project): close_project(project) if have_project(project): delete_project(project) create_project(project) open_project(project) parse_file(project, inp, version) '''try: parse_file(project, inp, version) except: close_project(project) delete_project(project) return False''' close_project(project) return True def import_inp(project: str, cs: ChangeSet, version: str = '3') -> bool: if version != '3' and version != '2': version = '2' if 'inp' not in cs.operations[0]: return False filename = f'inp/{project}_temp.inp' if os.path.exists(filename): os.remove(filename) _print_time(f'Start writing temp file "{filename}"...') with open(filename, 'w') as f: f.write(str(cs.operations[0]['inp'])) _print_time(f'End writing temp file "{filename}"...') result = read_inp(project, filename, version) os.remove(filename) return result