68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from .operation 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_cache(name: str, cs: ChangeSet) -> SqlChangeSet:
|
|
id = cs.operations[0]['id']
|
|
|
|
old = get_pattern(name, id)
|
|
new = { 'id': id, 'factors': [] }
|
|
|
|
f_id = f"'{id}'"
|
|
|
|
# TODO: transaction ?
|
|
redo_sql = f"delete from patterns where id = {f_id};"
|
|
redo_sql += f"\ndelete from _pattern where id = {f_id};"
|
|
if len(cs.operations[0]['factors']) > 0:
|
|
redo_sql += f"\ninsert into _pattern (id) values ({f_id});"
|
|
for factor in cs.operations[0]['factors']:
|
|
f_factor = float(factor)
|
|
redo_sql += f"\ninsert into patterns (id, factor) values ({f_id}, {f_factor});"
|
|
new['factors'].append(factor)
|
|
|
|
undo_sql = f"delete from patterns where id = {f_id};"
|
|
undo_sql += f"\ndelete from _pattern where id = {f_id};"
|
|
if len(old['factors']) > 0:
|
|
undo_sql += f"\ninsert 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 = { 'type': 'pattern' } | new
|
|
undo_cs = { 'type': 'pattern' } | old
|
|
|
|
return SqlChangeSet(redo_sql, undo_sql, redo_cs, undo_cs)
|
|
|
|
|
|
def set_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
result = set_pattern_cache(name, cs)
|
|
result.redo_cs |= g_update_prefix
|
|
result.undo_cs |= g_update_prefix
|
|
return execute_command(name, result)
|
|
|
|
|
|
def add_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
result = set_pattern_cache(name, cs)
|
|
result.redo_cs |= g_add_prefix
|
|
result.undo_cs |= g_delete_prefix
|
|
return execute_command(name, result)
|
|
|
|
|
|
def delete_pattern(name: str, cs: ChangeSet) -> ChangeSet:
|
|
cs.operations[0]['factors'] = []
|
|
result = set_pattern_cache(name, cs)
|
|
result.redo_cs |= g_delete_prefix
|
|
result.undo_cs |= g_add_prefix
|
|
return execute_command(name, result)
|