Update ftg and pm parsers to deal with automated activities

drawio2oml/ftg/abstract_syntax.py

@@ -16,7 +16,10 @@ class Transformation(Element):
     ctrl_out: List[str] # names of control output ports
     data_in: List[Tuple[str, Formalism]] # named data input ports
     data_out: List[Tuple[str, Formalism]] # named data output ports
-    automated: bool
+
+@dataclass(eq=False)
+class AutomatedTransformation(Transformation):
+    service_endpoint: str
 
 @dataclass(eq=False)
 class FTG(Element):

drawio2oml/ftg/oml_generator.py

@@ -34,6 +34,7 @@ def write_oml(
         model_name=model_name,
         enumerate=enumerate,
         concat=util.concat,
+        hasattr=hasattr,
         names=names,
         output_namespace=namespaces['artifacts']+model_name+"_ftg",
         shorthand=model_name+"_ftg",

drawio2oml/ftg/oml_template_ftg.j2

@@ -16,7 +16,7 @@ description <{{output_namespace}}#> as {{shorthand}} {
 
   {%- for t in asyntax.transformations %}
 
-  ci {{names[t]}} : ftg:Transformation [
+  ci {{names[t]}} : {{"ftg:" + t.__class__.__name__}} [
     base:hasGUID {{t.name|to_oml_string_literal}}
     {%- for ctrl_in in t.ctrl_in %}
     ftg:hasCtrlInput {{ctrl_in|to_oml_string_literal}}
@@ -24,6 +24,9 @@ description <{{output_namespace}}#> as {{shorthand}} {
     {%- for ctrl_out in t.ctrl_out %}
     ftg:hasCtrlOutput {{ctrl_out|to_oml_string_literal}}
     {%- endfor %}
+    {% if hasattr(t, 'service_endpoint') %}
+    ftg:hasEndpoint {{t.service_endpoint|to_oml_string_literal}}
+    {% endif %}
   ]
   {%- for (portname, f) in t.data_in %}
   ri {{names[t]}}_input{{loop.index0}} : ftg:HasInput [

drawio2oml/ftg/parser.py

@@ -19,16 +19,24 @@ def parse(parent: dio_as.Cell) -> (ftg_as.FTG, List[trace_as.TraceabilityLink]):
 
     for cell in parent.children:
         if cell.properties["pmRole"] == "transformation":
-            t = ftg_as.Transformation(
-                name=cell.properties["name"],
-                ctrl_in=[port.properties["portname"] for port in cell.children if port.properties["pmRole"] == "ctrl_in"],
-                ctrl_out=[port.properties["portname"] for port in cell.children if port.properties["pmRole"] == "ctrl_out"],
-                data_in=[(port.properties["portname"], name_to_formalism[port.incoming[0].source.properties["name"]]) for port in cell.children if port.properties["pmRole"] == "data_in"],
-                data_out=[(port.properties["portname"], name_to_formalism[port.outgoing[0].target.properties["name"]]) for port in cell.children if port.properties["pmRole"] == "data_out"],
-                automated="automated" in cell.properties and cell.properties["automated"]=="true")
+            properties = {
+                "name": cell.properties["name"],
+                "ctrl_in": [port.properties["portname"] for port in cell.children if port.properties["pmRole"] == "ctrl_in"],
+                "ctrl_out": [port.properties["portname"] for port in cell.children if port.properties["pmRole"] == "ctrl_out"],
+                "data_in": [(port.properties["portname"], name_to_formalism[port.incoming[0].source.properties["name"]]) for port in cell.children if port.properties["pmRole"] == "data_in"],
+                "data_out": [(port.properties["portname"], name_to_formalism[port.outgoing[0].target.properties["name"]]) for port in cell.children if port.properties["pmRole"] == "data_out"],            
+            }
+            if "automated" in cell.properties and cell.properties["automated"]=="true":
+                t = ftg_as.AutomatedTransformation(
+                    **properties,
+                    service_endpoint=cell.properties["serviceEndpoint"],
+                )
+            else:
+                t = ftg_as.Transformation(**properties)
             transformations.append(t)
             tracelinks.append(trace_as.TraceabilityLink(dio=cell, ftg=t))
 
     return (ftg_as.FTG(
         formalisms=formalisms,
-        transformations=transformations), tracelinks)
+        transformations=transformations),
+    tracelinks)

drawio2oml/pm/abstract_syntax.py

@@ -63,6 +63,10 @@ class Activity(PMNode):
     data_inports: List[DataInPort]
     data_outports: List[DataOutPort]
 
+@dataclass(eq=False)
+class AutomatedActivity(Activity):
+    timeout: int # milliseconds
+
 @dataclass(eq=False)
 class Artifact(PMNode, DataFlowSink, DataFlowSource):
     name: str

drawio2oml/pm/parser.py

@@ -65,10 +65,18 @@ def parse(parent: dio_as.Cell) -> (pm_as.ProcessModel, List[trace_as.Traceabilit
                 traceability_links.append(trace_as.TraceabilityLink(dio=child, pm=port))
                 collection.append(port)
                 dio_to_pm[child.id] = port
-            pm_element = pm_as.Activity(name=cell.properties["name"],
-                transformation=cell.properties["type"] if cell.properties["type"] != "" else None,
-                ctrl_inports=ctrl_inports, ctrl_outports=ctrl_outports,
-                data_inports=data_inports, data_outports=data_outports)
+            properties = {
+                "name": cell.properties["name"],
+                "transformation": cell.properties["type"] if cell.properties["type"] != "" else None,
+                "ctrl_inports": ctrl_inports,
+                "ctrl_outports": ctrl_outports,
+                "data_inports": data_inports,
+                "data_outports": data_outports,
+            }
+            if "automated" in cell.properties and cell.properties["automated"] == "true":
+                pm_element = pm_as.AutomatedActivity(**properties, timeout=int(cell.properties["timeout"]))
+            else:
+                pm_element = pm_as.Activity(**properties)
             activities.append(pm_element)
         elif pm_role == "artifact":
             pm_element = pm_as.Artifact(name=cell.properties["name"],