12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064 |
- import urllib
- import json
- import random
- import sys
- import time
- import threading
- import uuid
- from sccd.runtime.statecharts_core import Event
- COMPILER_PATH = "interface/HUTN"
- MODE_UNCONNECTED = 0
- MODE_UNAUTHORIZED = 1
- MODE_MODELLING = 2
- MODE_MODIFY = 3
- MODE_DIALOG = 4
- MODE_MANUAL = 5
- MODE_SERVICE = 6
- # Bind to the compiler (might have to update path manually!)
- sys.path.append(COMPILER_PATH)
- from hutn_compiler.compiler import main as do_compile
- import socket2event
- # Exceptions
- class ModelverseException(Exception):
- pass
- class UnknownError(ModelverseException):
- pass
- class UnknownIdentifier(ModelverseException):
- pass
- class CompilationError(ModelverseException):
- pass
- class NoSuchAttribute(ModelverseException):
- pass
- class UnknownModel(ModelverseException):
- pass
- class ConnectionError(ModelverseException):
- pass
- class ModelExists(ModelverseException):
- pass
- class PermissionDenied(ModelverseException):
- pass
- class InvalidMode(ModelverseException):
- pass
- class InterfaceMismatch(ModelverseException):
- pass
- class UnknownMetamodellingHierarchy(ModelverseException):
- pass
- # Helper functions and configuration: do not use yourself!
- taskname = None
- mode = MODE_UNCONNECTED
- prev_mode = None
- current_model = None
- registered_metamodels = {}
- outputs = [None]
- ctrl_input = None
- ctrl_output = None
- def _output_thread(outputs, taskname):
- req_out = ctrl_output.addOutputListener("request_out")
- my_id = str(uuid.uuid4())
- try:
- while 1:
- ctrl_output.addInput(Event("HTTP_input", "request_in", [urllib.urlencode({"op": "get_output", "taskname": taskname}), my_id]))
- event = req_out.fetch(-1)
- if event.parameters[1] == my_id:
- outputs.append(json.loads(event.parameters[0]))
- except:
- pass
- def _exec_on_statechart(statechart):
- def _exec_sc(controller, inport, outport):
- op = controller.addOutputListener(outport)
- while 1:
- if len(outputs) > 1:
- # Is an output message of the Mv, so put it in the SC
- del outputs[0]
- output_event = outputs[0]
- if output_event == "Success" or output_event == "Failure":
- # Is a stop event!
- controller.addInput(Event("terminate", inport, []))
- break
- else:
- # Is just a normal event!
- controller.addInput(Event("input", inport, [output_event]))
- input_event = op.fetch(0)
- if input_event is not None:
- # Expand the event and make it HTTP input
- _input(input_event.parameters)
-
- time.sleep(0.02)
-
- thrd = threading.Thread(target=_exec_sc, args=statechart)
- thrd.daemon = True
- thrd.start()
- return None
- def _get_metamodel(model):
- global registered_metamodels
- try:
- return registered_metamodels[model]
- except KeyError:
- raise UnknownMetamodellingHierarchy(model)
- def _check_type(value):
- if not isinstance(value, (int, long, float, str, unicode, bool)):
- raise UnsupportedValue("%s : %s" % (value, str(type(value))))
- def _check_type_list(value):
- if isinstance(value, list):
- [_check_type(i) for i in value]
- else:
- _check_type(value)
- def _dict_to_list(python_dict):
- lst = []
- for k, v in python_dict.items():
- lst += [k, v]
- return lst
- def _goto_mode(new_mode, model_name=None):
- global mode
- if mode == MODE_MANUAL and new_mode == MODE_MODIFY:
- if model_name != None and current_model != model_name:
- raise InvalidMode("Mode error: cannot modify other models!")
- else:
- return
- elif mode == MODE_MODELLING and new_mode == MODE_MODIFY:
- # Are in root view, but want to modify a model
- _model_modify(model_name, _get_metamodel(model_name))
- elif mode == MODE_MODIFY and new_mode == MODE_MODIFY and model_name != None and current_model != model_name:
- # Are in modify mode, but want to modify a different model
- mm = _get_metamodel(model_name)
- _model_exit()
- _model_modify(model_name, mm)
- elif mode == MODE_MODIFY and new_mode == MODE_MODELLING:
- _model_exit()
- elif mode == new_mode:
- return
- else:
- # Go to a mode that we have no automatic transfer to: raise exception
- raise InvalidMode("Required mode: %s, current mode: %s" % (new_mode, mode))
- def _input(value, port=None):
- # Ugly json encoding of primitives
- #print("[IN] %s" % value)
- if port is None:
- port = taskname
- if isinstance(value, type([])):
- value = json.dumps(value)
- ctrl_input.addInput(Event("HTTP_input", "request_in", [urllib.urlencode({"op": "set_input", "taskname": port, "data": value}), None]))
- else:
- value = json.dumps(value)
- ctrl_input.addInput(Event("HTTP_input", "request_in", [urllib.urlencode({"op": "set_input", "taskname": port, "value": value}), None]))
- def _input_raw(value, taskname):
- # Ugly json encoding of primitives
- ctrl_input.addInput(Event("HTTP_input", "request_in", [urllib.urlencode({"op": "set_input", "taskname": taskname, "value": value}), None]))
- def _compile_AL(code):
- # Compile an action language file and send the compiled code
- code_fragments = code.split("\n")
- code_fragments = [i for i in code_fragments if i.strip() != ""]
- code_fragments = [i.replace(" ", "\t") for i in code_fragments]
- initial_tabs = min([len(i) - len(i.lstrip("\t")) for i in code_fragments])
- code_fragments = [i[initial_tabs:] for i in code_fragments]
- code_fragments.append("")
- code = "\n".join(code_fragments)
- with open(".code.alc", "w") as f:
- f.write(code)
- f.flush()
- compiled = do_compile(".code.alc", COMPILER_PATH + "/grammars/actionlanguage.g", "CS")
- return compiled
- def _compile_model(code):
- # Compile a model and send the compiled graph
- # First change multiple spaces to a tab
- code_fragments = code.split("\n")
- code_fragments = [i for i in code_fragments if i.strip() != ""]
- code_fragments = [i.replace(" ", "\t") for i in code_fragments]
- initial_tabs = min([len(i) - len(i.lstrip("\t")) for i in code_fragments])
- code_fragments = [i[initial_tabs:] for i in code_fragments]
- code_fragments.append("")
- code = "\n".join(code_fragments)
- with open(".model.mvc", "w") as f:
- f.write(code)
- f.flush()
- return do_compile(".model.mvc", COMPILER_PATH + "/grammars/modelling.g", "M")
- def _output(expected=None):
- try:
- while len(outputs) < 2:
- time.sleep(0.02)
- del outputs[0]
- #print("[OUT] %s" % outputs[0])
- except:
- raise UnknownError()
- if expected is not None and _last_output() != expected:
- raise InterfaceMismatch(_last_output(), expected)
- return _last_output()
- def _last_output():
- return outputs[0]
- # Raise common exceptions
- def _handle_output(requested=None, split=False):
- value = _output()
- if value.startswith("Model exists: "):
- raise ModelExists(value.split(": ", 1)[1])
- elif value.startswith("Permission denied"):
- raise PermissionDenied(value.split(": ", 1)[1])
- elif value.startswith("Model not found: "):
- raise UnknownModel(value.split(": ", 1)[1])
- elif value.startswith("Element not found: "):
- raise UnknownIdentifier(value.split(": ", 1)[1])
- elif value.startswith("Element exists: "):
- raise ElementExists(value.split(": ", 1)[1])
- elif value.startswith("Attribute not found: "):
- raise NoSuchAttribute(value.split(": ", 1)[1])
- elif requested is not None and value.startswith(requested):
- if not split:
- return value
- else:
- splitted = value.strip().split(" ", 1)
- if len(splitted) > 1:
- return splitted[1].split("\n")
- else:
- return []
- else:
- raise InterfaceMismatch(value)
-
- def _model_modify(model_name, metamodel_name):
- """Modify an existing model."""
- global mode
- global prev_mode
- if mode == MODE_MANUAL:
- prev_mode = MODE_MANUAL
- mode = MODE_MODIFY
- return None
- _goto_mode(MODE_MODELLING)
- prev_mode = MODE_MODELLING
- _input(["model_modify", model_name, metamodel_name])
- _handle_output("Success")
- global current_model
- current_model = model_name
- # Mode has changed
- mode = MODE_MODIFY
- _output("Model loaded, ready for commands!")
- def _model_exit():
- """Leave model modify mode."""
- global mode
- global prev_mode
- if prev_mode == MODE_MANUAL:
- mode = MODE_MANUAL
- return
- if mode != MODE_MODIFY:
- raise InvalidMode()
- _input("exit")
- _output("Success")
- mode = MODE_MODELLING
- def alter_context(model_name, metamodel_name):
- global registered_metamodels
- registered_metamodels[model_name] = metamodel_name
- # Main MvC operations
- def init(address_param="127.0.0.1:8001", timeout=20.0):
- """Starts up the connection to the Modelverse."""
- # Start up the HTTP Client SC
- import http_client
- global ctrl_input
- global ctrl_output
- if ctrl_input is not None:
- ctrl_input.stop()
- if ctrl_output is not None:
- ctrl_output.stop()
- ctrl_input = http_client.Controller()
- ctrl_output = http_client.Controller()
- controllers = [ctrl_input, ctrl_output]
- addr, port = address_param.split(":", 1)
- port = int(port)
- for ctrl in controllers:
- listener = ctrl.addOutputListener("request_out")
- socket2event.boot_translation_service(ctrl)
- thrd = threading.Thread(target=ctrl.start)
- thrd.daemon = True
- thrd.start()
- evt = listener.fetch(-1)
- if evt.name != "http_client_initialized":
- raise Exception("HTTP client did not behave as expected during init: " + str(evt.name))
- ctrl.addInput(Event("connect", "request_in", [(addr, port), timeout]))
- evt = listener.fetch(-1)
- if evt.name == "http_client_timeout":
- raise Exception("HTTP client timeout")
- if evt.name != "http_client_ready":
- raise Exception("HTTP client did not behave as expected during connect: " + str(evt.name))
- global mode
- start_time = time.time()
- task = random.random()
- _input_raw('"%s"' % task, "task_manager")
- mode = MODE_UNAUTHORIZED
- global outputs
- global taskname
- outputs = [None]
- taskname = task
- # This re-assign also diconnects the previous get_output connections to the outputs variable
- thrd = threading.Thread(target=_output_thread, args=[outputs, task])
- thrd.daemon = True
- thrd.start()
- def login(username, password):
- """Log in a user, if user doesn't exist, it is created."""
- global mode
- _goto_mode(MODE_UNAUTHORIZED)
- _output("Log on as which user?")
- _input(username)
- if _output() == "Password for existing user?":
- _input(password)
- if _output() == "Welcome to the Model Management Interface v2.0!":
- _output("Use the 'help' command for a list of possible commands")
- _input("quiet")
- mode = MODE_MODELLING
- elif _last_output() == "Wrong password!":
- raise PermissionDenied()
- else:
- raise InterfaceMismatch(_last_output())
- elif _last_output() == "This is a new user: please give password!":
- _input(password)
- _output("Please repeat the password")
- _input(password)
- if _output() == "Passwords match!":
- _output("Welcome to the Model Management Interface v2.0!")
- _output("Use the 'help' command for a list of possible commands")
- _input("quiet")
- mode = MODE_MODELLING
- elif _last_output() == "Not the same password!":
- # We just sent the same password, so it should be identical, unless the interface changed
- raise InterfaceMismatch(_last_output())
- else:
- raise InterfaceMismatch(_last_output())
- else:
- raise InterfaceMismatch(_last_output())
- def model_add(model_name, metamodel_name, model_code=None):
- """Instantiate a new model."""
- _goto_mode(MODE_MODELLING)
- # Do this before creating the model, as otherwise compilation errors would make us inconsistent
- if model_code is not None:
- try:
- compiled = _compile_model(model_code)
- except Exception as e:
- raise CompilationError(e)
- else:
- compiled = [0]
- _input(["model_add", metamodel_name, model_name])
- _handle_output("Waiting for model constructors...")
- _input(compiled)
- _output("Success")
- global registered_metamodels
- registered_metamodels[model_name] = metamodel_name
- def upload_code(code):
- try:
- compiled = _compile_AL(code)
- except Exception as e:
- raise CompilationError(e)
- _input(compiled)
- def model_delete(model_name):
- """Delete an existing model."""
- _goto_mode(MODE_MODELLING)
- _input(["model_delete", model_name])
- _handle_output("Success")
- def model_list(location):
- """List all models."""
- _goto_mode(MODE_MODELLING)
- _input(["model_list", location])
- return set(_handle_output("Success: ", split=True))
- def model_list_full(location):
- """List full information on all models."""
- _goto_mode(MODE_MODELLING)
- _input(["model_list_full", location])
- output = _handle_output("Success: ", split=True)
- lst = set([])
- for v in output:
- m = v.strip()
- perm, own, grp, m = m.split(" ", 3)
- lst.add((m, own, grp, perm))
- return lst
- def verify(model_name, metamodel_name=None):
- """Verify if a model conforms to its metamodel."""
- _goto_mode(MODE_MODELLING)
- if metamodel_name is None:
- metamodel_name = _get_metamodel(model_name)
- _input(["verify", model_name, metamodel_name])
- return _handle_output("Success: ", split=True)[0]
- def model_overwrite(model_name, new_model=None, metamodel_name=None):
- """Upload a new model and overwrite an existing model."""
- _goto_mode(MODE_MODIFY, model_name)
- if new_model is not None:
- try:
- compiled = _compile_model(new_model)
- except Exception as e:
- raise CompilationError(e)
- else:
- compiled = [0]
- _input("upload")
- _handle_output("Waiting for model constructors...")
- _input(compiled)
- _output("Success")
- if metamodel_name is not None:
- global registered_metamodels
- registered_metamodels[model_name] = metamodel_name
- def user_logout():
- """Log out the current user and break the connection."""
- global mode
- _goto_mode(MODE_MODELLING)
- _input("exit")
- mode = MODE_UNCONNECTED
- def user_delete():
- """Removes the current user and break the connection."""
- global mode
- _goto_mode(MODE_MODELLING)
- _input("self-destruct")
- mode = MODE_UNCONNECTED
- def model_render(model_name, mapper_name):
- """Fetch a rendered verion of a model."""
- _goto_mode(MODE_MODELLING)
- _input(["model_render", model_name, mapper_name])
- return json.loads(_handle_output("Success: ", split=True)[0])
- def transformation_between(source, target):
- _goto_mode(MODE_MODELLING)
- _input(["transformation_between", source, target])
- output = _handle_output("Success: ", split=True)
- return set(output)
- def transformation_add_MT(source_metamodels, target_metamodels, operation_name, code, callback=lambda: None):
- """Create a new model transformation."""
- global mode
- _goto_mode(MODE_MODELLING)
- import time
- start = time.time()
- try:
- compiled = _compile_model(code)
- except Exception as e:
- raise CompilationError(e)
- #print("Compilation took: %ss" % (time.time() - start))
- start = time.time()
- mv_dict_rep = _dict_to_list(source_metamodels) + [""] + _dict_to_list(target_metamodels) + [""]
- _input(["transformation_add_MT"] + mv_dict_rep + [operation_name])
- # Possibly modify the merged metamodel first (add tracability links)
- if len(source_metamodels) + len(target_metamodels) > 0:
- mode = MODE_MANUAL
- _output("Model loaded, ready for commands!")
- callback()
- _input("exit")
- mode = MODE_MODELLING
- #print("Callbacks took: %ss" % (time.time() - start))
- start = time.time()
- # Done, so RAMify and upload the model
- _handle_output("Waiting for model constructors...")
- _input(compiled)
- _handle_output("Success")
- #print("Upload and RAMify took: %ss" % (time.time() - start))
- def transformation_add_AL(source_metamodels, target_metamodels, operation_name, code, callback=lambda: None):
- """Create a new action language model, which can be executed."""
- global mode
- _goto_mode(MODE_MODELLING)
- try:
- compiled = _compile_AL(code)
- except Exception as e:
- raise CompilationError(e)
- mv_dict_rep = _dict_to_list(source_metamodels) + [""] + _dict_to_list(target_metamodels) + [""]
- _input(["transformation_add_AL"] + mv_dict_rep + [operation_name])
- # Possibly modify the merged metamodel first (add tracability links)
- if len(source_metamodels) + len(target_metamodels) > 0:
- mode = MODE_MANUAL
- _output("Model loaded, ready for commands!")
- callback()
- _input("exit")
- mode = MODE_MODELLING
- _handle_output("Waiting for code constructors...")
- _input(compiled)
- _output("Success")
- def transformation_add_MANUAL(source_metamodels, target_metamodels, operation_name, callback=lambda: None):
- """Create a new manual model operation."""
- global mode
- _goto_mode(MODE_MODELLING)
- mv_dict_rep = _dict_to_list(source_metamodels) + [""] + _dict_to_list(target_metamodels) + [""]
- _input(["transformation_add_MANUAL"] + mv_dict_rep + [operation_name])
- # Possibly modify the merged metamodel first (add tracability links)
- if len(source_metamodels) + len(target_metamodels) > 0:
- mode = MODE_MANUAL
- _output("Model loaded, ready for commands!")
- callback()
- _input("exit")
- mode = MODE_MODELLING
- _handle_output("Success")
- def transformation_execute_AL(operation_name, input_models_dict, output_models_dict, statechart=None):
- """Execute an existing model operation."""
- global mode
- _goto_mode(MODE_MODELLING)
- mv_dict_rep = _dict_to_list(input_models_dict) + [""] + _dict_to_list(output_models_dict) + [""]
- _input(["transformation_execute", operation_name] + mv_dict_rep)
- _handle_output("Success: ready for AL execution")
- if statechart is not None:
- # We are to delegate this information to another statechart, which wants interaction
- # As such, we continue on a different thread, where we pipe the information, and let this call return immediately
- mode = MODE_DIALOG
- _exec_on_statechart(statechart)
- mode = MODE_MODELLING
- return None
- else:
- # No statechart associated, so just wait until we are finished
- while _output() not in ["Success", "Failure"]:
- pass
- if _last_output() == "Success":
- return True
- else:
- return False
- def transformation_execute_MANUAL(operation_name, input_models_dict, output_models_dict, callback=lambda i: None):
- """Execute an existing model operation."""
- global mode
- _goto_mode(MODE_MODELLING)
- mv_dict_rep = _dict_to_list(input_models_dict) + [""] + _dict_to_list(output_models_dict) + [""]
- _input(["transformation_execute", operation_name] + mv_dict_rep)
- _handle_output("Success: ready for MANUAL execution")
- # Skip over the begin of mini_modify
- _handle_output("Please perform manual operation ")
- _output("Model loaded, ready for commands!")
- # We are now executing, so everything we get is part of the dialog, except if it is the string for transformation termination
- mode = MODE_MANUAL
- callback()
- # Finished, so leave
- _input("exit")
- mode = MODE_MODELLING
- # Got termination message, so we are done!
- if _output() == "Success":
- return True
- else:
- return False
- def transformation_execute_MT(operation_name, input_models_dict, output_models_dict, statechart=None):
- """Execute an existing model operation."""
- global mode
- _goto_mode(MODE_MODELLING)
- mv_dict_rep = _dict_to_list(input_models_dict) + [""] + _dict_to_list(output_models_dict) + [""]
- _input(["transformation_execute", operation_name] + mv_dict_rep)
- _handle_output("Success: ready for MT execution")
- if statechart is not None:
- # We are to delegate this information to another statechart, which wants interaction
- # As such, we continue on a different thread, where we pipe the information, and let this call return immediately
- mode = MODE_DIALOG
- _exec_on_statechart(statechart)
- mode = MODE_MODELLING
- return None
- else:
- # No statechart associated, so just wait until we are finished
- while _output() not in ["Success", "Failure"]:
- pass
- if _last_output() == "Success":
- return True
- else:
- return False
- def transformation_list():
- """List existing model operations."""
- _goto_mode(MODE_MODELLING)
- _input("transformation_list")
- output = _handle_output("Success: ", split=True)
- lst = set([])
- for v in output:
- t, m = v.strip().split(" ", 1)
- t = t[1:-1].strip()
- m = m.strip().split(":")[0].strip()
- lst.add((t, m))
- return lst
- def process_execute(process_name, prefix, callbacks):
- """Execute a process model."""
- global mode
- _goto_mode(MODE_MODELLING)
- _input(["process_execute", process_name, prefix])
- _handle_output("Success")
- reuse = False
- while reuse or _output() != "Success":
- reuse = False
- output = _last_output()
- if output.startswith("Enacting "):
- # Next activity!
- t = output.split(" ", 1)[1].split(":", 1)[0]
- name = output.split(": ", 1)[1]
- if name in callbacks:
- callback = callbacks[name]
- if t == "ModelTransformation" or t == "ActionLanguage":
- while not (_output().startswith("Enacting ") or _last_output() == "Success"):
- mode = MODE_DIALOG
- reply = callback(_last_output())
- mode = MODE_MODELLING
- if reply is not None:
- _input(reply)
- if _last_output().startswith("Enacting "):
- reuse = True
- elif t == "ManualOperation":
- _handle_output("Please perform manual operation ")
- _output("Model loaded, ready for commands!")
- mode = MODE_MANUAL
- callback()
- _input("exit")
- mode = MODE_MODELLING
- def permission_modify(model_name, permissions):
- """Modify permissions of a model."""
- _goto_mode(MODE_MODELLING)
- _input(["permission_modify", model_name, permissions])
- _handle_output("Success")
- def permission_owner(model_name, owner):
- """Modify the owning user of a model."""
- _goto_mode(MODE_MODELLING)
- _input(["permission_owner", model_name, owner])
- _handle_output("Success")
- def permission_group(model_name, group):
- """Modify the owning group of a model."""
- _goto_mode(MODE_MODELLING)
- _input(["permission_group", model_name, group])
- _handle_output("Success")
- def group_create(group_name):
- """Create a new group."""
- _goto_mode(MODE_MODELLING)
- _input(["group_create", group_name])
- _handle_output("Success")
- def group_delete(group_name):
- """Delete a group of which you are an owner."""
- _goto_mode(MODE_MODELLING)
- _input(["group_delete", group_name])
- _handle_output("Success")
- def group_owner_add(group_name, user_name):
- """Add a new owning user to a group you own."""
- _goto_mode(MODE_MODELLING)
- _input(["owner_add", group_name, user_name])
- _handle_output("Success")
- def group_owner_delete(group_name, user_name):
- """Delete an owning user to a group you own."""
- _goto_mode(MODE_MODELLING)
- _input(["owner_delete", group_name, user_name])
- _handle_output("Success")
- def group_join(group_name, user_name):
- """Add a new user to a group you own."""
- _goto_mode(MODE_MODELLING)
- _input(["group_join", group_name, user_name])
- _handle_output("Success")
- def group_kick(group_name, user_name):
- """Delete a user from a group you own."""
- _goto_mode(MODE_MODELLING)
- _input(["group_kick", group_name, user_name])
- _handle_output("Success")
- def group_list():
- """List existing groups."""
- _goto_mode(MODE_MODELLING)
- _input(["group_list"])
- _handle_output("Success")
- def admin_promote(user_name):
- """Promote a user to admin status."""
- _goto_mode(MODE_MODELLING)
- _input(["admin_promote", user_name])
- _handle_output("Success")
- def admin_demote():
- """Demote a user from admin status."""
- _goto_mode(MODE_MODELLING)
- _input(["admin_demote", user_name])
- _handle_output("Success")
- # Actual operations on the model
- def element_list(model_name):
- """Return a list of all IDs and the type of the element"""
- _goto_mode(MODE_MODIFY, model_name)
- _input("list_full")
- lst = set([])
- output = _handle_output("Success: ", split=True)
- for v in output:
- m, mm = v.split(":")
- m = m.strip()
- mm = mm.strip()
- lst.add((m, mm))
- return lst
- def types(model_name):
- """Return a list of all types usable in the model"""
- _goto_mode(MODE_MODIFY, model_name)
- _input("types")
- lst = set([])
- output = _handle_output("Success: ", split=True)
- for v in output:
- m, mm = v.split(":")
- m = m.strip()
- lst.add(m)
- return lst
- def types_full(model_name):
- """Return a list of full types usable in the model"""
- _goto_mode(MODE_MODIFY, model_name)
- _input("types")
- lst = set([])
- output = _handle_output("Success: ", split=True)
- for v in output:
- m, mm = v.split(":")
- m = m.strip()
- mm = mm.strip()
- lst.add((m, mm))
- return lst
- def read(model_name, ID):
- """Return a tuple of information on the element: its type and source/target (None if not an edge)"""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read", ID])
- v = _handle_output("Success: ", split=True)
- t = v[1].split(":")[1].strip()
- if (not v[2].startswith("Source:")):
- rval = (t, None)
- else:
- src = v[2].split(":")[1].strip()
- trg = v[3].split(":")[1].strip()
- rval = (t, (src, trg))
- return rval
- def read_attrs(model_name, ID):
- """Return a dictionary of attribute value pairs"""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read", ID])
- output = _handle_output("Success: ", split=True)
- searching = True
- rval = {}
- for r in output:
- if searching:
- if r == "Attributes:":
- # Start working on attributes
- searching = False
- else:
- key, value = r.split(":", 1)
- _, value = value.split("=", 1)
- key = json.loads(key.strip())
- value = value.strip()
- if value == "None":
- value = None
- elif value == "True":
- value = True
- elif value == "False":
- value = False
- else:
- value = json.loads(value)
- rval[key] = value
- return rval
- def instantiate(model_name, typename, edge=None, ID=""):
- """Create a new instance of the specified typename, between the selected elements (if not None), and with the provided ID (if any)"""
- _goto_mode(MODE_MODIFY, model_name)
- if edge is None:
- _input(["instantiate_node", typename, ID])
- else:
- _input(["instantiate_edge", typename, ID, edge[0], edge[1]])
- return _handle_output("Success: ", split=True)[0]
- def delete_element(model_name, ID):
- """Delete the element with the given ID"""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["delete", ID])
- _handle_output("Success")
- def attr_assign(model_name, ID, attr, value):
- """Assign a value to an attribute"""
- _check_type(value)
- _goto_mode(MODE_MODIFY, model_name)
- _input(["attr_add", ID, attr, value])
- _handle_output("Success")
- def attr_assign_code(model_name, ID, attr, code):
- """Assign a piece of Action Language code to the attribute"""
- _check_type(code)
- try:
- compiled = _compile_AL(code)
- except Exception as e:
- raise CompilationError(e)
- _goto_mode(MODE_MODIFY, model_name)
- _input(["attr_add", ID, attr])
- _handle_output("Waiting for code constructors...")
- _input(compiled)
- _output("Success")
- def attr_delete(model_name, ID, attr):
- """Remove an attribute."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["attr_del", ID, attr])
- _handle_output("Success")
- def read_outgoing(model_name, ID, typename):
- """Returns a list of all outgoing associations of a specific type ("" = all)"""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read_outgoing", ID, typename])
- output = _handle_output("Success: ", split=True)
- return set(output)
- def read_incoming(model_name, ID, typename):
- """Returns a list of all incoming associations of a specific type ("" = all)"""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read_incoming", ID, typename])
- output = _handle_output("Success: ", split=True)
- return set(output)
- def read_association_source(model_name, ID):
- """Returns the source of an association."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read_association_source", ID])
- return _handle_output("Success: ", split=True)[0]
- def read_association_destination(model_name, ID):
- """Returns the destination of an association."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["read_association_destination", ID])
- return _handle_output("Success: ", split=True)[0]
- ##### To document:
- def service_register(name, function):
- """Register a function as a service with a specific name."""
- def service_process(port):
- while 1:
- thrd = threading.Thread(target=function, args=[service_get(port)])
- thrd.daemon = True
- thrd.start()
- global mode
- _goto_mode(MODE_MODELLING)
- _input(["service_register", name])
- # Now we are in service-mode
- mode = MODE_SERVICE
- port = _handle_output("Success: ", split=True)[0]
- # Process events in the background!
- threading.Thread(target=service_process, args=[port]).start()
- def service_stop():
- """Stop the currently executing process."""
- _goto_mode(MODE_SERVICE)
- _input("service_stop")
- _handle_output("Success")
- global mode
- mode = MODE_MODELLING
- def service_get(port):
- """Get the values on the specified port."""
- _goto_mode(MODE_SERVICE)
- return _output(port=port)
- def service_set(port, value):
- """Set a value on a specified port."""
- _check_type_list(value)
- _goto_mode(MODE_SERVICE)
- _input(value, port=port)
- def user_password(user, password):
- """Change a user's password."""
- raise NotImplementedError()
- def transformation_read_signature(transformation):
- """Reads an operation's signature, specifying the names and their required types."""
- raise NotImplementedError()
- def element_list_nice(model_name):
- """Fetches a nice representation of models."""
- _goto_mode(MODE_MODELLING)
- _input(["element_list_nice", model_name, _get_metamodel(model_name)])
- data = _handle_output("Success: ", split=True)[0]
- try:
- return json.loads(data)
- except:
- print(data)
- raise
- def connections_between(model_name, source_element, target_element):
- """Gets a list of all allowed connections between the source and target element in the model."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["connections_between", source_element, target_element])
- output = _handle_output("Success: ", split=True)
- return set(output)
- def define_attribute(model_name, node, attr_name, attr_type):
- """Create a new attribute, which can be instantiated one meta-level below."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["define_attribute", node, attr_name, attr_type])
- return _handle_output("Success: ", split=True)[0]
- def all_instances(model_name, type_name):
- """Returns a list of all elements of a specific type."""
- _goto_mode(MODE_MODIFY, model_name)
- _input(["all_instances", type_name])
- output = _handle_output("Success: ", split=True)
- return set(output)
- def service_poll(port):
- """Checks whether or not the Modelverse side has any input ready to be processed."""
- raise NotImplementedError()
- def user_name(user, username):
- """Change a user's name."""
- raise NotImplementedError()
- def remove_conformance(model_name, metamodel_name):
- """Remove a metamodel for a model."""
- _goto_mode(MODE_MODELLING)
- _input(["remove_conformance", model_name, metamodel_name])
- _handle_output("Success")
- def add_conformance(model_name, metamodel_name, partial_type_mapping=None):
- """Add a metamodel for a model."""
- raise NotImplementedError()
- _goto_mode(MODE_MODELLING)
- _input(["add_conformance", model_name, metamodel_name])
- _handle_output("Success")
- def folder_create(folder_name):
- _goto_mode(MODE_MODELLING)
- _input(["folder_create", folder_name])
- _handle_output("Success")
|