Ivy

amaranth/sim/_base.py

@@ -1,4 +1,27 @@
-__all__ = ["BaseProcess", "BaseSignalState", "BaseMemoryState", "BaseEngineState", "BaseEngine"]
+from abc import ABCMeta, abstractmethod
+
+__all__ = ["BaseProcess", "BaseSignalState", "BaseMemoryState", "BaseEngineState", "BaseEngine", "Observer", "DummyEngine", "PrintObserver"]
+
+
+class Observer(metaclass=ABCMeta):
+    def __init__(self, fs_per_delta=0):
+        self._fs_per_delta = fs_per_delta
+
+    @property
+    def fs_per_delta(self) -> int:
+        return self._fs_per_delta
+
+    @abstractmethod
+    def update_signal(self, timestamp, signal):
+       ...
+
+    @abstractmethod
+    def update_memory(self, timestamp, memory, addr):
+        ...
+
+    @abstractmethod
+    def close(self, timestamp):
+       assert False
 
 
 class BaseProcess:
@@ -62,6 +85,26 @@ def add_memory_waker(self, memory, waker):
 
 
 class BaseEngine:
+    # add storage for observers
+    def __init__(self):
+        self._observers = []
+
+    # append observer to list
+    def add_observer(self, observer: Observer):
+        self._observers.append(observer)
+
+    def notify_signal_change(self, signal):
+        for observer in self._observers:
+            observer.update_signal(self.now, signal)
+
+    def notify_memory_change(self, memory, addr):
+        for observer in self._observers:
+            observer.update_memory(self.now, memory, addr)
+
+    def notify_close(self):
+        for observer in self._observers:
+            observer.close(self.now)
+
     @property
     def state(self) -> BaseEngineState:
         raise NotImplementedError # :nocov:
@@ -97,5 +140,45 @@ def step_design(self):
     def advance(self):
         raise NotImplementedError # :nocov:
 
-    def write_vcd(self, *, vcd_file, gtkw_file, traces, fs_per_delta):
+    def observe(self, observer: Observer):
         raise NotImplementedError # :nocov:
+
+class DummyEngine(BaseEngine):
+    def __init__(self):
+        super().__init__()
+        self._now = 0  
+
+    @property
+    def now(self):
+        return self._now
+
+    def notify_signal_change(self, signal):
+        for obs in self._observers:
+            obs.update_signal(self.now, signal)
+
+    def notify_memory_change(self, memory, addr):
+        for obs in self._observers:
+            obs.update_memory(self.now, memory, addr)
+
+    def notify_close(self):
+        for obs in self._observers:
+            obs.close(self.now)
+
+
+class PrintObserver(Observer):
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @property
+    def fs_per_delta(self) -> int:
+        return 1
+
+    def update_signal(self, timestamp, signal):
+        print(f"[{timestamp}] Signal changed: {signal}")
+
+    def update_memory(self, timestamp, memory, addr):
+        print(f"[{timestamp}] Memory write at {addr}")
+
+    def close(self, timestamp):
+        print(f"[{timestamp}] Simulation ended")

amaranth/sim/_coverage.py

@@ -0,0 +1,59 @@
+from ._base import Observer
+from amaranth.sim._vcdwriter import eval_value, eval_format
+
+class ToggleCoverageObserver(Observer):
+    def __init__(self, state, **kwargs):
+        self.state = state
+        self._prev_values = {}
+        self._toggles = {}
+        self._signal_names = {}
+        super().__init__(**kwargs)
+
+    def update_signal(self, timestamp, signal):
+        if getattr(signal, "name", "") != "out":
+            return
+
+        sig_id = id(signal)
+        try:
+            val = eval_value(self.state, signal)
+        except Exception:
+            val = int(self.state.get_signal(signal))
+        try:
+            curr_val = int(val)
+        except TypeError:
+            curr_val = val
+        print(f"[DEBUG] Signal {getattr(signal, 'name', signal)} = {curr_val}")
+
+        if sig_id not in self._prev_values:
+            self._prev_values[sig_id] = curr_val
+            self._toggles[sig_id] = {"0->1": 0, "1->0": 0}
+            self._signal_names[sig_id] = signal.name  
+            return
+
+        prev_val = self._prev_values[sig_id]
+
+        if prev_val == 0 and curr_val == 1:
+            self._toggles[sig_id]["0->1"] += 1
+        elif prev_val == 1 and curr_val == 0:
+            self._toggles[sig_id]["1->0"] += 1
+
+        self._prev_values[sig_id] = curr_val
+
+    def update_memory(self, timestamp, memory, addr):
+        pass 
+
+    def get_results(self):
+        return {
+            self._signal_names[sig_id]: toggles
+            for sig_id, toggles in self._toggles.items()
+        }
+
+    def close(self, timestamp):
+        results = self.get_results()
+        print("=== Toggle Coverage Report ===")
+        for signal, toggles in results.items():
+            print(f"{signal}: 0→1={toggles['0->1']}, 1→0={toggles['1->0']}")
+
+
+
+