155 lines
6.0 KiB
Python
155 lines
6.0 KiB
Python
from .operation import *
|
|
from .s2_junctions import *
|
|
|
|
def get_demand_schema(name: str) -> dict[str, dict[str, Any]]:
|
|
return { 'junction' : {'type': 'str' , 'optional': False , 'readonly': True },
|
|
'demands' : {'type': 'list' , 'optional': False , 'readonly': False,
|
|
'element': { 'demand' : {'type': 'float' , 'optional': False , 'readonly': False },
|
|
'pattern' : {'type': 'str' , 'optional': True , 'readonly': False },
|
|
'category': {'type': 'str' , 'optional': True , 'readonly': False }}}}
|
|
|
|
|
|
def get_demand(name: str, junction: str) -> dict[str, Any]:
|
|
des = read_all(name, f"select * from demands where junction = '{junction}' order by _order")
|
|
ds = []
|
|
for r in des:
|
|
d = {}
|
|
d['demand'] = float(r['demand'])
|
|
d['pattern'] = str(r['pattern']) if r['pattern'] != None else None
|
|
d['category'] = str(r['category']) if r['category'] != None else None
|
|
ds.append(d)
|
|
return { 'junction': junction, 'demands': ds }
|
|
|
|
|
|
def set_demand_cache(name: str, cs: ChangeSet) -> DbChangeSet:
|
|
junction = cs.operations[0]['junction']
|
|
old = get_demand(name, junction)
|
|
new = { 'junction': junction, 'demands': [] }
|
|
|
|
f_junction = f"'{junction}'"
|
|
|
|
# TODO: transaction ?
|
|
redo_sql = f"delete from demands where junction = {f_junction};"
|
|
for r in cs.operations[0]['demands']:
|
|
demand = float(r['demand'])
|
|
pattern = str(r['pattern']) if 'pattern' in r and r['pattern'] != None else None
|
|
category = str(r['category']) if 'category' in r and r['category'] != None else None
|
|
f_demand = demand
|
|
f_pattern = f"'{pattern}'" if pattern != None else 'null'
|
|
f_category = f"'{category}'" if category != None else 'null'
|
|
redo_sql += f"\ninsert into demands (junction, demand, pattern, category) values ({f_junction}, {f_demand}, {f_pattern}, {f_category});"
|
|
new['demands'].append({ 'demand': demand, 'pattern': pattern, 'category': category })
|
|
|
|
_undo_sql = f"delete from demands where junction = {f_junction};"
|
|
for r in old['demands']:
|
|
demand = float(r['demand'])
|
|
pattern = str(r['pattern']) if 'pattern' in r and r['pattern'] != None else None
|
|
category = str(r['category']) if 'category' in r and r['category'] != None else None
|
|
f_demand = demand
|
|
f_pattern = f"'{pattern}'" if pattern != None else 'null'
|
|
f_category = f"'{category}'" if category != None else 'null'
|
|
_undo_sql += f"\ninsert into demands (junction, demand, pattern, category) values ({f_junction}, {f_demand}, {f_pattern}, {f_category});"
|
|
|
|
redo_cs = []
|
|
redo_cs.append(g_update_prefix | { 'type': 'demand' } | new)
|
|
undo_cs = []
|
|
undo_cs.append(g_update_prefix | { 'type': 'demand' } | old)
|
|
|
|
cache = None
|
|
if len(cs.operations[0]['demands']) > 0:
|
|
r = cs.operations[0]['demands'][0]
|
|
demand = float(r['demand'])
|
|
pattern = str(r['pattern']) if 'pattern' in r and r['pattern'] != None else None
|
|
cache = set_junction_cache(name, ChangeSet({'id': junction, 'demand': demand, 'pattern': pattern}))
|
|
else:
|
|
cache = set_junction_cache(name, ChangeSet({'id': junction, 'demand': None, 'pattern': None}))
|
|
|
|
undo_sql = ''
|
|
if cache != None:
|
|
redo_sql += '\n'
|
|
redo_sql += cache.redo_sql
|
|
|
|
undo_sql += cache.undo_sql
|
|
undo_sql += '\n'
|
|
undo_sql += _undo_sql
|
|
|
|
redo_cs.append(cache.redo_cs)
|
|
|
|
undo_cs.append(cache.undo_cs)
|
|
undo_cs.reverse()
|
|
|
|
return DbChangeSet(redo_sql, undo_sql, redo_cs, undo_cs)
|
|
|
|
|
|
def set_demand(name: str, cs: ChangeSet) -> ChangeSet:
|
|
return execute_command(name, set_demand_cache(name, cs))
|
|
|
|
|
|
class InpDemand:
|
|
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.junction = str(tokens[0])
|
|
self.demand = float(tokens[1])
|
|
self.pattern = str(tokens[2]) if num_without_desc >= 3 else None
|
|
self.category = str(tokens[3]) if num_without_desc >= 4 else None
|
|
|
|
|
|
def inp_in_demand(section: list[str]) -> ChangeSet:
|
|
objs: dict[str, list[InpDemand]] = {}
|
|
for s in section:
|
|
# skip comment
|
|
if s.startswith(';'):
|
|
continue
|
|
obj = InpDemand(s)
|
|
if obj.junction not in objs:
|
|
objs[obj.junction] = []
|
|
objs[obj.junction].append(obj)
|
|
|
|
cs = ChangeSet()
|
|
for junction, demands in objs.items():
|
|
obj_cs : dict[str, Any] = g_update_prefix | {'type': 'demand', 'junction' : junction, 'demands' : []}
|
|
for obj in demands:
|
|
obj_cs['demands'].append({'demand': obj.demand, 'pattern' : obj.pattern, 'category': obj.category})
|
|
cs.append(obj_cs)
|
|
return cs
|
|
|
|
|
|
def fill_demand(junction_cs : ChangeSet, demand_cs : ChangeSet) -> ChangeSet:
|
|
cs = ChangeSet()
|
|
|
|
for j_cs in junction_cs.operations:
|
|
if 'demand' not in j_cs:
|
|
continue
|
|
|
|
in_demand = False
|
|
for d_cs in demand_cs.operations:
|
|
if j_cs['id'] == d_cs['junction']:
|
|
in_demand = True
|
|
break
|
|
|
|
if not in_demand:
|
|
obj_cs : dict[str, Any] = g_update_prefix | {'type': 'demand', 'junction' : j_cs['id'], 'demands' : []}
|
|
j_demand = j_cs['demand']
|
|
j_pattern = j_cs['pattern'] if 'pattern' in j_cs else None
|
|
obj_cs['demands'].append({'demand': j_demand, 'pattern' : j_pattern, 'category': None})
|
|
cs.append(obj_cs)
|
|
|
|
return cs
|
|
|
|
|
|
def inp_out_demand(name: str) -> list[str]:
|
|
lines = []
|
|
objs = read_all(name, f"select * from demands order by _order")
|
|
for obj in objs:
|
|
junction = obj['junction']
|
|
demand = obj['demand']
|
|
pattern = obj['pattern'] if obj['pattern'] != None else ''
|
|
category = f";{obj['category']}" if obj['category'] != None else ';'
|
|
lines.append(f'{junction} {demand} {pattern} {category}')
|
|
return lines
|