123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807 |
- from sccd.runtime.statecharts_core import Event
- import sccd.runtime.socket2event as socket2event
- import modelverse_SCCD as modelverse_SCCD
- import time
- import threading
- import sys
- if sys.version_info[0] < 3:
- from urllib2 import urlopen as urlopen
- from urllib2 import Request as Request
- from urllib import urlencode as urlencode
- else:
- from urllib.request import urlopen as urlopen
- from urllib.request import Request as Request
- from urllib.parse import urlencode as urlencode
- # Exceptions
- class ModelverseException(Exception):
- pass
- class SuperclassAttribute(ModelverseException):
- pass
- class CallbackOnEmptySignature(ModelverseException):
- pass
- class NotAModel(ModelverseException):
- pass
- class ManualActivityRequiresIO(ModelverseException):
- pass
- class UserNotInGroup(ModelverseException):
- pass
- class IncorrectFormat(ModelverseException):
- pass
- class SignatureMismatch(ModelverseException):
- pass
- class EmptySignature(SignatureMismatch):
- pass
- class SourceModelNotBound(SignatureMismatch):
- pass
- class TargetModelNotBound(SignatureMismatch):
- pass
- class DifferingModelsForKey(SignatureMismatch):
- pass
- class TypeMismatch(SignatureMismatch):
- pass
- class UnknownError(ModelverseException):
- pass
- class UnknownM3(ModelverseException):
- pass
- class UnknownIdentifier(ModelverseException):
- pass
- class CompilationError(ModelverseException):
- pass
- class NotAnActivity(ModelverseException):
- pass
- class NotAProcess(ModelverseException):
- pass
- class NotAValidProcess(ModelverseException):
- pass
- class UnknownAttribute(UnknownIdentifier):
- pass
- class UnknownElement(UnknownIdentifier):
- pass
- class UnknownModel(UnknownIdentifier):
- pass
- class UnknownLocation(UnknownIdentifier):
- pass
- class UnknownGroup(UnknownIdentifier):
- pass
- class UnknownUser(UnknownIdentifier):
- pass
- class ConnectionError(ModelverseException):
- pass
- class ExistsError(ModelverseException):
- pass
- class AttributeExists(ExistsError):
- pass
- class ElementExists(ExistsError):
- pass
- class LocationExists(ExistsError):
- pass
- class FolderExists(ExistsError):
- pass
- class GroupExists(ExistsError):
- pass
- class UserExists(ExistsError):
- pass
- class PermissionDenied(ModelverseException):
- pass
- class ReadPermissionDenied(PermissionDenied):
- pass
- class WritePermissionDenied(PermissionDenied):
- pass
- class ExecutePermissionDenied(PermissionDenied):
- pass
- class UserPermissionDenied(PermissionDenied):
- pass
- class GroupPermissionDenied(PermissionDenied):
- pass
- class AdminPermissionDenied(PermissionDenied):
- pass
- class InterfaceMismatch(ModelverseException):
- pass
- class UnknownMetamodellingHierarchy(ModelverseException):
- pass
- class NotAnAssociation(ModelverseException):
- pass
- def run_controller():
- try:
- controller.start()
- finally:
- controller.stop()
- def _next_ID():
- global ID
- ID += 1
- return ID
- def __run_new_modelverse(address, username, password, callback, model):
- init(address)
- login(username, password)
- callback(model)
- exit_save(model)
- disconnect()
- def __run_new_modelverse_activity(address, username, password, taskname, pipe, callback):
- init(address, taskname=taskname)
- controller.username = username
- controller.password = password
- t = OUTPUT()
- if t == "OP":
- model = OUTPUT()
- if callback is not None:
- __invoke(callback, model)
- controller.addInput(Event("data_input", "action_in", [None]))
- time.sleep(2)
- elif t == "SC":
- while 1:
- empty = True
- # Fetch output from the MV
- response = responses.fetch(0)
- if response is not None:
- if response.name == "data_output":
- # Got output of MV, so forward to SCCD
- if pipe is not None:
- pipe.send(("input", response.parameters))
- elif response.name == "result":
- # Finished execution, so continue and return result
- if pipe is not None:
- pipe.send(("terminate", []))
- pipe.close()
- return
- else:
- raise Exception("Unknown data from MV to SC: " + str(response))
- empty = False
- # Fetch output from the SC
- if pipe is not None and pipe.poll():
- response = pipe.recv()
- if response.name == "output":
- controller.addInput(Event("data_input", "action_in", [response.parameters]))
- else:
- raise Exception("Unknown data from SC to MV: " + str(response))
- empty = False
- if empty:
- time.sleep(0.05)
- def __invoke(callback, model):
- import multiprocessing
- p = multiprocessing.Process(target=__run_new_modelverse, args=[controller.address, controller.username, controller.password, callback, model])
- p.start()
- p.join()
- def _process_SC(statechart, port_sc, taskname):
- import multiprocessing
- p2c_pipe, c2p_pipe = multiprocessing.Pipe()
- p = multiprocessing.Process(target=__run_new_modelverse_activity, args=[controller.address, controller.username, controller.password, taskname, c2p_pipe, None])
- p.start()
- while 1:
- empty = True
- if p2c_pipe.poll():
- response = p2c_pipe.recv()
- statechart[0].addInput(Event(response[0], statechart[1], response[1]))
-
- if response[0] == "terminate":
- p2c_pipe.close()
- break
- empty = False
- response = port_sc.fetch(0)
- if response is not None:
- p2c_pipe.send(response)
- empty = False
- if empty:
- time.sleep(0.05)
- p.join()
- def _process_OP(callback, taskname):
- import multiprocessing
- p = multiprocessing.Process(target=__run_new_modelverse_activity, args=[controller.address, controller.username, controller.password, taskname, None, callback])
- p.start()
- p.join()
- def INPUT(action, parameters):
- controller.addInput(Event("action", "action_in", [action, _next_ID(), parameters]))
- def OUTPUT():
- while 1:
- response = responses.fetch(-1)
- if response.name == "result":
- return response.parameters[1]
- elif response.name == "exception":
- try:
- raise eval(response.parameters[1])(*response.parameters[2:])
- except NameError:
- raise UnknownError(response.parameters[1:])
- def init(address_param="127.0.0.1:8001", timeout=60.0, taskname=None):
- global controller
- global ID
- global responses
- controller = modelverse_SCCD.Controller(taskname)
- socket2event.boot_translation_service(controller)
- ID = 0
- thrd = threading.Thread(target=run_controller)
- thrd.daemon = True
- thrd.start()
- responses = controller.addOutputListener("action_out")
- controller.addOutputListener("ready").fetch(-1)
- INPUT("init", [address_param, timeout])
- controller.address = address_param
- OUTPUT()
- return
- def login(username, password):
- if username == None:
- import uuid
- username = str(uuid.uuid4())
- if password == None:
- import uuid
- password = str(uuid.uuid4())
- controller.username = username
- controller.password = password
- INPUT("login", [username, password])
- return OUTPUT()
- def model_list(location):
- INPUT("model_list", [location])
- return OUTPUT()
- def user_password(username, password):
- INPUT("user_password", [username, password])
- return OUTPUT()
- def model_add(model_name, metamodel_name, model_code=""):
- INPUT("model_add", [model_name, metamodel_name, model_code])
- return OUTPUT()
- def model_move(source_name, target_name):
- INPUT("model_move", [source_name, target_name])
- return OUTPUT()
- def model_delete(model_name):
- INPUT("model_delete", [model_name])
- return OUTPUT()
- def model_list_full(location):
- INPUT("model_list_full", [location])
- return OUTPUT()
- def verify(model_name, metamodel_name, conformance_function=""):
- INPUT("verify", [model_name, metamodel_name, conformance_function])
- return OUTPUT()
- def model_overwrite(model_name, new_model):
- INPUT("model_overwrite", [model_name, new_model])
- return OUTPUT()
- def disconnect():
- INPUT("disconnect", [])
- return OUTPUT()
- def user_logout():
- INPUT("user_logout", [])
- return OUTPUT()
- def model_render(model_name, mapper_name, rendered_name):
- INPUT("model_render", [model_name, mapper_name, rendered_name])
- return OUTPUT()
- def transformation_between(sources, targets):
- INPUT("transformation_between", [sources, targets])
- return OUTPUT()
- def transformation_add_MT(source_metamodels, target_metamodels, operation_name, code, callback=None):
- if len(source_metamodels) + len(target_metamodels) == 0:
- if callback is not None:
- raise CallbackOnEmptySignature()
- INPUT("transformation_add_MT", [source_metamodels, target_metamodels, operation_name, code, True])
- model = OUTPUT()
- if model is None:
- return OUTPUT()
- else:
- if callback is not None:
- __invoke(callback, model)
- controller.addInput(Event("data_input", "action_in", [None]))
- return OUTPUT()
- def transformation_add_AL(source_metamodels, target_metamodels, operation_name, code, callback=None):
- if len(source_metamodels) + len(target_metamodels) == 0:
- if callback is not None:
- raise CallbackOnEmptySignature()
- INPUT("transformation_add_AL", [source_metamodels, target_metamodels, operation_name, code, True])
- model = OUTPUT()
- if model is None:
- return OUTPUT()
- else:
- if callback is not None:
- __invoke(callback, model)
- controller.addInput(Event("data_input", "action_in", [None]))
- return OUTPUT()
- def transformation_add_MANUAL(source_metamodels, target_metamodels, operation_name, callback=None):
- if len(source_metamodels) + len(target_metamodels) == 0:
- if callback is not None:
- raise CallbackOnEmptySignature()
- INPUT("transformation_add_MANUAL", [source_metamodels, target_metamodels, operation_name, True])
- model = OUTPUT()
- if model is None:
- return None
- else:
- if callback is not None:
- __invoke(callback, model)
- controller.addInput(Event("data_input", "action_in", [None]))
- return OUTPUT()
- def __transformation_execute(operation_name, input_models_dict, output_models_dict, statechart, tracability_model, fetch_output):
- if statechart is not None:
- port_sc = statechart[0].addOutputListener(statechart[2])
- INPUT("transformation_execute", [operation_name, input_models_dict, output_models_dict, tracability_model, fetch_output])
- taskname = OUTPUT()
- if statechart is not None:
- threading.Thread(target=_process_SC, args=[statechart, port_sc, taskname]).start()
- return OUTPUT()
- def transformation_execute_MT(operation_name, input_models_dict, output_models_dict, statechart=None, traceability_model="", fetch_output=True):
- return __transformation_execute(operation_name, input_models_dict, output_models_dict, statechart, traceability_model, fetch_output)
- def transformation_execute_AL(operation_name, input_models_dict, output_models_dict, statechart=None, traceability_model="", fetch_output=True):
- return __transformation_execute(operation_name, input_models_dict, output_models_dict, statechart, traceability_model, fetch_output)
- def transformation_execute_MANUAL(operation_name, input_models_dict, output_models_dict, callback=None, traceability_model=""):
- INPUT("transformation_execute", [operation_name, input_models_dict, output_models_dict, traceability_model])
- taskname = OUTPUT()
- _process_OP(callback, taskname)
- return OUTPUT()
- def transformation_signature(operation_name):
- INPUT("transformation_signature", [operation_name])
- return OUTPUT()
- def process_signature(process_name):
- INPUT("process_signature", [process_name])
- return OUTPUT()
- def permission_modify(model_name, permissions):
- INPUT("permission_modify", [model_name, permissions])
- return OUTPUT()
- def permission_owner(model_name, permission):
- INPUT("permission_owner", [model_name, permission])
- return OUTPUT()
- def permission_group(model_name, group):
- INPUT("permission_group", [model_name, group])
- return OUTPUT()
- def group_create(group_name):
- INPUT("group_create", [group_name])
- return OUTPUT()
- def group_delete(group_name):
- INPUT("group_delete", [group_name])
- return OUTPUT()
- def group_owner_add(group_name, user_name):
- INPUT("group_owner_add", [group_name, user_name])
- return OUTPUT()
- def group_owner_delete(group_name, user_name):
- INPUT("group_owner_delete", [group_name, user_name])
- return OUTPUT()
- def group_join(group_name, user_name):
- INPUT("group_join", [group_name, user_name])
- return OUTPUT()
- def group_kick(group_name, user_name):
- INPUT("group_kick", [group_name, user_name])
- return OUTPUT()
- def group_list():
- INPUT("group_list", [])
- return OUTPUT()
- def admin_promote(user_name):
- INPUT("admin_promote", [user_name])
- return OUTPUT()
- def admin_demote(user_name):
- INPUT("admin_demote", [user_name])
- return OUTPUT()
- def conformance_delete(model_name, metamodel_name, type_mapping_name):
- INPUT("conformance_delete", [model_name, metamodel_name, type_mapping_name])
- return OUTPUT()
- def conformance_add(model_name, metamodel_name):
- INPUT("conformance_add", [model_name, metamodel_name])
- return OUTPUT()
- def folder_create(folder_name):
- INPUT("folder_create", [folder_name])
- return OUTPUT()
- def model_types(model_name):
- INPUT("model_types", [model_name])
- return OUTPUT()
- def alter_context(model_name, metamodel_name):
- INPUT("alter_context", [model_name, metamodel_name])
- def reset_context():
- INPUT("reset_context", [])
- def element_list(model_name):
- INPUT("element_list", [model_name])
- return OUTPUT()
- def element_list_nice(model_name):
- INPUT("element_list_nice", [model_name])
- return OUTPUT()
- def types(model_name):
- INPUT("types", [model_name])
- return OUTPUT()
- def read_info(model_name, ID):
- INPUT("read_info", [model_name, ID])
- return OUTPUT()
- def read_attrs(model_name, ID):
- INPUT("read_attrs", [model_name, ID])
- return OUTPUT()
- def instantiate(model_name, typename, edge=None, ID=""):
- INPUT("instantiate", [model_name, typename, edge, ID])
- return OUTPUT()
- def delete_element(model_name, ID):
- INPUT("delete_element", [model_name, ID])
- return OUTPUT()
- def attr_assign(model_name, ID, attr, value):
- INPUT("attr_assign", [model_name, ID, attr, value])
- return OUTPUT()
- def attr_assign_code(model_name, ID, attr, code):
- INPUT("attr_assign_code", [model_name, ID, attr, code])
- return OUTPUT()
- def attr_delete(model_name, ID, attr):
- INPUT("attr_delete", [model_name, ID, attr])
- return OUTPUT()
- def AL_text(code_location):
- INPUT("AL_text", [code_location])
- return OUTPUT()
- def read_outgoing(model_name, ID, typename):
- INPUT("read_outgoing", [model_name, ID, typename])
- return OUTPUT()
- def read_incoming(model_name, ID, typename):
- INPUT("read_incoming", [model_name, ID, typename])
- return OUTPUT()
- def read_association_source(model_name, ID):
- INPUT("read_association_source", [model_name, ID])
- return OUTPUT()
- def read_association_destination(model_name, ID):
- INPUT("read_association_destination", [model_name, ID])
- return OUTPUT()
- def connections_between(model_name, source, target):
- INPUT("connections_between", [model_name, source, target])
- return OUTPUT()
- def define_attribute(model_name, node, attr_name, attr_type):
- INPUT("define_attribute", [model_name, node, attr_name, attr_type])
- return OUTPUT()
- def undefine_attribute(model_name, node, attr_name):
- INPUT("undefine_attribute", [model_name, node, attr_name])
- return OUTPUT()
- def attribute_optional(model_name, node, attr_name, optionality):
- INPUT("attr_optional", [model_name, node, attr_name, optionality])
- return OUTPUT()
- def attribute_name(model_name, node, attr_name, new_name):
- INPUT("attr_name", [model_name, node, attr_name, new_name])
- return OUTPUT()
- def attribute_type(model_name, node, attr_name, new_type):
- INPUT("attr_type", [model_name, node, attr_name, new_type])
- return OUTPUT()
- def read_defined_attrs(model_name, node):
- INPUT("read_defined_attrs", [model_name, node])
- return OUTPUT()
- def all_instances(model_name, type_name):
- INPUT("all_instances", [model_name, type_name])
- return OUTPUT()
- def read_permissions(model_name):
- INPUT("read_permissions", [model_name])
- return OUTPUT()
- def process_execute(process_name, model_mapping, callbacks={}):
- # for all callbacks to SCs, start up the output port already
- sc_ports = {}
- for k, v in callbacks.items():
- if isinstance(v, (tuple, list)):
- # Is a statechart, so register already
- sc_ports[k] = v[0].addOutputListener(v[2])
- INPUT("process_execute", [process_name, model_mapping])
- while 1:
- result = OUTPUT()
- if result == "Success":
- # Finished
- return None
- else:
- taskname, operation = result
- if (operation in callbacks):
- data = callbacks[operation]
- if isinstance(data, (tuple, list)):
- # Statechart, so consider like that
- threading.Thread(target=_process_SC, args=[data, sc_ports[operation], taskname]).start()
- else:
- # Assume function
- threading.Thread(target=_process_OP, args=[data, taskname]).start()
- else:
- # Assume empty function
- threading.Thread(target=_process_OP, args=[None, taskname]).start()
- def get_taskname():
- """Fetch the taskname of the current connection."""
- return controller.taskname
- def exit_save(model_name):
- INPUT("exit_save", [model_name])
- return OUTPUT()
- """ Some hardcoded functions... Way easier to express them with code than with statecharts!"""
- import json
- import urllib
- try:
- import urllib2
- except ImportError:
- import urllib as urllib2
- def service_register(name, function):
- """Register a function as a service with a specific name."""
- INPUT("service_register", [name, function])
- port = OUTPUT()
- return port
- def service_stop():
- """Stop the currently executing process."""
- INPUT("service_stop", [])
- return OUTPUT()
- def service_get(port):
- """Get the values on the specified port."""
- data = urlencode({"op": "get_output", "taskname": port}).encode()
- val = json.loads(urlopen(Request("http://%s" % controller.address, data), timeout=99999).read().decode('utf-8'))
- return val
- def service_set(port, value):
- """Set a value on a specified port."""
- if isinstance(value, type([])):
- value = json.dumps(value)
- data = urlencode({"op": "set_input", "data": value, "taskname": port}).encode()
- urlopen(Request("http://%s" % controller.address, data), timeout=99999).read()
- else:
- value = json.dumps(value)
- data = urlencode({"op": "set_input", "value": value, "taskname": port}).encode()
- urlopen(Request("http://%s" % controller.address, data), timeout=99999).read()
- def service_poll(port):
- """Checks whether or not the Modelverse side has any input ready to be processed."""
- raise NotImplementedError()
- def show(model_name):
- data_list = INPUT("element_list_nice", [model_name])
- result = OUTPUT()
- import uuid
- INPUT("model_types", [model_name])
- is_scd = len([i for i in OUTPUT() if i[0] == "formalisms/SimpleClassDiagrams"]) > 0
- edges = []
- texts = []
- rectangles = []
- defined_attributes = {}
- names = {}
- todo_edges = []
- locations = {}
- text_skip = 15
- past_x = 3
- past_y = 3
- max_x = 0
- max_y = 0
- spacing = 50
- if is_scd:
- for elem in result:
- if elem["__type"] == "AttributeLink":
- defined_attributes.setdefault(elem["__source"], []).append((elem["name"], elem["__target"], True if elem["optional"] == True else False))
- if "name" in elem:
- names[elem["__id"]] = elem["name"]
- for elem in list(result):
- is_edge = False
- attrs = {}
- for key, value in elem.items():
- if key == "__id":
- element_id = value
- elif key == "__source":
- element_source = value
- is_edge = True
- elif key == "__target":
- element_target = value
- is_edge = True
- elif key == "__type":
- element_type = value
- else:
- attrs[key] = value
- max_text = 0
- if is_edge:
- edge_edge = len([x for x in result if (x["__id"] == element_source or x["__id"] == element_target) and "__source" in x]) > 0
- else:
- edge_edge = False
- if is_edge:
- if not (elem["__type"] == "AttributeLink" and is_scd):
- todo_edges.append({"source": element_source, "target": element_target, "source_x": 0 if edge_edge else 100, "source_y": 0 if edge_edge else 30, "target_x": 100, "target_y": 30})
- else:
- # Add the group
- x = past_x
- y = past_y
- locations[element_id] = (x, y)
- # Add the text elements
- # First the header
- texts.append({"text": "%s : %s" % (element_id, element_type), "x": x + 10, "y": y + text_skip})
- max_text = max(len(texts[-1]["text"]), max_text)
- # Then the attributes
- text_counter = 1
- for name, attr_type, optional in defined_attributes.get(elem["__id"], []):
- text = "%s : %s" % (name, names.get(attr_type, "(%s)" % attr_type))
- if optional:
- text = text.replace(" : ", " ?: ")
- texts.append({"text": text, "x": x + 10, "y": y + text_skip * 1.25 + text_counter * text_skip})
- max_text = max(len(text), max_text)
- text_counter += 1
- if text_counter > 1:
- text_counter += 1
- y_loc_edge = y + text_skip + text_counter * text_skip
- for key, value in attrs.items():
- if isinstance(value, dict):
- if value["AL"] == "":
- text = "(%s)" % key
- else:
- text = "%s = ^%s" % (key, value['AL'])
- else:
- text = ("%s = %s" % (key, json.dumps(value))) if value is not None else ("(%s)" % key)
- texts.append({"text": text, "x": x + 10, "y": y + text_skip * 2.5 + text_counter * text_skip})
- max_text = max(len(text), max_text)
- text_counter += 1
- # Add the rectangle
- rectangles.append({"x": x, "y": y, "width": max_text * 9 + 10, "height": text_skip * 3 + text_counter * text_skip})
- # Add the line
- edges.append({"sx": x, "sy": y_loc_edge, "tx": x + max_text * 9 + 10, "ty": y_loc_edge})
- edges.append({"sx": x, "sy": y + 20, "tx": x + max_text * 9 + 10, "ty": y + 20})
- past_x = x + max_text * 9 + spacing
- if past_x > 1000:
- max_x = max(max_x, past_x)
- past_x = 3
- past_y = max_y + spacing
- max_x = max(max_x, past_x)
- max_y = max(max_y, past_y + text_skip * 3 + text_counter * text_skip)
- data_content = '<svg width="%s" height="%s">' % (max_x + 3, max_y + 3)
- data_content += '<marker orient="auto" refY="0.0" refX="0.0" id="triangle" style="overflow:visible;"><path id="path941" d="M 0.0,0.0 L 5.0,-5.0 L -12.5,0.0 L 5.0,5.0 L 0.0,0.0 z " style="fill-rule:evenodd;stroke:#000000;stroke-width:1pt;stroke-opacity:1;fill:#000000;fill-opacity:1" transform="scale(0.8) rotate(180) translate(12.5,0)"/></marker>'
- for edge in todo_edges:
- sx = locations[edge["source"]][0] + edge["source_x"]
- sy = locations[edge["source"]][1] + edge["source_y"]
- tx = locations[edge["target"]][0] + edge["target_x"]
- ty = locations[edge["target"]][1] + edge["target_y"]
- midx = (tx - sx) / 2 + sx
- midy = (ty - sy) / 2 + sy
- data_content += '<line x1="%s" y1="%s" x2="%s" y2="%s" stroke="black" style="marker-end:url(#triangle)"/>' % (sx, sy, midx, midy)
- data_content += '<line x1="%s" y1="%s" x2="%s" y2="%s" stroke="black"/>' % (midx, midy, tx, ty)
- for rec in rectangles:
- data_content += '<rect x="%s" y="%s" width="%s" height="%s" fill="white" stroke="black"/>' % (rec["x"], rec["y"], rec["width"], rec["height"])
- for edge in edges:
- data_content += '<line x1="%s" y1="%s" x2="%s" y2="%s" stroke="black"/>' % (edge["sx"], edge["sy"], edge["tx"], edge["ty"])
- for text in texts:
- data_content += '<text x="%s" y="%s" fill="black">%s</text>' % (text["x"], text["y"], text["text"])
- data_content += '</svg>'
- return data_content
|