Parse [PIPES]

api/s5_pipes.py

@@ -121,3 +121,48 @@ def delete_pipe_cache(name: str, cs: ChangeSet) -> SqlChangeSet:
 
 def delete_pipe(name: str, cs: ChangeSet) -> ChangeSet:
     return execute_command(name, delete_pipe_cache(name, cs))
+
+
+class InpPipe:
+    def __init__(self, line: str) -> None:
+        tokens = line.split()
+
+        num = len(tokens)
+        has_desc = tokens[-1].startswith(';')
+        num_without_desc = (num - 1) if has_desc else num
+
+        self.id = str(tokens[0])
+        self.node1 = str(tokens[1])
+        self.node2 = str(tokens[2])
+        self.length = float(tokens[3])
+        self.diameter = float(tokens[4])
+        self.roughness = float(tokens[5])
+        self.minor_loss = float(tokens[6])
+        # status is must-have, here fix input
+        self.status = str(tokens[7]) if num_without_desc >= 8 else PIPE_STATUS_OPEN
+        self.desc = str(tokens[-1]) if has_desc else None
+
+
+def inp_in_pipe(section: list[str]) -> ChangeSet:
+    cs = ChangeSet()
+    for s in section:
+        obj = InpPipe(s)
+        cs.append({'operation': API_ADD, 'type': 'pipe', 'id': obj.id, 'node1': obj.node1, 'node2': obj.node2, 'length': obj.length, 'diameter': obj.diameter, 'roughness': obj.roughness, 'minor_loss': obj.minor_loss, 'status': obj.status})
+    return cs
+
+
+def inp_out_pipe(name: str) -> list[str]:
+    lines = []
+    objs = read_all(name, 'select * from pipes')
+    for obj in objs:
+        id = obj['id']
+        node1 = obj['node1']
+        node2 = obj['node2']
+        length = obj['length']
+        diameter = obj['diameter']
+        roughness = obj['roughness']
+        minor_loss = obj['minor_loss']
+        status = obj['vol_curve']
+        desc = ';'
+        lines.append(f'{id} {node1} {node2} {length} {diameter} {roughness} {minor_loss} {status} {desc}')
+    return lines