Support VD

api/s35_vd.py

@@ -0,0 +1,202 @@
+from .database import *
+from .s0_base import is_node
+from .s32_region_util import to_postgis_polygon
+from .s32_region import get_region
+
+def get_virtual_district_schema(name: str) -> dict[str, dict[str, Any]]:
+    return { 'id'         : {'type': 'str'        , 'optional': False , 'readonly': True  },
+             'boundary'   : {'type': 'tuple_list' , 'optional': False , 'readonly': False },
+             'center'     : {'type': 'str'        , 'optional': False , 'readonly': False } }
+
+def get_virtual_district(name: str, id: str) -> dict[str, Any]:
+    vd = get_region(name, id)
+    if vd == {}:
+        return {}
+    r = try_read(name, f"select * from region_vd where id = '{id}'")
+    if r == None:
+        return {}
+    vd['center'] = r['center']
+    vd['nodes'] = list(eval(r['nodes']))
+    return vd
+
+def _set_virtual_district(name: str, cs: ChangeSet) -> DbChangeSet:
+    id = cs.operations[0]['id']
+
+    new_boundary = cs.operations[0]['boundary']
+    old_boundary = get_region(name, id)['boundary']
+
+    new_center = cs.operations[0]['center']
+    f_new_center = f"'{new_center}'"
+
+    new_nodes = cs.operations[0]['nodes']
+    str_new_nodes = str(new_nodes).replace("'", "''")
+
+    old = get_virtual_district(name, id)
+    old_center = old['center']
+    f_old_center = f"'{old_center}'"
+
+    old_nodes = old['nodes']
+    str_old_nodes = str(old_nodes).replace("'", "''")
+
+    redo_sql = f"update region set boundary = st_geomfromtext('{to_postgis_polygon(new_boundary)}') where id = '{id}';"
+    redo_sql += f"update region_vd set center = {f_new_center}, nodes = '{str_new_nodes}' where id = '{id}';"
+
+    undo_sql = f"update region_vd set center = {f_old_center}, nodes = '{str_old_nodes}' where id = '{id}';"
+    undo_sql += f"update region set boundary = st_geomfromtext('{to_postgis_polygon(old_boundary)}') where id = '{id}';"
+
+    redo_cs = g_update_prefix | { 'type': 'virtual_district', 'id': id, 'boundary': new_boundary, 'center': new_center, 'nodes': new_nodes }
+    undo_cs = g_update_prefix | { 'type': 'virtual_district', 'id': id, 'boundary': old_boundary, 'center': old_center, 'nodes': old_nodes }
+
+    return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
+
+
+def set_virtual_district(name: str, cs: ChangeSet) -> ChangeSet:
+    ops = cs.operations
+
+    if len(cs.operations) == 0:
+        return ChangeSet()
+    
+    op = ops[0]
+
+    if 'id' not in op:
+        return ChangeSet()
+    
+    vd = get_virtual_district(name, op['id'])
+    if vd == {}:
+        return ChangeSet()
+
+    if 'boundary' not in op:
+        op['boundary'] = vd['boundary']
+    else:
+        b = op['boundary']
+        if len(b) < 4 or b[0] != b[-1]:
+            return ChangeSet()
+
+    if 'center' not in op:
+        op['center'] = vd['center']
+
+    if not is_node(name, op['center']):
+        return ChangeSet()
+
+    if 'nodes' not in op:
+        op['nodes'] = vd['nodes']
+    else:
+        for node in op['nodes']:
+            if not is_node(name, node):
+                return ChangeSet()
+
+    return execute_command(name, _set_virtual_district(name, cs))
+
+
+def _add_virtual_district(name: str, cs: ChangeSet) -> DbChangeSet:
+    id = cs.operations[0]['id']
+
+    boundary = cs.operations[0]['boundary']
+
+    center = cs.operations[0]['center']
+    f_center = f"'{center}'"
+
+    nodes = cs.operations[0]['nodes']
+    str_nodes = str(nodes).replace("'", "''")
+
+    redo_sql = f"insert into region (id, boundary, r_type) values ('{id}', '{to_postgis_polygon(boundary)}', 'VD');"
+    redo_sql += f"insert into region_vd (id, center, nodes) values ('{id}', {f_center}, '{str_nodes}');"
+
+    undo_sql = f"delete from region_vd where id = '{id}';"
+    undo_sql += f"delete from region where id = '{id}';"
+
+    redo_cs = g_add_prefix | { 'type': 'virtual_district', 'id': id, 'boundary': boundary, 'center': center, 'nodes': nodes }
+    undo_cs = g_delete_prefix | { 'type': 'virtual_district', 'id': id }
+
+    return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
+
+
+def add_virtual_district(name: str, cs: ChangeSet) -> ChangeSet:
+    ops = cs.operations
+
+    if len(cs.operations) == 0:
+        return ChangeSet()
+    
+    op = ops[0]
+
+    if 'id' not in op:
+        return ChangeSet()
+    
+    vd = get_virtual_district(name, op['id'])
+    if vd != {}:
+        return ChangeSet()
+
+    if 'boundary' not in op:
+        return ChangeSet()
+    else:
+        b = op['boundary']
+        if len(b) < 4 or b[0] != b[-1]:
+            return ChangeSet()
+
+    if 'center' not in op:
+        return ChangeSet()
+
+    if not is_node(name, op['center']):
+        return ChangeSet()
+
+    if 'nodes' not in op:
+        op['nodes'] = []
+    else:
+        for node in op['nodes']:
+            if not is_node(name, node):
+                return ChangeSet()
+
+    return execute_command(name, _add_virtual_district(name, cs))
+
+
+def _delete_virtual_district(name: str, cs: ChangeSet) -> DbChangeSet:
+    id = cs.operations[0]['id']
+    vd = get_virtual_district(name, id)
+    boundary = vd['boundary']
+    center = vd['center']
+    f_center = f"'{center}'"
+    nodes = vd['nodes']
+    str_nodes = str(nodes).replace("'", "''")
+
+    redo_sql = f"delete from region_vd where id = '{id}';"
+    redo_sql += f"delete from region where id = '{id}';"
+
+    undo_sql = f"insert into region (id, boundary, r_type) values ('{id}', '{to_postgis_polygon(boundary)}', 'VD');"
+    undo_sql += f"insert into region_vd (id, center, nodes) values ('{id}', {f_center}, '{str_nodes}');"
+
+    redo_cs = g_delete_prefix | { 'type': 'virtual_district', 'id': id }
+    undo_cs = g_add_prefix | { 'type': 'virtual_district', 'id': id, 'boundary': boundary, 'center': center, 'nodes': nodes }
+
+    return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
+
+
+def delete_virtual_district(name: str, cs: ChangeSet) -> ChangeSet:
+    ops = cs.operations
+
+    if len(cs.operations) == 0:
+        return ChangeSet()
+    
+    op = ops[0]
+
+    if 'id' not in op:
+        return ChangeSet()
+    
+    vd = get_virtual_district(name, op['id'])
+    if vd == {}:
+        return ChangeSet()
+
+    return execute_command(name, _delete_virtual_district(name, cs))
+
+
+def get_all_virtual_district_ids(name: str) -> list[str]:
+    ids = []
+    for row in read_all(name, f"select id from region_vd"):
+        ids.append(row['id'])
+    return ids
+
+
+def get_all_virtual_districts(name: str) -> list[dict[str, Any]]:
+    result = []
+    for id in get_all_virtual_district_ids(name):
+        result.append(get_virtual_district(name, id))
+    return result