modelverse/kernel/test/functions/utils.py

import sys
sys.path.append("../state")
from modelverse_state.main import ModelverseState
import json

class MvSWrapper(object):
    def __init__(self):
        self.mvs = ModelverseState("../bootstrap/minimal.m.gz")

        self.mapping = {
                   "RD": self.mvs.read_dict,
                   "RR": self.mvs.read_root,
                   "RO": self.mvs.read_outgoing,
                   "RI": self.mvs.read_incoming,
                   "RE": self.mvs.read_edge,
                   "RDE": self.mvs.read_dict_edge,
                   "RDK": self.mvs.read_dict_keys,
                   "RDN": self.mvs.read_dict_node,
                   "RDNE": self.mvs.read_dict_node_edge,
                   "RRD": self.mvs.read_reverse_dict,
                   "RV": self.mvs.read_value,
                   "CN": self.mvs.create_node,
                   "CNV": self.mvs.create_nodevalue,
                   "CE": self.mvs.create_edge,
                   "DE": self.mvs.delete_edge,
                   "DN": self.mvs.delete_node,
                   "CD": self.mvs.create_dict,
                  }

    def execute(self, command, params):
        params = json.loads(json.dumps(params))
        retval = self.mapping[command](*params)
        retval = json.loads(json.dumps(retval))

        return [retval[0]]

def add_new_task(root, mvs, taskname):
    hierarchy = mvs.execute("RD", [root, "__hierarchy"])[0]
    initial = mvs.execute("RD", [hierarchy, "__IP"])[0]

    task_root = mvs.execute("CN", [])[0]
    _globals = mvs.execute("CN", [])[0]
    frame = mvs.execute("CN", [])[0]
    evalstack = mvs.execute("CN", [])[0]
    symbols = mvs.execute("CN", [])[0]
    _input = mvs.execute("CN", [])[0]
    _output = mvs.execute("CN", [])[0]
    returnvalue = mvs.execute("CN", [])[0]
    phase = mvs.execute("CNV", ["init"])[0]

    mvs.execute("CD", [root, taskname, task_root])
    mvs.execute("CD", [task_root, "frame", frame])
    mvs.execute("CD", [task_root, "globals", _globals])
    mvs.execute("CD", [task_root, "input", _input])
    mvs.execute("CD", [task_root, "last_input", _input])
    mvs.execute("CD", [task_root, "output", _output])
    mvs.execute("CD", [task_root, "last_output", _output])
    mvs.execute("CD", [frame, "evalstack", evalstack])
    mvs.execute("CD", [frame, "returnvalue", returnvalue])
    mvs.execute("CD", [frame, "phase", phase])
    mvs.execute("CD", [frame, "IP", initial])
    mvs.execute("CD", [frame, "symbols", symbols])

def execute_until_finished(mvk, mvs, operation="execute_rule", params=[]):
    taskname = "task_1"
    if operation == "initialize_new_task":
        return add_new_task(mvk.root, mvs, taskname)
    response = None
    while 1:
        mvs_commands = mvk.execute_yields(taskname, operation, params, response)
        if mvs_commands is None:
            break
        response = []
        for command, param in mvs_commands:
            response.append(mvs.execute(command, param)[0])

def get_inst(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    inst = mvs.execute("RD", [task_frame, "IP"])[0]
    return mvs.execute("RV", [inst])[0]["value"]

def get_phase(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    phase = mvs.execute("RD", [task_frame, "phase"])[0]
    return mvs.execute("RV", [phase])[0]

def get_returnvalue(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    returnvalue = mvs.execute("RD", [task_frame, "returnvalue"])[0]
    return mvs.execute("RV", [returnvalue])[0]

def get_inst_ref(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    return mvs.execute("RD", [task_frame, "IP"])[0]

def get_returnvalue_ref(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    return mvs.execute("RD", [task_frame, "returnvalue"])[0]

def get_phase_ref(root, mvs):
    task_root = mvs.execute("RD", [root, "task_1"])[0]
    task_frame = mvs.execute("RD", [task_root, "frame"])[0]
    phase = mvs.execute("RD", [task_frame, "phase"])[0]
    return phase

def read_primitive_interfaces(root, mvs):
    hierarchy = mvs.execute("RD", [root, "__hierarchy"])[0]
    primitives = mvs.execute("RD", [hierarchy, "primitives"])[0]
    keys = mvs.execute("RDK", [primitives])[0]
    d = {}
    for k in keys:
        value = mvs.execute("RDN", [primitives, k])[0]
        name = mvs.execute("RV", [k])[0]
        d[name] = value
    return d