Move statechart-specific functionality from action_lang to statechart module

src/sccd/action_lang/dynamic/memory.py

@@ -70,85 +70,3 @@ class Memory(MemoryInterface):
   def store(self, offset: int, value: Any):
     frame, offset = self._get_frame(offset)
     frame.storage[offset] = value
-
-  def flush_transition(self, read_only: bool = False):
-    pass
-
-  def flush_round(self):
-    pass
-
-class MemoryPartialSnapshot(MemoryInterface):
-
-  def __init__(self, description: str, memory: Memory, read_only: bool = False):
-    self.description = description
-    self.memory = memory
-    self.read_only = read_only
-
-    self.frame = memory.current_frame()
-
-    self.actual: List[Any] = self.frame.storage
-    self.snapshot: List[Any] = list(self.actual)
-
-    # Positions in stack frame written to by current transition.
-    self.trans_dirty = Bitmap()
-
-    # Positions in stack frame written to during current big, combo or small step (depending on semantic option chosen)
-    self.round_dirty = Bitmap()
-
-  def current_frame(self) -> StackFrame:
-    return self.memory.current_frame()
-
-  def push_frame(self, scope: Scope):
-    return self.memory.push_frame(scope)
-
-  def push_frame_w_context(self, scope: Scope, context: StackFrame):
-    return self.memory.push_frame_w_context(scope, context)
-
-  def pop_frame(self):
-    # The frame we are snapshotting should never be popped.
-    assert self.memory.current_frame() is not self.frame
-
-    return self.memory.pop_frame()
-
-  def load(self, offset: int) -> Any:
-    frame, offset = self.memory._get_frame(offset)
-    # Sometimes read from snapshot
-    if frame is self.frame:
-      # "our" frame! :)
-      if self.trans_dirty.has(offset):
-        return self.actual[offset]
-      else:
-        return self.snapshot[offset]
-    else:
-      return frame.storage[offset]
-
-  def store(self, offset: int, value: Any):
-    frame, offset = self.memory._get_frame(offset)
-    if frame is self.frame:
-      if self.read_only:
-        raise SCCDRuntimeException("Attempt to write to read-only %s memory." % self.description)
-      # "our" frame! :)
-      # Remember that we wrote, such that next read during same transition will be the value we wrote.
-      self.trans_dirty |= bit(offset)
-
-    # Always write to 'actual' storage
-    frame.storage[offset] = value
-
-
-  def flush_transition(self, read_only: bool = False):
-    race_conditions = self.trans_dirty & self.round_dirty
-    if race_conditions:
-      variables = self.frame.scope.variables
-      # some variable written to twice before refresh
-      raise SCCDRuntimeException("Race condition in %s memory: More than one transition assigned a new value to variables: %s" %
-          (self.description, ", ".join(variables[offset].name for offset in race_conditions.items())))
-
-    self.round_dirty |= self.trans_dirty
-    self.trans_dirty = Bitmap() # reset
-
-  def flush_round(self):
-    assert not self.trans_dirty # only allowed to be called right after flush_temp
-
-    # Probably quickest to just copy the entire list in Python
-    self.snapshot = list(self.actual) # refresh
-    self.round_dirty = Bitmap() # reset

src/sccd/action_lang/static/expression.py

@@ -30,15 +30,6 @@ class MemoryInterface(ABC):
   def store(self, offset: int, value: Any):
     pass
 
-  @abstractmethod
-  def flush_transition(self, read_only: bool = False):
-    pass
-
-  @abstractmethod
-  def flush_round(self):
-    pass
-
-
 # Thrown if the type checker encountered something illegal.
 # Not to be confused with Python's TypeError exception.
 class StaticTypeError(ModelError):

src/sccd/statechart/dynamic/memory_snapshot.py

@@ -0,0 +1,78 @@
+from sccd.action_lang.dynamic.memory import *
+
+
+class MemoryPartialSnapshot(MemoryInterface):
+
+  def __init__(self, description: str, memory: Memory, read_only: bool = False):
+    self.description = description
+    self.memory = memory
+    self.read_only = read_only
+
+    self.frame = memory.current_frame()
+
+    self.actual: List[Any] = self.frame.storage
+    self.snapshot: List[Any] = list(self.actual)
+
+    # Positions in stack frame written to by current transition.
+    self.trans_dirty = Bitmap()
+
+    # Positions in stack frame written to during current big, combo or small step (depending on semantic option chosen)
+    self.round_dirty = Bitmap()
+
+  def current_frame(self) -> StackFrame:
+    return self.memory.current_frame()
+
+  def push_frame(self, scope: Scope):
+    return self.memory.push_frame(scope)
+
+  def push_frame_w_context(self, scope: Scope, context: StackFrame):
+    return self.memory.push_frame_w_context(scope, context)
+
+  def pop_frame(self):
+    # The frame we are snapshotting should never be popped.
+    assert self.memory.current_frame() is not self.frame
+
+    return self.memory.pop_frame()
+
+  def load(self, offset: int) -> Any:
+    frame, offset = self.memory._get_frame(offset)
+    # Sometimes read from snapshot
+    if frame is self.frame:
+      # "our" frame! :)
+      if self.trans_dirty.has(offset):
+        return self.actual[offset]
+      else:
+        return self.snapshot[offset]
+    else:
+      return frame.storage[offset]
+
+  def store(self, offset: int, value: Any):
+    frame, offset = self.memory._get_frame(offset)
+    if frame is self.frame:
+      if self.read_only:
+        raise SCCDRuntimeException("Attempt to write to read-only %s memory." % self.description)
+      # "our" frame! :)
+      # Remember that we wrote, such that next read during same transition will be the value we wrote.
+      self.trans_dirty |= bit(offset)
+
+    # Always write to 'actual' storage
+    frame.storage[offset] = value
+
+
+  def flush_transition(self, read_only: bool = False):
+    race_conditions = self.trans_dirty & self.round_dirty
+    if race_conditions:
+      variables = self.frame.scope.variables
+      # some variable written to twice before refresh
+      raise SCCDRuntimeException("Race condition in %s memory: More than one transition assigned a new value to variables: %s" %
+          (self.description, ", ".join(variables[offset].name for offset in race_conditions.items())))
+
+    self.round_dirty |= self.trans_dirty
+    self.trans_dirty = Bitmap() # reset
+
+  def flush_round(self):
+    assert not self.trans_dirty # only allowed to be called right after flush_temp
+
+    # Probably quickest to just copy the entire list in Python
+    self.snapshot = list(self.actual) # refresh
+    self.round_dirty = Bitmap() # reset

src/sccd/statechart/dynamic/statechart_instance.py

@@ -1,13 +1,13 @@
 import termcolor
 import functools
 from typing import List, Tuple, Iterable
-from sccd.statechart.dynamic.builtin_scope import *
-from sccd.statechart.static.statechart import *
 from sccd.util.debug import print_debug
 from sccd.util.bitmap import *
+from sccd.statechart.static.statechart import *
+from sccd.statechart.dynamic.builtin_scope import *
 from sccd.statechart.dynamic.round import *
 from sccd.statechart.dynamic.statechart_execution import *
-from sccd.action_lang.dynamic.memory import *
+from sccd.statechart.dynamic.memory_snapshot import *
 
 # Interface for all instances and also the Object Manager
 class Instance(ABC):
@@ -93,7 +93,6 @@ class StatechartInstance(Instance):
         raise_nextbs = lambda e, time_offset: self.execution.output.append(OutputEvent(e, InstancesTarget([self]), time_offset))
 
         raise_internal = {
-            # InternalEventLifeline.QUEUE: self._big_step.add_next_event,
             InternalEventLifeline.QUEUE: lambda e: raise_nextbs(e, 0),
             InternalEventLifeline.NEXT_COMBO_STEP: combo_step.add_next_event,
             InternalEventLifeline.NEXT_SMALL_STEP: small_step.add_next_event,