Switch entirely to event-driven parsing.

src/sccd/model/context.py

@@ -33,7 +33,7 @@ class Context:
     # Ensure delta not too big
     if self.fixed_delta:
       if self.delta < self.fixed_delta:
-        raise Exception("Model contains duration deltas (smallest = %s) than representable with delta given (%s)." % (str(self.delta), str(self.fixed_delta)))
+        raise Exception("Model contains duration deltas (smallest = %s) not representable with fixed delta of %s." % (str(self.delta), str(self.fixed_delta)))
       else:
         self.delta = self.fixed_delta
 

src/sccd/model/statechart_parser.py

@@ -0,0 +1,340 @@
+import dataclasses
+from lxml import etree
+from sccd.model.expression_parser import *
+from sccd.syntax.statechart import *
+from sccd.syntax.tree import *
+
+
+class XmlLoadError(Exception):
+  def __init__(self, src_file: str, el: etree.Element, err):
+    parent = el.getparent()
+    if parent is None:
+      parent = el
+    # el = parent
+    lines = etree.tostring(parent).decode('utf-8').strip().split('\n')
+    nbr_lines = len(etree.tostring(el).decode('utf-8').strip().split('\n'))
+    lines_numbers = []
+    l = parent.sourceline
+    for line in lines:
+      ll = ("%4d: " % l) + line
+      if l >= el.sourceline and l < el.sourceline + nbr_lines:
+        ll = termcolor.colored(ll, 'yellow')
+      lines_numbers.append(ll)
+      l += 1
+    super().__init__("\n\n%s\n\n%s:\nline %d: <%s>: %s" % ('\n'.join(lines_numbers), src_file,el.sourceline, el.tag, str(err)))
+
+
+class Parser:
+
+  def _raise(self, el, err):
+    src_file = self.require("src_file")
+    raise XmlLoadError(src_file, el, err)
+
+  def _get_stack(self, name):
+    stack = getattr(self, '_'+name, None)
+    if stack is None:
+      stack = []
+      setattr(self, '_'+name, stack)
+    return stack
+
+  def push(self, name, value):
+    stack = self._get_stack(name)
+    stack.append(value)
+
+  def pop(self, name):
+    stack = getattr(self, '_'+name)
+    return stack.pop()
+
+  def get(self, name, default=None):
+    stack = getattr(self, '_'+name, [])
+    return stack[-1] if len(stack) else default
+
+  def require(self, name):
+    stack = getattr(self, '_'+name, [])
+    if len(stack) == 0:
+      raise Exception("Element expected only within context: %s" % name)
+    return stack[-1]
+
+  def all(self, name):
+    stack = self._get_stack(name)
+    return stack
+
+  def parse(self, event_generator):
+    result = None
+    for event, el in event_generator:
+      print(event, el.tag)
+      try:
+        if event == "start":
+          start_method = getattr(self, "start_"+el.tag, None)
+          if start_method:
+            start_method(el)
+
+        elif event == "end":
+          end_method = getattr(self, "end_"+el.tag)
+          if end_method:
+            result = end_method(el)
+      except XmlLoadError:
+        raise
+      # Decorate non-XmlLoadErrors
+      except Exception as e:
+        self._raise(el, e)
+
+      # We don't need anything from this element anymore, so we clear it to save memory.
+      # This is a technique mentioned in the lxml documentation:
+      # https://lxml.de/tutorial.html#event-driven-parsing
+      # el.clear()
+    return result
+
+
+class StatechartParser(Parser):
+
+  def end_var(self, el):
+    context = self.require("context")
+    datamodel = self.require("datamodel")
+
+    id = el.get("id")
+    expr = el.get("expr")
+    parsed = parse_expression(context, datamodel, expr=expr)
+    datamodel.create(id, parsed.eval([], datamodel))
+
+  def start_datamodel(self, el):
+    statechart = self.require("statechart")
+    self.push("datamodel", statechart.datamodel)
+
+  def end_datamodel(self, el):
+    self.pop("datamodel")
+
+
+  def end_raise(self, el):
+    context = self.require("context")
+    actions = self.require("actions")
+    name = el.get("event")
+    port = el.get("port")
+    if not port:
+      event_id = context.events.assign_id(name)
+      a = RaiseInternalEvent(name=name, parameters=[], event_id=event_id)
+    else:
+      context.outports.assign_id(port)
+      a = RaiseOutputEvent(name=name, parameters=[], outport=port, time_offset=0)
+    actions.append(a)
+
+  def end_code(self, el):
+    context = self.require("context")
+    datamodel = self.require("datamodel")
+    actions = self.require("actions")
+
+    block = parse_block(context, datamodel, block=el.text)
+    a = Code(block)
+    actions.append(a)
+
+
+  def _internal_start_state(self, el, constructor):
+    parent = self.get("state", default=None)
+
+    short_name = el.get("id", "")
+    if parent is None:
+      if short_name:
+        raise Exception("Root <state> must not have 'id' attribute.")
+    else:
+      if not short_name:
+        raise Exception("Non-root <state> must have 'id' attribute.")
+
+    state = constructor(short_name, parent)
+
+    parent_children = self.require("state_children")
+    already_there = parent_children.setdefault(short_name, state)
+    if already_there is not state:
+      if parent:
+        raise Exception("Sibling state with the same id exists.")
+      else:
+        raise Exception("Only 1 root <state> allowed.")
+
+    self.push("state", state)
+    self.push("state_children", {})
+
+  def _internal_end_state(self):
+    state_children = self.pop("state_children")
+    state = self.pop("state")
+    return (state, state_children)
+
+
+  def start_state(self, el):
+    self._internal_start_state(el, State)
+
+  def end_state(self, el):
+    state, state_children = self._internal_end_state()
+
+    initial = el.get("initial", None)
+    if initial is not None:
+      state.default_state = state_children[initial]
+    elif len(state.children) == 1:
+      state.default_state = state.children[0]
+    elif len(state.children) > 1:
+      raise Exception("More than 1 child state: must set 'initial' attribute.")
+
+  def start_parallel(self, el):
+    self._internal_start_state(el, ParallelState)
+
+  def end_parallel(self, el):
+    self._internal_end_state()
+
+  def start_history(self, el):
+    if el.get("type", "shallow") == "deep":
+      self._internal_start_state(el, DeepHistoryState)
+    else:
+      self._internal_start_state(el, ShallowHistoryState)
+
+  def end_history(self, el):
+    return self._internal_end_state()
+
+
+  def start_onentry(self, el):
+    self.push("actions", [])
+
+  def end_onentry(self, el):
+    actions = self.pop("actions")
+    self.require("state").enter = actions
+
+  def start_onexit(self, el):
+    self.push("actions", [])
+
+  def end_onexit(self, el):
+    actions = self.pop("actions")
+    self.require("state").exit = actions
+
+
+  def start_transition(self, el):
+    self.push("actions", [])
+
+  def end_transition(self, el):
+    actions = self.pop("actions")
+    # simply accumulate transition elements
+    # we'll deal with them in end_tree()
+    source = self.require("state")
+
+    # get stuff from element
+    target = el.get("target", "")
+    event = el.get("event")
+    port = el.get("port")
+    after = el.get("after")
+    cond = el.get("cond")
+
+    self.require("transitions").append((el, target, event, port, after, cond, source, actions))
+
+
+  def start_tree(self, el):
+    statechart = self.require("statechart")
+    self.push("datamodel", statechart.datamodel)
+    self.push("transitions", [])
+    self.push("state_children", {})
+
+  def end_tree(self, el):
+    statechart = self.require("statechart")
+    context = self.require("context")
+    datamodel = self.pop("datamodel")
+
+    root_states = self.pop("state_children")
+    if len(root_states) == 0:
+      raise Exception("Missing root <state> !")
+    root = list(root_states.values())[0]
+
+    # Add transitions.
+    # Only now that our tree structure is complete can we resolve 'target' states of transitions.
+    next_after_id = 0
+    transitions = self.pop("transitions")
+    for t_el, target_string, event, port, after, cond, source, actions in transitions:
+      try:
+        # Parse and find target state
+        parse_tree = parse_state_ref(target_string)
+      except Exception as e:
+        self._raise(t_el, "Parsing target '%s': %s" % (target_string, str(e)))
+
+      def find_state(sequence) -> State:
+        if sequence.data == "relative_path":
+          state = source
+        elif sequence.data == "absolute_path":
+          state = root
+        for item in sequence.children:
+          if item.type == "PARENT_NODE":
+            state = state.parent
+          elif item.type == "CURRENT_NODE":
+            continue
+          elif item.type == "IDENTIFIER":
+            state = [x for x in state.children if x.short_name == item.value][0]
+        return state
+
+      try:
+        targets = [find_state(seq) for seq in parse_tree.children]
+      except:
+        self._raise(t_el, "Could not find target '%s'." % (target_string))
+
+      transition = Transition(source, targets)
+
+      # Trigger
+      if after is not None:
+        after_expr = parse_expression(context, datamodel, expr=after)
+        # print(after_expr)
+        event = "_after%d" % next_after_id # transition gets unique event name
+        next_after_id += 1
+        trigger = AfterTrigger(context.events.assign_id(event), event, after_expr)
+      elif event is not None:
+        trigger = Trigger(context.events.assign_id(event), event, port)
+        context.inports.assign_id(port)
+      else:
+        trigger = None
+      transition.trigger = trigger
+      # Actions
+      transition.actions = actions
+      # Guard
+      if cond is not None:
+        try:
+          expr = parse_expression(context, datamodel, expr=cond)
+        except Exception as e:
+          self._raise(t_el, "Condition '%s': %s" % (cond, str(e)))
+        transition.guard = expr
+      source.transitions.append(transition)
+
+    statechart.tree = StateTree(root)
+
+
+  def _internal_end_semantics(self, el):
+    statechart = self.require("statechart")
+    # Use reflection to find the possible XML attributes and their values
+    for aspect in dataclasses.fields(Semantics):
+      key = el.get(aspect.name)
+      if key is not None:
+        if key == "*":
+          setattr(statechart.semantics, aspect.name, None)
+        else:
+          value = aspect.type[key.upper()]
+          setattr(statechart.semantics, aspect.name, value)
+
+  def end_semantics(self, el):
+    self._internal_end_semantics(el)
+
+  def end_override_semantics(self, el):
+    self._internal_end_semantics(el)
+
+
+  def start_statechart(self, el):
+    src_file = self.require("src_file")
+    ext_file = el.get("src")
+    if ext_file is None:
+      statechart = Statechart(
+        tree=None, semantics=Semantics(), datamodel=DataModel())
+    else:
+      ext_file_path = os.path.join(os.path.dirname(src_file), ext_file)
+      self.push("src_file", ext_file_path)
+      self.push("statecharts", [])
+      self.parse(etree.iterparse(ext_file_path, events=("start", "end")))
+      statecharts = self.pop("statecharts")
+      if len(statecharts) != 1:
+        raise Exception("Expected exactly 1 <statechart> node, got %d." % len(statecharts))
+      statechart = statecharts[0]
+      self.pop("src_file")
+    self.push("statechart", statechart)
+
+  def end_statechart(self, el):
+    statecharts = self.require("statecharts")
+    sc = self.pop("statechart")
+    statecharts.append(sc)

src/sccd/model/xml_loader.py

@@ -1,193 +0,0 @@
-import lxml.etree as ET
-import dataclasses
-from copy import deepcopy
-
-from sccd.syntax.statechart import *
-from sccd.model.context import *
-from sccd.model.parser import *
-from sccd.model.xml_parser import *
-
-# parent_node: XML node containing any number of action nodes as direct children
-def load_actions(context: Context, datamodel, parent_node) -> List[Action]:
-  def load_action(action_node) -> Optional[Action]:
-      # tag = ET.QName(action_node).localname
-      tag = action_node.tag
-      if tag == "raise":
-        name = action_node.get("event")
-        port = action_node.get("port")
-        if not port:
-          event_id = context.events.assign_id(name)
-          return RaiseInternalEvent(name=name, parameters=[], event_id=event_id)
-        else:
-          context.outports.assign_id(port)
-          return RaiseOutputEvent(name=name, parameters=[], outport=port, time_offset=0)
-      elif tag == "code":
-        try:
-          block = parse_block(context, datamodel, block=action_node.text)
-        except Exception as e:
-          raise XmlLoadError(action_node, "Parsing code: %s" % str(e))
-          # raise Exception("Line %d: <%s>: Error parsing code: '%s'" % (action_node.sourceline, tag, action_node.text))
-        return Code(block)
-      else:
-        raise XmlLoadError(action_node, "Unsupported action tag.")
-  return [load_action(child) for child in parent_node if child.tag is not ET.Comment]
-
-# Load state tree from XML <tree> node.
-# Context is required for building event namespace and in/outport discovery.
-def load_tree(context: Context, datamodel, tree_node) -> StateTree:
-
-  transition_nodes: List[Tuple[Any, State]] = [] # List of (<transition>, State) tuples
-
-  # Recursively create state hierarchy from XML node
-  # Adding <transition> elements to the 'transitions' list as a side effect
-  def load_state(state_node, parent: State) -> Optional[State]:
-    name = state_node.get("id", "")
-    # tag = ET.QName(state_node).localname
-    tag = state_node.tag
-    if tag == "state":
-        state = State(name, parent)
-        initial = state_node.get("initial")
-    elif tag == "parallel" : 
-        state = ParallelState(name, parent)
-    elif tag == "history":
-      is_deep = state_node.get("type", "shallow") == "deep"
-      if is_deep:
-        state = DeepHistoryState(name, parent)
-      else:
-        state = ShallowHistoryState(name, parent)
-    else:
-      return None
-
-    for xml_child in state_node.getchildren():
-        if xml_child.tag is ET.Comment:
-          continue # skip comments
-        child = load_state(xml_child, parent=state) # may throw
-        if child and tag == "state" and child.short_name == initial:
-            state.default_state = child
-
-    if tag == "state" and len(state.children) > 0 and not state.default_state:
-      if len(state.children) == 1 and initial is None:
-        state.default_state = state.children[0]
-
-      if state.default_state is None:
-        raise XmlLoadError(state_node, "Must set 'initial' attribute.")
-
-    for xml_t in state_node.findall("transition", state_node.nsmap):
-      transition_nodes.append((xml_t, state))
-
-    # Parse enter/exit actions
-    def _get_enter_exit(tag):
-      node = state_node.find(tag, state_node.nsmap)
-      if node is not None:
-        return load_actions(context, datamodel, node)
-      else:
-        return []
-
-    state.enter = _get_enter_exit("onentry")
-    state.exit = _get_enter_exit("onexit")
-
-    return state
-
-  # Build tree structure
-  root_node = tree_node.find("state")
-  root = load_state(root_node, parent=None)
-
-  # Add transitions
-  next_after_id = 0
-  for t_node, source in transition_nodes:
-    try:
-      # Parse and find target state
-      target_string = t_node.get("target")
-      parse_tree = parse_state_ref(target_string)
-      def find_state(sequence) -> State:
-        if sequence.data == "relative_path":
-          el = source
-        elif sequence.data == "absolute_path":
-          el = root
-        for item in sequence.children:
-          if item.type == "PARENT_NODE":
-            el = el.parent
-          elif item.type == "CURRENT_NODE":
-            continue
-          elif item.type == "IDENTIFIER":
-            el = [x for x in el.children if x.short_name == item.value][0]
-        return el
-      targets = [find_state(seq) for seq in parse_tree.children]
-    except:
-      raise XmlLoadError(t_node, "Could not find target '%s'" % target_string)
-
-    transition = Transition(source, targets)
-
-    # Trigger
-    name = t_node.get("event")
-    port = t_node.get("port")
-    after = t_node.get("after")
-    if after is not None:
-      after_expr = parse_expression(context, datamodel, expr=after)
-      # print(after_expr)
-      name = "_after%d" % next_after_id # transition gets unique event name
-      next_after_id += 1
-      trigger = AfterTrigger(context.events.assign_id(name), name, after_expr)
-    elif name is not None:
-      trigger = Trigger(context.events.assign_id(name), name, port)
-      context.inports.assign_id(port)
-    else:
-      trigger = None
-    transition.trigger = trigger
-    # Actions
-    actions = load_actions(context, datamodel, t_node)
-    transition.actions = actions
-    # Guard
-    cond = t_node.get("cond")
-    if cond is not None:
-      try:
-        expr = parse_expression(context, datamodel, expr=cond)
-        # print(expr)
-      except Exception as e:
-        raise XmlLoadError(t_node, "Condition '%s': %s" % (cond, str(e)))
-      transition.guard = expr
-    source.transitions.append(transition)
-
-  return StateTree(root=root)
-
-def load_semantics(semantics: Semantics, semantics_node):
-  if semantics_node is not None:
-    # Use reflection to find the possible XML attributes and their values
-    for aspect in dataclasses.fields(Semantics):
-      key = semantics_node.get(aspect.name)
-      if key is not None:
-        if key == "*":
-          setattr(semantics, aspect.name, None)
-        else:
-          value = aspect.type[key.upper()]
-          setattr(semantics, aspect.name, value)
-
-def load_datamodel(context: Context, datamodel_node) -> DataModel:
-  datamodel = DataModel()
-  if datamodel_node is not None:
-    for var_node in datamodel_node.findall("var"):
-      id = var_node.get("id")
-      expr = var_node.get("expr")
-      val = parse_expression(context, datamodel, expr=expr)
-      datamodel.create(id, val.eval([], datamodel))
-      # datamodel.names[id] = Variable(val.eval([], datamodel))
-  return datamodel
-
-# Context is required for building event namespace and in/outport discovery.
-def load_statechart(context: Context, sc_node) -> Statechart:
-  datamodel_node = sc_node.find("datamodel")
-  datamodel = load_datamodel(context, datamodel_node)
-
-  tree_node = sc_node.find("tree")
-  handler = TreeHandler(context, datamodel)
-  parse(ET.iterwalk(tree_node, events=("start", "end")), handler)
-  state_tree = handler.tree
-
-  # tree_node = sc_node.find("tree")
-  # state_tree = load_tree(context, datamodel, tree_node)
-
-  semantics_node = sc_node.find("semantics")
-  semantics = Semantics() # start with default semantics
-  load_semantics(semantics, semantics_node)
-
-  return Statechart(tree=state_tree, semantics=semantics, datamodel=datamodel)

src/sccd/model/xml_parser.py

@@ -1,262 +0,0 @@
-from lxml import etree
-from sccd.model.context import *
-from sccd.model.parser import *
-from sccd.syntax.tree import *
-
-class XmlLoadError(Exception):
-  def __init__(self, el: etree.Element, err: Exception):
-    parent = el.getparent()
-    if parent is None:
-      parent = el
-    # el = parent
-    lines = etree.tostring(parent).decode('utf-8').strip().split('\n')
-    nbr_lines = len(etree.tostring(el).decode('utf-8').strip().split('\n'))
-    lines_numbers = []
-    l = parent.sourceline
-    for line in lines:
-      ll = ("%4d: " % l) + line
-      if l >= el.sourceline and l < el.sourceline + nbr_lines:
-        ll = termcolor.colored(ll, 'yellow')
-      lines_numbers.append(ll)
-      l += 1
-    super().__init__("\n\n%s\n\nLine %d: <%s>: %s" % ('\n'.join(lines_numbers), el.sourceline, el.tag, str(err)))
-
-class ElementHandler:
-
-  def _get_stack(self, name):
-    stack = getattr(self, name, None)
-    if stack is None:
-      stack = []
-      setattr(self, name, stack)
-    return stack
-
-  def push(self, name, value):
-    stack = self._get_stack(name)
-    stack.append(value)
-
-  def pop(self, name):
-    stack = getattr(self, name)
-    return stack.pop()
-
-  def top(self, name, default=None):
-    stack = getattr(self, name, [])
-    return stack[-1] if len(stack) else default
-
-  def all(self, name):
-    stack = self._get_stack(name)
-    return stack
-
-class TreeHandler(ElementHandler):
-
-  def __init__(self, context, datamodel):
-    self.context = context
-    self.datamodel = datamodel
-
-  def end_raise(self, el):
-    name = el.get("event")
-    port = el.get("port")
-    if not port:
-      event_id = self.context.events.assign_id(name)
-      a = RaiseInternalEvent(name=name, parameters=[], event_id=event_id)
-    else:
-      self.context.outports.assign_id(port)
-      a = RaiseOutputEvent(name=name, parameters=[], outport=port, time_offset=0)
-    self.top("actions").append(a)
-    return a
-
-  def end_code(self, el):
-    block = parse_block(self.context, self.datamodel, block=el.text)
-    a = Code(block)
-    self.top("actions").append(a)
-    return a
-
-  def _start_state(self, el, constructor):
-    parent = self.top("state", default=None)
-
-    short_name = el.get("id", "")
-    if parent is None:
-      if short_name:
-        raise XmlLoadError(el, "Root <state> must not have 'id' attribute.")
-    else:
-      if not short_name:
-        raise XmlLoadError(el, "Non-root <state> must have 'id' attribute.")
-
-    state = constructor(short_name, parent)
-
-    parent_children = self.top("state_children")
-    already_there = parent_children.setdefault(short_name, state)
-    if already_there is not state:
-      if parent:
-        raise XmlLoadError(el, "Sibling state with the same id exists.")
-      else:
-        raise XmlLoadError(el, "Only 1 root <state> allowed.")
-
-    self.push("state", state)
-    self.push("state_children", {})
-
-  def _end_state(self):
-    # self.pop("state_prefix")
-    state_children = self.pop("state_children")
-    state = self.pop("state")
-    return (state, state_children)
-
-  def start_state(self, el):
-    self._start_state(el, State)
-
-  def end_state(self, el):
-    state, state_children = self._end_state()
-
-    initial = el.get("initial", None)
-    if initial is not None:
-      state.default_state = state_children[initial]
-    elif len(state.children) == 1:
-      state.default_state = state.children[0]
-    elif len(state.children) > 1:
-      raise XmlLoadError(el, "More than 1 child state: must set 'initial' attribute.")
-
-  def start_parallel(self, el):
-    self._start_state(el, ParallelState)
-
-  def end_parallel(self, el):
-    self._end_state()
-
-  def start_history(self, el):
-    if el.get("type", "shallow") == "deep":
-      self._start_state(el, DeepHistoryState)
-    else:
-      self._start_state(el, ShallowHistoryState)
-
-  def end_history(self, el):
-    return self._end_state()
-
-  def start_onentry(self, el):
-    self.push("actions", [])
-
-  def end_onentry(self, el):
-    actions = self.pop("actions")
-    self.top("state").enter = actions
-
-  def start_onexit(self, el):
-    self.push("actions", [])
-
-  def end_onexit(self, el):
-    actions = self.pop("actions")
-    self.top("state").exit = actions
-
-  def start_transition(self, el):
-    self.push("actions", [])
-
-  def end_transition(self, el):
-    actions = self.pop("actions")
-    # simply accumulate transition elements
-    # we'll deal with them in end_tree()
-    source = self.top("state")
-
-    # get stuff from element
-    target = el.get("target", "")
-    event = el.get("event")
-    port = el.get("port")
-    after = el.get("after")
-    cond = el.get("cond")
-
-    self.top("transitions").append((el, target, event, port, after, cond, source, actions))
-
-
-  def start_tree(self, el):
-    # self.push("tree", StateTree())
-    self.push("transitions", [])
-    self.push("state_children", {})
-
-  def end_tree(self, el):
-    root_states = self.pop("state_children")
-    if len(root_states) == 0:
-      raise XmlLoadError(el, "Missing root <state> !")
-    root = list(root_states.values())[0]
-
-    transitions = self.pop("transitions")
-
-    # Add transitions.
-    # Only now that our tree structure is complete can we resolve 'target' states of transitions.
-    next_after_id = 0
-    for t_el, target_string, event, port, after, cond, source, actions in transitions:
-      try:
-        # Parse and find target state
-        parse_tree = parse_state_ref(target_string)
-      except Exception as e:
-        raise XmlLoadError(t_el, "Parsing target '%s': %s" % (target_string, str(e)))
-
-      def find_state(sequence) -> State:
-        if sequence.data == "relative_path":
-          state = source
-        elif sequence.data == "absolute_path":
-          state = root
-        for item in sequence.children:
-          if item.type == "PARENT_NODE":
-            state = state.parent
-          elif item.type == "CURRENT_NODE":
-            continue
-          elif item.type == "IDENTIFIER":
-            state = [x for x in state.children if x.short_name == item.value][0]
-        return state
-
-      try:
-        targets = [find_state(seq) for seq in parse_tree.children]
-      except:
-        raise XmlLoadError(t_el, "Could not find target '%s'." % (target_string))
-
-      transition = Transition(source, targets)
-
-      # Trigger
-      if after is not None:
-        after_expr = parse_expression(self.context, self.datamodel, expr=after)
-        # print(after_expr)
-        event = "_after%d" % next_after_id # transition gets unique event name
-        next_after_id += 1
-        trigger = AfterTrigger(self.context.events.assign_id(event), event, after_expr)
-      elif event is not None:
-        trigger = Trigger(self.context.events.assign_id(event), event, port)
-        self.context.inports.assign_id(port)
-      else:
-        trigger = None
-      transition.trigger = trigger
-      # Actions
-      transition.actions = actions
-      # Guard
-      if cond is not None:
-        try:
-          expr = parse_expression(self.context, self.datamodel, expr=cond)
-        except Exception as e:
-          raise XmlLoadError(t_el, "Condition '%s': %s" % (cond, str(e)))
-        transition.guard = expr
-      source.transitions.append(transition)
-
-    self.tree = StateTree(root)
-
-  def start_datamodel(self, el):
-    pass
-
-def parse(event_generator, handler: ElementHandler):
-  # for event, el in etree.iterparse(file, events=("start", "end")):
-  for event, el in event_generator:
-
-    try:
-      if event == "start":
-        start_method = getattr(handler, "start_"+el.tag, None)
-        if start_method:
-          start_method(el)
-
-      elif event == "end":
-        end_method = getattr(handler, "end_"+el.tag)
-        if end_method:
-          end_method(el)
-
-    except XmlLoadError:
-      raise
-    # Decorate non-XmlLoadErrors
-    except Exception as e:
-      raise XmlLoadError(el, e)
-
-      # We don't need anything from this element anymore, so we clear it to save memory.
-      # This is a technique mentioned in the lxml documentation:
-      # https://lxml.de/tutorial.html#event-driven-parsing
-      # el.clear()

src/sccd/syntax/tree.py

@@ -3,6 +3,7 @@ from typing import *
 from sccd.syntax.action import *
 from sccd.util.bitmap import *
 
+
 @dataclass
 class State:
     short_name: str # value of 'id' attribute in XML
@@ -27,65 +28,29 @@ class State:
         if self.default_state:
             targets.extend(self.default_state.getEffectiveTargetStates(instance))
         return targets
-
-    # # Recursively assigns unique state_id to each state in the tree,
-    # # as well as some other optimization stuff
-    # # Should only be called once for the root of the state tree,
-    # # after the tree has been built.
-    # # Returns state_id + total number of states in tree
-    # def init_tree(self, state_id: int = 0, name_prefix: str = "", states = {}, state_list = [], transition_list = []) -> int:
-    #     self.state_id = state_id
-    #     next_id = state_id + 1
-    #     self.name = name_prefix + self.short_name if name_prefix == '/' else name_prefix + '/' + self.short_name
-    #     states[self.name] = self
-    #     state_list.append(self)
-    #     for t in self.transitions:
-    #         transition_list.append(t)
-    #     for i, c in enumerate(self.children):
-    #         if isinstance(c, HistoryState):
-    #             self.history.append(c)
-    #         c.parent = self
-    #         c.ancestors.append(self)
-    #         c.ancestors.extend(self.ancestors)
-    #         next_id = c.init_tree(next_id, self.name, states, state_list, transition_list)
-    #     self.descendants.extend(self.children)
-    #     for c in self.children:
-    #         self.descendants.extend(c.descendants)
-    #     for d in self.descendants:
-    #         self.descendant_bitmap |= bit(d.state_id)
-    #     return next_id
-
-    # def print(self, w = FormattedWriter()):
-    #     w.write(self.name)
-    #     w.indent()
-    #     for c in self.children:
-    #         c.print(w)
-    #     w.dedent()
                     
     def __repr__(self):
         return "State(\"%s\")" % (self.gen.full_name)
 
 # Generated fields (for optimization) of a state
-@dataclass
+@dataclass(frozen=True)
 class StateGenerated:
     state_id: int
     full_name: str
-    ancestors: List[State] = field(default_factory=list) # order: close to far away, i.e. first element is parent
-    descendants: List[State] = field(default_factory=list)  # order: breadth-first
-    descendant_bitmap: Bitmap = Bitmap()
-    history: List[State] = field(default_factory=list) # subset of children
-    has_eventless_transitions: bool = False
-    after_triggers: List['AfterTrigger'] = field(default_factory=list)
+    ancestors: List[State] # order: close to far away, i.e. first element is parent
+    descendants: List[State]  # order: breadth-first
+    descendant_bitmap: Bitmap
+    history: List[State] # subset of children
+    has_eventless_transitions: bool
+    after_triggers: List['AfterTrigger']
 
 
 class HistoryState(State):
-    pass
-    # def __init__(self, name):
-        # State.__init__(self, name)
+    @abstractmethod
+    def getEffectiveTargetStates(self, instance):
+        pass
         
 class ShallowHistoryState(HistoryState):
-    # def __init__(self, name):
-        # HistoryState.__init__(self, name)
         
     def getEffectiveTargetStates(self, instance):
         if self.state_id in instance.history_values:
@@ -98,8 +63,6 @@ class ShallowHistoryState(HistoryState):
             return self.parent.getEffectiveTargetStates(instance)
         
 class DeepHistoryState(HistoryState):
-    # def __init__(self, name):
-        # HistoryState.__init__(self, name)
         
     def getEffectiveTargetStates(self, instance):
         if self.state_id in instance.history_values:
@@ -109,8 +72,6 @@ class DeepHistoryState(HistoryState):
             return self.parent.getEffectiveTargetStates(instance)
         
 class ParallelState(State):
-    # def __init__(self, name):
-        # State.__init__(self, name)
         
     def getEffectiveTargetStates(self, instance):
         targets = [self]
@@ -165,7 +126,7 @@ class Transition:
         return termcolor.colored("%s 🡪 %s" % (self.source.gen.full_name, self.targets[0].gen.full_name), 'green')
 
 # Generated fields (for optimization) of a transition
-@dataclass
+@dataclass(frozen=True)
 class TransitionGenerated:
     lca: State
     arena_bitmap: Bitmap
@@ -185,6 +146,9 @@ class StateTree:
         def init_tree(state: State, parent_full_name: str, ancestors: List[State]):
             nonlocal next_id
 
+            state_id = next_id
+            next_id += 1
+
             if state is root:
                 full_name = '/'
             elif state.parent is root:
@@ -192,36 +156,39 @@ class StateTree:
             else:
                 full_name = parent_full_name + '/' + state.short_name
 
-            # full_name = parent_full_name + '/' + state.short_name
-
-            state.gen = gen = StateGenerated(
-                state_id=next_id,
-                full_name=full_name,
-                ancestors=ancestors)
-
-            next_id += 1
-
-            self.state_dict[gen.full_name] = state
+            self.state_dict[full_name] = state
             self.state_list.append(state)
 
+            descendants = []
+            history = []
+            has_eventless_transitions = False
+            after_triggers = []
+
             for t in state.transitions:
                 self.transition_list.append(t)
                 if t.trigger is None:
-                    gen.has_eventless_transitions = True
+                    has_eventless_transitions = True
                 elif isinstance(t.trigger, AfterTrigger):
-                    gen.after_triggers.append(t.trigger)
+                    after_triggers.append(t.trigger)
 
             for c in state.children:
-                init_tree(c, gen.full_name, [state] + gen.ancestors)
+                init_tree(c, full_name, [state] + ancestors)
                 if isinstance(c, HistoryState):
-                    gen.history.append(c)
+                    history.append(c)
 
-            gen.descendants.extend(state.children)
+            descendants.extend(state.children)
             for c in state.children:
-                gen.descendants.extend(c.gen.descendants)
+                descendants.extend(c.gen.descendants)
 
-            for d in gen.descendants:
-                gen.descendant_bitmap |= bit(d.gen.state_id)
+            state.gen = StateGenerated(
+                state_id=state_id,
+                full_name=full_name,
+                ancestors=ancestors,
+                descendants=descendants,
+                descendant_bitmap=reduce(lambda x,y: x | bit(y.gen.state_id), descendants, Bitmap(0)),
+                history=history,
+                has_eventless_transitions=has_eventless_transitions,
+                after_triggers=after_triggers)
 
         init_tree(root, "", [])
         self.root = root
@@ -239,6 +206,7 @@ class StateTree:
                         if a in target.gen.ancestors:
                             lca = a
                             break
+
             t.gen = TransitionGenerated(
                 lca=lca,
                 arena_bitmap=lca.gen.descendant_bitmap.set(lca.gen.state_id))

src/sccd/test/test_parser.py

@@ -0,0 +1,74 @@
+import os
+import lxml.etree as etree
+from sccd.model.statechart_parser import *
+from sccd.test.test import *
+from copy import deepcopy
+
+
+class TestParser(StatechartParser):
+
+  def end_event(self, el):
+    big_step = self.require("big_step")
+    name = el.get("name")
+    port = el.get("port")
+    big_step.append(Event(id=0, name=name, port=port, parameters=[]))
+
+  def start_big_step(self, el):
+    self.require("test_output")
+    self.push("big_step", [])
+
+  def end_big_step(self, el):
+    output = self.require("test_output")
+    big_step = self.pop("big_step")
+    output.append(big_step)
+
+  def start_input(self, el):
+    self.require("test_input")
+
+  def end_input(self, el):
+    pass
+
+  def start_output(self, el):
+    self.require("test_output")
+
+  def end_output(self, el):
+    pass
+
+  def start_test(self, el):
+    self.push("context", Context(fixed_delta = None))
+    self.push("test_input", [])
+    self.push("test_output", [])
+    self.push("statecharts", [])
+
+  def end_test(self, el):
+    tests = self.require("tests")
+    src_file = self.require("src_file")
+
+    statecharts = self.pop("statecharts")
+    input = self.pop("test_input")
+    output = self.pop("test_output")
+    context = self.pop("context")
+
+    if len(statecharts) != 1:
+      raise Exception("Expected exactly 1 <statechart> node, got %d." % len(statecharts))
+    statechart = statecharts[0]
+
+    context.process_durations()
+
+    def variant_description(i, variant) -> str:
+      if not variant:
+        return ""
+      return " (variant %d: %s)" % (i, ", ".join(str(val) for val in variant.values()))
+
+    # Generate test variants for all semantic wildcards filled in
+    tests.extend( 
+      Test(
+        name=src_file + variant_description(i, variant),
+        model=SingleInstanceModel(
+          context,
+          Statechart(tree=statechart.tree, datamodel=deepcopy(statechart.datamodel), semantics=dataclasses.replace(statechart.semantics, **variant))),
+        input=input,
+        output=output)
+
+      for i, variant in enumerate(statechart.semantics.wildcard_cart_product())
+    )

src/sccd/test/xml_loader.py

@@ -1,109 +0,0 @@
-import os
-import lxml.etree as etree
-from lark import Lark, Transformer
-from sccd.test.test import *
-from sccd.model.model import *
-from sccd.model.xml_loader import *
-from sccd.syntax.statechart import *
-from sccd.util.debug import *
-from copy import deepcopy
-
-class PseudoSucceededTest(unittest.TestCase):
-  def __init__(self, name: str, msg):
-    super().__init__()
-    self.name = name
-    self.msg = msg
-
-  def __str__(self):
-    return self.name
-
-  def runTest(self):
-    print_debug(self.msg)
-
-class PseudoFailedTest(unittest.TestCase):
-  def __init__(self, name: str, e: Exception):
-    super().__init__()
-    self.name = name
-    self.e = e
-
-  def __str__(self):
-    return self.name
-
-  def runTest(self):
-    raise self.e
-
-# Returned list contains more than one test if the semantic configuration contains wildcard values.
-def load_test(src_file) -> List[Test]:
-  should_fail = os.path.basename(src_file).startswith("fail_")
-
-  context = Context(fixed_delta = None)
-
-  test_node = etree.parse(src_file).getroot()
-
-  try:
-    sc_node = test_node.find("statechart")
-    src = sc_node.get("src")
-    if src is None:
-      statechart = load_statechart(context, sc_node)
-    else:
-      external_file = os.path.join(os.path.dirname(src_file), src)
-      # print("loading", external_file, "...")
-      external_node = etree.parse(external_file).getroot()
-      statechart = load_statechart(context, external_node)
-      semantics_node = sc_node.find("override_semantics")
-      load_semantics(statechart.semantics, semantics_node)
-
-    input_node = test_node.find("input")
-    output_node = test_node.find("output")
-    input = load_input(input_node)
-    output = load_output(output_node)
-
-    context.process_durations()
-
-    def variant_description(i, variant) -> str:
-      if not variant:
-        return ""
-      return " (variant %d: %s)" % (i, ", ".join(str(val) for val in variant.values()))
-
-    if should_fail:
-      return [PseudoFailedTest(name=src_file, e=Exception("Unexpectedly succeeded at loading."))]
-    else:
-      return [
-        Test(
-          name=src_file + variant_description(i, variant),
-          model=SingleInstanceModel(
-            context,
-            Statechart(tree=statechart.tree, datamodel=deepcopy(statechart.datamodel), semantics=dataclasses.replace(statechart.semantics, **variant))),
-          input=input,
-          output=output)
-        for i, variant in enumerate(statechart.semantics.wildcard_cart_product())
-      ]
-
-  except Exception as e:
-    if should_fail:
-      return [PseudoSucceededTest(name=src_file, msg=str(e))]
-    else:
-      return [PseudoFailedTest(name=src_file, e=e)]
-
-def load_input(input_node) -> TestInput:
-  input = []
-  if input_node is not None:
-    for event_node in input_node:
-      name = event_node.get("name")
-      port = event_node.get("port")
-      time = int(event_node.get("time"))
-      input.append(InputEvent(name, port, [], time))
-  return input
-
-def load_output(output_node) -> TestOutput:
-  output = []
-  if output_node is not None: 
-    for big_step_node in output_node:
-      big_step = []
-      for event_node in big_step_node:
-        name = event_node.get("name")
-        port = event_node.get("port")
-        parameters = [] # todo: read params
-        big_step.append(Event(id=0, name=name, port=port, parameters=parameters))
-      output.append(big_step)
-  return output

test/test.py

@@ -1,27 +1,74 @@
 import argparse
 from lib.os_tools import *
-from sccd.test.xml_loader import *
+from sccd.test.test_parser import *
+from sccd.util.debug import *
+
+class PseudoSucceededTest(unittest.TestCase):
+  def __init__(self, name: str, msg):
+    super().__init__()
+    self.name = name
+    self.msg = msg
+
+  def __str__(self):
+    return self.name
+
+  def runTest(self):
+    print_debug(self.msg)
+
+class PseudoFailedTest(unittest.TestCase):
+  def __init__(self, name: str, e: Exception):
+    super().__init__()
+    self.name = name
+    self.e = e
+
+  def __str__(self):
+    return self.name
+
+  def runTest(self):
+    raise self.e
+
 
 if __name__ == '__main__':
-    parser = argparse.ArgumentParser(
+    argparser = argparse.ArgumentParser(
         description="Run SCCD tests.",
         epilog="Set environment variable SCCDDEBUG=1 to display debug information about the inner workings of the runtime.")
-    parser.add_argument('path', metavar='PATH', type=str, nargs='*', help="Tests to run. Can be a XML file or a directory. If a directory, it will be recursively scanned for XML files.")
-    parser.add_argument('--build-dir', metavar='BUILD_DIR', type=str, default='build', help="Directory for built tests. Defaults to 'build'")
-    args = parser.parse_args()
+    argparser.add_argument('path', metavar='PATH', type=str, nargs='*', help="Tests to run. Can be a XML file or a directory. If a directory, it will be recursively scanned for XML files.")
+    argparser.add_argument('--build-dir', metavar='BUILD_DIR', type=str, default='build', help="Directory for built tests. Defaults to 'build'")
+    args = argparser.parse_args()
 
     src_files = get_files(args.path,
         filter=lambda file: (file.startswith("test_") or file.startswith("fail_")) and file.endswith(".xml"))
 
-    suite = unittest.TestSuite()
-    for src_file in src_files:
-        tests = load_test(src_file)
-        for test in tests:
-            suite.addTest(test)
-
     if len(src_files) == 0:
         print("No input files specified.")
         print()
-        parser.print_usage()
-    else:
-        unittest.TextTestRunner(verbosity=2).run(suite)
+        argparser.print_usage()
+        exit()
+
+    suite = unittest.TestSuite()
+
+    for src_file in src_files:
+        parser = TestParser()
+        should_fail = os.path.basename(src_file).startswith("fail_")
+
+        try:
+            parser.push("src_file", src_file)
+            parser.push("tests", [])
+            statechart = parser.parse(etree.iterparse(src_file, events=("start", "end")))
+            tests = parser.pop("tests")
+            parser.pop("src_file")
+
+            if should_fail:
+                suite.addTest(PseudoFailedTest(name=src_file, e=Exception("Unexpectedly succeeded at loading.")))
+            else:
+                for test in tests:
+                    suite.addTest(test)
+
+        except Exception as e:
+            if should_fail:
+                suite.addTest(PseudoSucceededTest(name=src_file, msg=str(e)))
+            else:
+                suite.addTest(PseudoFailedTest(name=src_file, e=e))
+
+
+    unittest.TextTestRunner(verbosity=2).run(suite)