Refactor junction

api/s2_junctions.py

@@ -4,71 +4,36 @@ from .s0_base import *
 from .operation import *
 from .change_set import ChangeSet
 from .s24_coordinates import *
+from .utility import *
 
 
 def add_junction(name: str, id: str, x: float, y: float, elevation: float) -> ChangeSet:
-    if is_node(name, id):
-        return
+    sql = f"insert into junctions (id, elevation) values ('{id}', {elevation});"
+    undo_sql = f'delete from junctions where id = "{id}";'
+    return add_node(name, JUNCTION, id, x, y, sql, undo_sql)
 
-    with conn[name].cursor() as cur:
-        sql = f"insert into _node (id, type) values ('{id}', 'JUNCTION');"
-        sql += f" insert into junctions (id, elevation) values ('{id}', {elevation});"
-        sql += f" insert into coordinates (node, coord) values ('{id}', '({x}, {y})');"
-        cur.execute(sql)
 
-        redo = sql.replace("'", '"')
-        undo = f'delete from coordinates where node = "{id}";'
-        undo += f' delete from junctions where id = "{id}";'
-        undo += f' delete from _node where id = "{id}";'
-        add_operation(name, redo, undo)
-
-    change = ChangeSet()
-    change.add('junction', id)
-    return change
+def _get_junction(name: str, id: str) -> Row | None:
+    return query(f"select elevation, demand, pattern from junctions where id = '{id}'")
 
 
 def delete_junction(name: str, id: str) -> ChangeSet:
     if not is_junction(name, id):
         return
 
-    with conn[name].cursor(row_factory=dict_row) as cur:
-        cur.execute(f"select * from junctions where id = '{id}'")
-        row = cur.fetchone()
-        if row == None:
-            return
+    row = _get_junction(name, id)
+    if row == None:
+        return
 
-        elevation = row['elevation']
-        demand = 'NULL' if row['demand'] == None else row['demand']
-        pattern = 'NULL' if row['pattern'] == None else row['pattern']
-        pattern = f'"{pattern}"' if pattern != 'NULL' else pattern
+    elevation = row['elevation']
+    demand = 'NULL' if row['demand'] == None else row['demand']
+    pattern = 'NULL' if row['pattern'] == None else row['pattern']
+    pattern = f'"{pattern}"' if pattern != 'NULL' else pattern
 
-        cur.execute(f"select * from coordinates where node = '{id}'")
-        row = cur.fetchone()
-        if row == None:
-            return
+    sql = f"delete from junctions where id = '{id}';"
+    undo_sql = f'insert into junctions (id, elevation, demand, pattern) values ("{id}", {elevation}, {demand}, {pattern});'
 
-        coord = row['coord']
-
-        sql = f"delete from coordinates where node = '{id}';"
-        sql += f" delete from junctions where id = '{id}';"
-        sql += f" delete from _node where id = '{id}';"
-        cur.execute(sql)
-
-        redo = sql.replace("'", '"')
-        undo = f'insert into _node (id, type) values ("{id}", "JUNCTION");'
-        undo += f' insert into junctions (id, elevation, demand, pattern) values ("{id}", {elevation}, {demand}, {pattern});'
-        undo += f' insert into coordinates (node, coord) values ("{id}", "{coord}");'
-        add_operation(name, redo, undo)
-
-    change = ChangeSet()
-    change.delete('junction', id)
-    return change
-
-
-def _get_junction(name: str, id: str) -> Row | None:
-    with conn[name].cursor(row_factory=dict_row) as cur:
-        cur.execute(f"select elevation, demand, pattern from junctions where id = '{id}'")
-        return cur.fetchone()
+    return delete_node(name, JUNCTION, id, sql, undo_sql)
 
 
 def get_junction_elevation(name: str, id: str) -> float | None:
@@ -96,68 +61,35 @@ def get_junction_coord(name: str, id: str) -> dict[str, float] | None:
     return get_node_coord(name, id)
 
 
-def set_junction_elevation(name: str, id: str, elevation: float) -> ChangeSet:
+def _set_junction(name: str, id: str, key: str, key_type: str, value: str, optional: bool = False) -> ChangeSet:
     if not is_junction(name, id):
         return
 
-    old = get_junction_elevation(name, id)
-    if old == None:
+    row = _get_junction(name, id)
+    if row == None:
         return
 
-    with conn[name].cursor() as cur:
-        sql = f"update junctions set elevation = {elevation} where id = '{id}'"
-        cur.execute(sql)
-        redo = sql.replace("'", '"')
-        undo = f'update junctions set elevation = {old} where id = "{id}"'
-        add_operation(name, redo, undo)
+    old = decorate(row[key])
+
+    sql = f"update junctions set {key} = {value} where id = '{id}'"
+    undo = f'update junctions set {key} = {old} where id = "{id}"'
+    update(name, sql, undo)
 
     change = ChangeSet()
-    change.update('junction', id, 'elevation', 'float', str(elevation))
+    change.update('junction', id, key, key_type, value)
     return change
 
 
+def set_junction_elevation(name: str, id: str, elevation: float) -> ChangeSet:
+    return _set_junction(name, id, 'elevation', 'float', str(elevation))
+
+
 def set_junction_demand(name: str, id: str, demand: float) -> ChangeSet:
-    if not is_junction(name, id):
-        return
-
-    old = get_junction_demand(name, id)
-    if old == None:
-        return
-
-    with conn[name].cursor() as cur:
-        sql = f"update junctions set demand = {demand} where id = '{id}'"
-        cur.execute(sql)
-        redo = sql.replace("'", '"')
-        undo = f'update junctions set demand = {old} where id = "{id}"'
-        add_operation(name, redo, undo)
-
-    change = ChangeSet()
-    change.update('junction', id, 'demand', 'float', str(demand))
-    return change
+    return _set_junction(name, id, 'demand', 'float', str(demand), True)
 
 
 def set_junction_pattern(name: str, id: str, pattern: str) -> ChangeSet:
-    if not is_junction(name, id):
-        return
-    if not is_pattern(name, id):
-        return
-
-    old = get_junction_pattern(name, id)
-    if old == None:
-        return
-
-    old = f'"{old}"' if old != 'NULL' else old
-
-    with conn[name].cursor() as cur:
-        sql = f"update junctions set pattern = '{pattern}' where id = '{id}'"
-        cur.execute(sql)
-        redo = sql.replace("'", '"')
-        undo = f'update junctions set pattern = {old} where id = "{id}"'
-        add_operation(name, redo, undo)
-
-    change = ChangeSet()
-    change.update('junction', id, 'pattern', 'str', str(pattern))
-    return change
+    return _set_junction(name, id, 'pattern', 'str', pattern, True)
 
 
 def set_junction_coord(name: str, id: str, x: float, y: float) -> ChangeSet: