97 lines
4.2 KiB
Python
97 lines
4.2 KiB
Python
from ..core.database import (
|
|
ChangeSet,
|
|
DatabaseCommand,
|
|
execute_command,
|
|
g_update_prefix,
|
|
read_all,
|
|
sql_literal,
|
|
try_read,
|
|
)
|
|
from typing import Any
|
|
|
|
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, "select base_demand as demand, pattern_id as pattern, category from network.demands where junction_id = %s order by sequence_no", (junction,))
|
|
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(name: str, cs: ChangeSet) -> DatabaseCommand:
|
|
junction = cs.operations[0]['junction']
|
|
new = { 'junction': junction, 'demands': [] }
|
|
|
|
f_junction = sql_literal(junction)
|
|
|
|
statement = f"delete from network.demands where junction_id = {f_junction};"
|
|
for sequence_no, r in enumerate(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 = sql_literal(demand)
|
|
f_pattern = sql_literal(pattern)
|
|
f_category = sql_literal(category)
|
|
statement += f"\ninsert into network.demands (junction_id, sequence_no, base_demand, pattern_id, category) values ({f_junction}, {sequence_no}, {f_demand}, {f_pattern}, {f_category});"
|
|
new['demands'].append({ 'demand': demand, 'pattern': pattern, 'category': category })
|
|
|
|
change = g_update_prefix | { 'type': 'demand' } | new
|
|
|
|
return DatabaseCommand(statement, [change])
|
|
|
|
|
|
def set_demand(name: str, cs: ChangeSet) -> ChangeSet:
|
|
return execute_command(name, _set_demand(name, cs))
|
|
|
|
|
|
#--------------------------------------------------------------
|
|
# [EPA2][EPA3][IN][OUT]
|
|
# node base_demand (pattern) ;category
|
|
#--------------------------------------------------------------
|
|
|
|
|
|
def inp_in_demand(line: str) -> str:
|
|
tokens = line.split()
|
|
|
|
num = len(tokens)
|
|
has_desc = tokens[-1].startswith(';')
|
|
num_without_desc = (num - 1) if has_desc else num
|
|
|
|
junction = str(tokens[0])
|
|
demand = float(tokens[1])
|
|
pattern = str(tokens[2]) if num_without_desc >= 3 else None
|
|
category = str(tokens[3]) if num_without_desc >= 4 else None
|
|
|
|
junction_sql = sql_literal(junction)
|
|
return str(f"insert into network.demands (junction_id, sequence_no, base_demand, pattern_id, category) values ({junction_sql}, (select coalesce(max(sequence_no) + 1, 0) from network.demands where junction_id = {junction_sql}), {sql_literal(demand)}, {sql_literal(pattern)}, {sql_literal(category)});")
|
|
|
|
|
|
def inp_out_demand(name: str) -> list[str]:
|
|
lines = []
|
|
objs = read_all(name, "select junction_id as junction, base_demand as demand, pattern_id as pattern, category from network.demands order by junction_id, sequence_no")
|
|
for obj in objs:
|
|
junction = obj['junction']
|
|
demand = obj['demand']
|
|
pattern = obj['pattern'] if obj['pattern'] is not None else ''
|
|
category = f";{obj['category']}" if obj['category'] is not None else ';'
|
|
lines.append(f'{junction} {demand} {pattern} {category}')
|
|
return lines
|
|
|
|
|
|
def delete_demand_by_junction(name: str, junction: str) -> ChangeSet:
|
|
row = try_read(name, "select 1 from network.demands where junction_id = %s", (junction,))
|
|
if row is None:
|
|
return ChangeSet()
|
|
return ChangeSet(g_update_prefix | {'type': 'demand', 'junction': junction, 'demands': []})
|