hdl._ir: remove Fragment.drivers.

Author: wanda-phi

amaranth/hdl/_dsl.py

@@ -675,15 +675,7 @@ def elaborate(self, platform):
         for domain, statements in self._statements.items():
             statements = resolve_statements(statements)
             fragment.add_statements(domain, statements)
-            visitor = _Visitor()
-            visitor.visit_stmt(statements)
-            for signal in visitor.driven_signals:
-                fragment.add_driver(signal, domain)
         fragment.add_statements("comb", self._top_comb_statements)
-        visitor = _Visitor()
-        visitor.visit_stmt(self._top_comb_statements)
-        for signal in visitor.driven_signals:
-            fragment.add_driver(signal, "comb")
         fragment.add_domains(self._domains.values())
         fragment.generated.update(self._generated)
         return fragment

amaranth/hdl/_ir.py

@@ -63,7 +63,6 @@ def get(obj, platform):
             obj = new_obj
 
     def __init__(self, *, src_loc=None):
-        self.drivers = OrderedDict()
         self.statements = {}
         self.domains = OrderedDict()
         self.subfragments = []
@@ -73,28 +72,6 @@ def __init__(self, *, src_loc=None):
         self.origins = None
         self.domains_propagated_up = {}
 
-    def add_driver(self, signal, domain="comb"):
-        assert isinstance(domain, str)
-        if domain not in self.drivers:
-            self.drivers[domain] = _ast.SignalSet()
-        self.drivers[domain].add(signal)
-
-    def iter_drivers(self):
-        for domain, signals in self.drivers.items():
-            for signal in signals:
-                yield domain, signal
-
-    def iter_comb(self):
-        if "comb" in self.drivers:
-            yield from self.drivers["comb"]
-
-    def iter_sync(self):
-        for domain, signals in self.drivers.items():
-            if domain == "comb":
-                continue
-            for signal in signals:
-                yield domain, signal
-
     def add_domains(self, *domains):
         for domain in flatten(domains):
             assert isinstance(domain, _cd.ClockDomain)

amaranth/hdl/_mem.py

@@ -221,8 +221,6 @@ def read_port(self, *, domain, addr, data, en, transparent_for):
             assert isinstance(idx, int)
             assert idx in range(len(self._write_ports))
             assert self._write_ports[idx]._domain == port._domain
-        for signal in port._data._rhs_signals():
-            self.add_driver(signal, port._domain)
         self._read_ports.append(port)
 
     def write_port(self, *, domain, addr, data, en):

amaranth/hdl/_xfrm.py

@@ -252,10 +252,6 @@ def map_statements(self, fragment, new_fragment):
             for domain, statements in fragment.statements.items():
                 new_fragment.add_statements(domain, statements)
 
-    def map_drivers(self, fragment, new_fragment):
-        for domain, signal in fragment.iter_drivers():
-            new_fragment.add_driver(signal, domain)
-
     def map_memory_ports(self, fragment, new_fragment):
         if hasattr(self, "on_value"):
             for port in new_fragment._read_ports:
@@ -322,7 +318,6 @@ def on_fragment(self, fragment):
         self.map_subfragments(fragment, new_fragment)
         self.map_domains(fragment, new_fragment)
         self.map_statements(fragment, new_fragment)
-        self.map_drivers(fragment, new_fragment)
         return new_fragment
 
     def __call__(self, value, *, src_loc_at=0):
@@ -518,13 +513,6 @@ def map_statements(self, fragment, new_fragment):
                 map(self.on_statement, statements)
             )
 
-    def map_drivers(self, fragment, new_fragment):
-        for domain, signals in fragment.drivers.items():
-            if domain in self.domain_map:
-                domain = self.domain_map[domain]
-            for signal in signals:
-                new_fragment.add_driver(self.on_value(signal), domain)
-
     def map_memory_ports(self, fragment, new_fragment):
         super().map_memory_ports(fragment, new_fragment)
         for port in new_fragment._read_ports:
@@ -560,10 +548,6 @@ def _resolve(self, domain, context):
         self._warn_on_propagation_up(domain, context.src_loc)
         return self.domains[domain]
 
-    def map_drivers(self, fragment, new_fragment):
-        for domain, signal in fragment.iter_drivers():
-            new_fragment.add_driver(self.on_value(signal), domain)
-
     def replace_value_src_loc(self, value, new_value):
         return not isinstance(value, (ClockSignal, ResetSignal))
 
@@ -605,9 +589,12 @@ def __init__(self, controls):
 
     def on_fragment(self, fragment):
         new_fragment = super().on_fragment(fragment)
-        for domain, signals in fragment.drivers.items():
+        for domain, statements in fragment.statements.items():
             if domain == "comb" or domain not in self.controls:
                 continue
+            signals = SignalSet()
+            for stmt in statements:
+                signals |= stmt._lhs_signals()
             self._insert_control(new_fragment, domain, signals)
         return new_fragment
 

tests/test_hdl_dsl.py

@@ -631,12 +631,6 @@ def test_FSM_basic(self):
             )
         )
         """)
-        self.assertEqual({repr(sig): k for k, v in frag.drivers.items() for sig in v}, {
-            "(sig a)": "comb",
-            "(sig fsm_state)": "sync",
-            "(sig b)": "sync",
-            "(sig)": "comb",
-        })
         fsm  = frag.find_generated("fsm")
         self.assertIsInstance(fsm.state, Signal)
         self.assertEqual(fsm.encoding, OrderedDict({
@@ -960,9 +954,6 @@ def test_lower(self):
             (eq (sig c1) (sig s1))
         )
         """)
-        self.assertEqual(f1.drivers, {
-            "comb": SignalSet((self.c1,))
-        })
         self.assertEqual(len(f1.subfragments), 1)
         (f2, f2_name, _), = f1.subfragments
         self.assertEqual(f2_name, "foo")
@@ -976,8 +967,4 @@ def test_lower(self):
             (eq (sig c3) (sig s3))
         )
         """)
-        self.assertEqual(f2.drivers, {
-            "comb": SignalSet((self.c2,)),
-            "sync": SignalSet((self.c3,))
-        })
         self.assertEqual(len(f2.subfragments), 0)