return module.name, port_map
-def convert(fragment, ports=[]):
- fragment._propagate_domains(ensure_sync_exists=True)
-
- # Clock domain reset always takes priority over all other logic. To ensure this, insert
- # decision trees for clock domain reset as the very last step before synthesis.
- fragment = xfrm.ResetInserter({
- cd.name: cd.rst for cd in fragment.domains.values() if cd.rst is not None
- })(fragment)
-
- ins, outs = fragment._propagate_ports(ports)
-
+def convert(fragment, name="top", **kwargs):
+ fragment = fragment.prepare(**kwargs)
builder = _Builder()
- convert_fragment(builder, fragment, name="top", top=True)
+ convert_fragment(builder, fragment, name=name, top=True)
return str(builder)
for signal in signals:
yield domain, signal
+ def iter_signals(self):
+ signals = ValueSet()
+ signals |= self.ports.keys()
+ for domain, domain_signals in self.drivers.items():
+ signals |= domain_signals
+ return signals
+
def add_domains(self, *domains):
for domain in domains:
assert isinstance(domain, ClockDomain)
subfrag._propagate_domains_down()
- def _propagate_domains(self, ensure_sync_exists=False):
+ def _propagate_domains(self, ensure_sync_exists):
self._propagate_domains_up()
if ensure_sync_exists and not self.domains:
self.add_domains(ClockDomain("sync"))
self._propagate_domains_down()
+ def _insert_domain_resets(self):
+ from .xfrm import ResetInserter
+
+ return ResetInserter({
+ cd.name: cd.rst for cd in self.domains.values() if cd.rst is not None
+ })(self)
+
def _propagate_ports(self, ports):
# Collect all signals we're driving (on LHS of statements), and signals we're using
# (on RHS of statements, or in clock domains).
self.add_ports(outs, kind="o")
return ins, outs
+
+ def prepare(self, ports=(), ensure_sync_exists=True):
+ from .xfrm import FragmentTransformer
+
+ fragment = FragmentTransformer()(self)
+ fragment._propagate_domains(ensure_sync_exists)
+ fragment = fragment._insert_domain_resets()
+ fragment._propagate_ports(ports)
+ return fragment
f1.add_domains(cd)
f1.add_subfragment(f2)
- f1._propagate_domains()
+ f1._propagate_domains(ensure_sync_exists=False)
self.assertEqual(f1.domains, {"cd": cd})
self.assertEqual(f2.domains, {"cd": cd})
- def test_propagate_default(self):
+ def test_propagate_ensure_sync(self):
f1 = Fragment()
f2 = Fragment()
f1.add_subfragment(f2)