Ivy

1 changed from string to enum

@@ -0,0 +1,3 @@
+[ivy 58ebf99] fixed reviewer's comments, multi-bit toggle DUT works, 0-
+ 4 files changed, 69 insertions(+), 44 deletions(-)
+ create mode 100644 .coverage.Ivys-MacBook-Air.local.61006.XgklDXGx

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,95 @@
+from enum import Enum, auto
+from ._base import Observer
+from amaranth.sim._vcdwriter import eval_value, eval_format
+
+class ToggleDirection(Enum):
+    ZERO_TO_ONE = auto()
+    ONE_TO_ZERO = auto()
+
+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 (KeyError, AttributeError): 
+            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] = {
+                i: {ToggleDirection.ZERO_TO_ONE: 0, ToggleDirection.ONE_TO_ZERO: 0}
+                for i in range(signal.shape().width)
+            }
+            self._signal_names[sig_id] = signal.name  
+            return
+
+        prev_val = self._prev_values[sig_id]
+
+        for bit in range(signal.shape().width):
+            prev_bit = (prev_val >> bit) & 1
+            curr_bit = (curr_val >> bit) & 1
+            if prev_bit == 0 and curr_bit == 1:
+                self._toggles[sig_id][bit][ToggleDirection.ZERO_TO_ONE] += 1
+            elif prev_bit == 1 and curr_bit == 0:
+                self._toggles[sig_id][bit][ToggleDirection.ONE_TO_ZERO] += 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, bit_toggles in results.items():
+            print(f"{signal}:")
+            for bit, counts in bit_toggles.items():
+                print(f"  Bit {bit}: 0→1={counts[ToggleDirection.ZERO_TO_ONE]}, 1→0={counts[ToggleDirection.ONE_TO_ZERO]}")
+
+
+
+
+class StatementCoverageObserver(Observer):
+    def __init__(self, **kwargs):
+        self._statement_hits = {}
+        super().__init__(**kwargs)
+
+    def record_statement_hit(self, statement_id: str):
+        if statement_id not in self._statement_hits:
+            self._statement_hits[statement_id] = 0
+        self._statement_hits[statement_id] += 1
+
+    def update_signal(self, timestamp, signal):
+        pass
+
+    def update_memory(self, timestamp, memory, addr):
+        pass
+
+    def get_result(self):
+        return self._statement_hits
+
+    def close(self, timestamp):
+        print("=== Statement Coverage Report ===")
+        for stmt_id, count in sorted(self._statement_hits.items()):
+            print(f"{stmt_id}:{'HIT' if count > 0 else 'MISS'} ({count} times)")