217 lines
6.8 KiB
Python
217 lines
6.8 KiB
Python
from .database import *
|
|
|
|
|
|
def get_pattern_schema(name: str) -> dict[str, dict[str, Any]]:
|
|
return { 'id' : {'type': 'str' , 'optional': False , 'readonly': True },
|
|
'factors' : {'type': 'float_list' , 'optional': False , 'readonly': False } }
|
|
|
|
|
|
def get_pattern(name: str, id: str) -> dict[str, Any]:
|
|
pas = read_all(name, f"select * from patterns where id = '{id}' order by _order")
|
|
ps = []
|
|
for r in pas:
|
|
ps.append(float(r['factor']))
|
|
return { 'id': id, 'factors': ps }
|
|
|
|
|
|
def set_pattern_cmd(name: str, cs: ChangeSet) -> DbChangeSet:
|
|
id = cs.operations[0]['id']
|
|
f_id = f"'{id}'"
|
|
|
|
old = get_pattern(name, id)
|
|
|
|
new = { 'id': id }
|
|
if 'factors' in cs.operations[0]:
|
|
new['factors'] = cs.operations[0]['factors']
|
|
else:
|
|
new['factors'] = old['factors']
|
|
|
|
# TODO: transaction ?
|
|
redo_sql = f"delete from patterns where id = {f_id};"
|
|
for f_factor in new['factors']:
|
|
redo_sql += f"\ninsert into patterns (id, factor) values ({f_id}, {f_factor});"
|
|
|
|
undo_sql = f"delete from patterns where id = {f_id};"
|
|
for f_factor in old['factors']:
|
|
undo_sql += f"\ninsert into patterns (id, factor) values ({f_id}, {f_factor});"
|
|
|
|
redo_cs = g_update_prefix | { 'type': 'pattern' } | new
|
|
undo_cs = g_update_prefix | { 'type': 'pattern' } | old
|
|
|
|
return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
|
|
|
|
|
|
def set_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
return execute_command(name, set_pattern_cmd(name, cs))
|
|
|
|
|
|
def add_pattern_cmd(name: str, cs: ChangeSet) -> DbChangeSet:
|
|
id = cs.operations[0]['id']
|
|
f_id = f"'{id}'"
|
|
|
|
new = { 'id': id, 'factors': cs.operations[0]['factors'] }
|
|
|
|
# TODO: transaction ?
|
|
redo_sql = f"insert into _pattern (id) values ({f_id});"
|
|
for f_factor in new['factors']:
|
|
redo_sql += f"\ninsert into patterns (id, factor) values ({f_id}, {f_factor});"
|
|
|
|
undo_sql = f"delete from patterns where id = {f_id};"
|
|
undo_sql += f"\ndelete from _pattern where id = {f_id};"
|
|
|
|
redo_cs = g_add_prefix | { 'type': 'pattern' } | new
|
|
undo_cs = g_delete_prefix | { 'type': 'pattern' } | { 'id': id }
|
|
|
|
return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
|
|
|
|
|
|
def add_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
return execute_command(name, add_pattern_cmd(name, cs))
|
|
|
|
|
|
def delete_pattern_cmd(name: str, cs: ChangeSet) -> DbChangeSet:
|
|
id = cs.operations[0]['id']
|
|
f_id = f"'{id}'"
|
|
|
|
old = get_pattern(name, id)
|
|
|
|
redo_sql = f"delete from patterns where id = {f_id};"
|
|
redo_sql += f"\ndelete from _pattern where id = {f_id};"
|
|
|
|
# TODO: transaction ?
|
|
undo_sql = f"insert into _pattern (id) values ({f_id});"
|
|
for f_factor in old['factors']:
|
|
undo_sql += f"\ninsert into patterns (id, factor) values ({f_id}, {f_factor});"
|
|
|
|
redo_cs = g_delete_prefix | { 'type': 'pattern' } | { 'id': id }
|
|
undo_cs = g_add_prefix | { 'type': 'pattern' } | old
|
|
|
|
return DbChangeSet(redo_sql, undo_sql, [redo_cs], [undo_cs])
|
|
|
|
|
|
def delete_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
return execute_command(name, delete_pattern_cmd(name, cs))
|
|
|
|
|
|
#--------------------------------------------------------------
|
|
# [EPA2][IN][OUT]
|
|
# ;desc
|
|
# id mult1 mult2 .....
|
|
#--------------------------------------------------------------
|
|
def inp_in_pattern(section: list[str]) -> ChangeSet:
|
|
descs = {}
|
|
patterns: dict[str, list[float]] = {}
|
|
|
|
count = len(section)
|
|
for i in range(0, count):
|
|
if section[i].startswith(';'):
|
|
# this is description
|
|
next = i + 1
|
|
if next < count and section[next].startswith(';') == False:
|
|
next_tokens = section[next].split()
|
|
descs[next_tokens[0]] = section[i].removeprefix(';')
|
|
continue
|
|
|
|
tokens = section[i].split()
|
|
if tokens[0] not in patterns:
|
|
patterns[tokens[0]] = []
|
|
for token in tokens[1:]:
|
|
patterns[tokens[0]].append(float(token))
|
|
|
|
cs = ChangeSet()
|
|
for id, factors in patterns.items():
|
|
cs.append(g_add_prefix | {'type': 'pattern', 'id' : id, 'factors' : factors})
|
|
|
|
#print(descs)
|
|
return cs
|
|
|
|
|
|
def inp_in_pattern_new(name: str, section: list[str]) -> None:
|
|
descs = {}
|
|
patterns: list[str] = []
|
|
|
|
count = len(section)
|
|
for i in range(0, count):
|
|
if section[i].startswith(';'):
|
|
# this is description
|
|
next = i + 1
|
|
if next < count and section[next].startswith(';') == False:
|
|
next_tokens = section[next].split()
|
|
descs[next_tokens[0]] = section[i].removeprefix(';')
|
|
continue
|
|
|
|
tokens = section[i].split()
|
|
if tokens[0] not in patterns:
|
|
patterns.append(tokens[0])
|
|
write(name, f"insert into _pattern (id) values ('{tokens[0]}');")
|
|
for token in tokens[1:]:
|
|
write(name, f"insert into patterns (id, factor) values ('{tokens[0]}', {float(token)});")
|
|
|
|
|
|
def inp_out_pattern(name: str) -> list[str]:
|
|
lines = []
|
|
objs = read_all(name, f"select * from patterns order by _order")
|
|
for obj in objs:
|
|
id = obj['id']
|
|
factor = obj['factor']
|
|
lines.append(f'{id} {factor}')
|
|
return lines
|
|
|
|
|
|
#--------------------------------------------------------------
|
|
# [EPA3][IN][OUT]
|
|
# id FIXED (interval)
|
|
# id factor1 factor2 ...
|
|
# id VARIABLE
|
|
# id time1 factor1 time2 factor2 ...
|
|
#--------------------------------------------------------------
|
|
def inp_in_pattern_v3(section: list[str]) -> ChangeSet:
|
|
patterns: dict[str, list[float]] = {}
|
|
|
|
variable_patterns: list[str] = []
|
|
|
|
count = len(section)
|
|
for i in range(0, count):
|
|
if section[i].startswith(';'):
|
|
continue
|
|
|
|
tokens = section[i].split()
|
|
|
|
# for EPA3, ignore time of variable pattern...
|
|
if tokens[1] == 'VARIABLE':
|
|
variable_patterns.append(tokens[0])
|
|
continue
|
|
elif tokens[1] == 'FIXED':
|
|
continue
|
|
|
|
if tokens[0] not in patterns:
|
|
patterns[tokens[0]] = []
|
|
if tokens[0] not in variable_patterns:
|
|
for token in tokens[1:]:
|
|
patterns[tokens[0]].append(float(token))
|
|
else:
|
|
for token in tokens[1::2]:
|
|
patterns[tokens[0]].append(float(token))
|
|
|
|
cs = ChangeSet()
|
|
for id, factors in patterns.items():
|
|
cs.append(g_add_prefix | {'type': 'pattern', 'id' : id, 'factors' : factors})
|
|
|
|
#print(descs)
|
|
return cs
|
|
|
|
|
|
def inp_out_pattern_v3(name: str) -> list[str]:
|
|
lines = []
|
|
objs = read_all(name, f"select * from patterns order by _order")
|
|
ids = []
|
|
for obj in objs:
|
|
id = obj['id']
|
|
if id not in ids:
|
|
# for EPA3, ignore time of variable pattern...
|
|
lines.append(f'{id} FIXED')
|
|
ids.append(id)
|
|
factor = obj['factor']
|
|
lines.append(f'{id} {factor}')
|
|
return lines
|