Repository: amaranth-lang/amaranth
Pull request #1274 (2024-04-03 17:33:06)

Implement RFC 58: Core support for `ValueCastable` formatting.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang) Report
Attention: Patch coverage is `85.71429%` with `2 lines` in your changes are missing coverage. Please review.
> Project coverage is 89.62%. Comparing base [(`f71bee4`)](https://app.codecov.io/gh/amaranth-lang/amaranth/commit/f71bee499d5fef2151dd88abd98b17b94e6791e8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang) to head [(`0c88464`)](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang).
> Report is 2 commits behind head on main.

| [Files](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang) | Patch % | Lines |
|---|---|---|
| [amaranth/hdl/\_ast.py](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?src=pr&el=tree&filepath=amaranth%2Fhdl%2F_ast.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang#diff-YW1hcmFudGgvaGRsL19hc3QucHk=) | 85.71% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #1274      +/-   ##
==========================================
- Coverage   89.88%   89.62%   -0.27%     
==========================================
  Files          44       44              
  Lines       10325    10429     +104     
  Branches     2510     2551      +41     
==========================================
+ Hits         9281     9347      +66     
- Misses        844      872      +28     
- Partials      200      210      +10     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/amaranth-lang/amaranth/pull/1274?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=amaranth-lang).


Patch:
diff --git a/amaranth/hdl/_ast.py b/amaranth/hdl/_ast.py
index d60e6af..a0d4ebe 100644
--- a/amaranth/hdl/_ast.py
+++ b/amaranth/hdl/_ast.py
@@ -385,6 +385,27 @@ class ShapeCastable:
         """
         return super().__call__(*args, **kwargs) # :nocov:
 
+    def format(self, obj, spec):
+        """Format a value.
+
+        This method is called by the Amaranth language to implement formatting for custom
+        shapes. Whenever :py:`"{obj:spec}"` is encountered by :class:`Format`, and :py:`obj`
+        has a custom shape that has a :meth:`format` method, :py:`obj.shape().format(obj, "spec")`
+        is called, and the format specifier is replaced with the result.
+
+        The default :meth:`format` implementation is:
+
+        .. code::
+
+            def format(self, obj, spec):
+                return Format(f"{{:{spec}}}", Value.cast(obj))
+
+        Returns
+        -------
+        :class:`Format`
+        """
+        return Format(f"{{:{spec}}}", Value.cast(obj))
+
     # TODO: write an RFC for turning this into a proper interface method
     def _value_repr(self, value):
         return (_repr.Repr(_repr.FormatInt(), value),)
@@ -2576,14 +2597,26 @@ class Format:
             chunks.append(literal)
             if field_name is not None:
                 obj = get_field(field_name)
-                obj = fmt.convert_field(obj, conversion)
+                if conversion == "v":
+                    obj = Value.cast(obj)
+                else:
+                    obj = fmt.convert_field(obj, conversion)
                 format_spec = subformat(format_spec)
                 if isinstance(obj, Value):
                     # Perform validation.
                     self._parse_format_spec(format_spec, obj.shape())
                     chunks.append((obj, format_spec))
                 elif isinstance(obj, ValueCastable):
-                    raise TypeError("'ValueCastable' formatting is not supported")
+                    shape = obj.shape()
+                    if isinstance(shape, ShapeCastable):
+                        fmt = shape.format(obj, format_spec)
+                        if not isinstance(fmt, Format):
+                            raise TypeError(f"`ShapeCastable.format` must return a 'Format' instance, not {fmt!r}")
+                        chunks += fmt._chunks
+                    else:
+                        obj = Value.cast(obj)
+                        self._parse_format_spec(format_spec, obj.shape())
+                        chunks.append((obj, format_spec))
                 elif isinstance(obj, Format):
                     if format_spec != "":
                         raise ValueError(f"Format specifiers ({format_spec!r}) cannot be used for 'Format' objects")
diff --git a/amaranth/hdl/_ir.py b/amaranth/hdl/_ir.py
index 5046232..9f797ba 100644
--- a/amaranth/hdl/_ir.py
+++ b/amaranth/hdl/_ir.py
@@ -669,12 +669,14 @@ class NetlistDriver:
 
 
 class NetlistEmitter:
-    def __init__(self, netlist: _nir.Netlist, design):
+    def __init__(self, netlist: _nir.Netlist, design, *, all_undef_to_ff=False):
         self.netlist = netlist
         self.design = design
+        self.all_undef_to_ff = all_undef_to_ff
         self.drivers = _ast.SignalDict()
         self.io_ports: dict[_ast.IOPort, int] = {}
         self.rhs_cache: dict[int, Tuple[_nir.Value, bool, _ast.Value]] = {}
+        self.fragment_module_idx: dict[Fragment, int] = {}
 
         # Collected for driver conflict diagnostics only.
         self.late_net_to_signal = {}
@@ -1358,6 +1360,39 @@ class NetlistEmitter:
                 src_loc = driver.signal.src_loc
             self.connect(self.emit_signal(driver.signal), value, src_loc=src_loc)
 
+    def emit_undef_ff(self):
+        # Connect all completely undriven signals to flip-flops with const-0 clock. This is used
+        # for simulation targets, so that undriven signals have allocated storage that can be
+        # used by the testbench to drive them, instead of being hardwired to the init value
+        # constant.
+        for signal, value in self.netlist.signals.items():
+            fragment = self.design.signal_lca[signal]
+            module_idx = self.fragment_module_idx[fragment]
+            pos = 0
+            while pos < len(signal):
+                net = value[pos]
+                if not net.is_late or net in self.netlist.connections:
+                    pos += 1
+                else:
+                    end_pos = pos
+                    while (end_pos < len(signal) and
+                            value[end_pos].is_late and
+                            value[end_pos] not in self.netlist.connections):
+                        end_pos += 1
+                    init = (signal.init >> pos) & ((1 << (end_pos - pos)) - 1)
+                    cell = _nir.FlipFlop(module_idx,
+                        data=value[pos:end_pos],
+                        init=init,
+                        clk=_nir.Net.from_const(0),
+                        clk_edge="pos",
+                        arst=_nir.Net.from_const(0),
+                        attributes={},
+                        src_loc=signal.src_loc,
+                    )
+                    ff_value = self.netlist.add_value_cell(end_pos - pos, cell)
+                    self.connect(value[pos:end_pos], ff_value, src_loc=signal.src_loc)
+                    pos = end_pos
+
     def emit_undriven(self):
         # Connect all undriven signal bits to their initial values. This can only happen for entirely
         # undriven signals, or signals that are partially driven by instances.
@@ -1373,6 +1408,7 @@ class NetlistEmitter:
         if isinstance(fragment, _ir.Instance):
             assert parent_module_idx is not None
             self.emit_instance(parent_module_idx, fragment, name=fragment_name[-1])
+            self.fragment_module_idx[fragment] = parent_module_idx
         elif isinstance(fragment, _mem.MemoryInstance):
             assert parent_module_idx is not None
             memory = self.emit_memory(parent_module_idx, fragment, name=fragment_name[-1])
@@ -1381,11 +1417,14 @@ class NetlistEmitter:
                 write_ports.append(self.emit_write_port(parent_module_idx, fragment, port, memory))
             for port in fragment._read_ports:
                 self.emit_read_port(parent_module_idx, fragment, port, memory, write_ports)
+            self.fragment_module_idx[fragment] = parent_module_idx
         elif isinstance(fragment, _ir.IOBufferInstance):
             assert parent_module_idx is not None
             self.emit_iobuffer(parent_module_idx, fragment)
+            self.fragment_module_idx[fragment] = parent_module_idx
         elif type(fragment) is _ir.Fragment:
             module_idx = self.netlist.add_module(parent_module_idx, fragment_name, src_loc=fragment.src_loc, cell_src_loc=cell_src_loc)
+            self.fragment_module_idx[fragment] = module_idx
             signal_names = self.design.fragments[fragment].signal_names
             self.netlist.modules[module_idx].signal_names = signal_names
             io_port_names = self.design.fragments[fragment].io_port_names
@@ -1402,13 +1441,15 @@ class NetlistEmitter:
             if parent_module_idx is None:
                 self.emit_drivers()
                 self.emit_top_ports(fragment)
+                if self.all_undef_to_ff:
+                    self.emit_undef_ff()
                 self.emit_undriven()
         else:
             assert False # :nocov:
 
 
-def _emit_netlist(netlist: _nir.Netlist, design):
-    NetlistEmitter(netlist, design).emit_fragment(design.fragment, None)
+def _emit_netlist(netlist: _nir.Netlist, design, *, all_undef_to_ff=False):
+    NetlistEmitter(netlist, design, all_undef_to_ff=all_undef_to_ff).emit_fragment(design.fragment, None)
 
 
 def _compute_net_flows(netlist: _nir.Netlist):
@@ -1640,13 +1681,13 @@ def _compute_io_ports(netlist: _nir.Netlist, ports):
                 visited.update(value)
 
 
-def build_netlist(fragment, ports=(), *, name="top", **kwargs):
+def build_netlist(fragment, ports=(), *, name="top", all_undef_to_ff=False, **kwargs):
     if isinstance(fragment, Design):
         design = fragment
     else:
         design = fragment.prepare(ports=ports, hierarchy=(name,), **kwargs)
     netlist = _nir.Netlist()
-    _emit_netlist(netlist, design)
+    _emit_netlist(netlist, design, all_undef_to_ff=all_undef_to_ff)
     netlist.resolve_all_nets()
     _compute_net_flows(netlist)
     _compute_ports(netlist)
diff --git a/docs/changes.rst b/docs/changes.rst
index b289733..c22dadc 100644
--- a/docs/changes.rst
+++ b/docs/changes.rst
@@ -56,6 +56,7 @@ Implemented RFCs
 .. _RFC 51: https://amaranth-lang.org/rfcs/0051-const-from-bits.html
 .. _RFC 53: https://amaranth-lang.org/rfcs/0053-ioport.html
 .. _RFC 55: https://amaranth-lang.org/rfcs/0055-lib-io.html
+.. _RFC 58: https://amaranth-lang.org/rfcs/0058-valuecastable-format.html
 .. _RFC 59: https://amaranth-lang.org/rfcs/0059-no-domain-upwards-propagation.html
 .. _RFC 62: https://amaranth-lang.org/rfcs/0062-memory-data.html
 
@@ -68,6 +69,7 @@ Implemented RFCs
 * `RFC 50`_: ``Print`` statement and string formatting
 * `RFC 51`_: Add ``ShapeCastable.from_bits`` and ``amaranth.lib.data.Const``
 * `RFC 53`_: Low-level I/O primitives
+* `RFC 58`_: Core support for ``ValueCastable`` formatting
 * `RFC 59`_: Get rid of upwards propagation of clock domains
 * `RFC 62`_: The `MemoryData`` class
 


Test patch:
diff --git a/tests/test_hdl_ast.py b/tests/test_hdl_ast.py
index 583de3e..1e9737d 100644
--- a/tests/test_hdl_ast.py
+++ b/tests/test_hdl_ast.py
@@ -1412,6 +1412,79 @@ class MockValueCastable(ValueCastable):
         return self.dest
 
 
+class MockShapeCastableFormat(ShapeCastable):
+    def __init__(self, dest):
+        self.dest = dest
+
+    def as_shape(self):
+        return self.dest
+
+    def __call__(self, value):
+        return value
+
+    def const(self, init):
+        return Const(init, self.dest)
+
+    def from_bits(self, bits):
+        return bits
+
+    def format(self, value, format_desc):
+        return Format("_{}_{}_", Value.cast(value), format_desc)
+
+
+class MockValueCastableFormat(ValueCastable):
+    def __init__(self, dest):
+        self.dest = dest
+
+    def shape(self):
+        return MockShapeCastableFormat(Value.cast(self.dest).shape())
+
+    def as_value(self):
+        return self.dest
+
+
+class MockValueCastableNoFormat(ValueCastable):
+    def __init__(self, dest):
+        self.dest = dest
+
+    def shape(self):
+        return MockShapeCastable(Value.cast(self.dest).shape())
+
+    def as_value(self):
+        return self.dest
+
+
+class MockShapeCastableFormatWrong(ShapeCastable):
+    def __init__(self, dest):
+        self.dest = dest
+
+    def as_shape(self):
+        return self.dest
+
+    def __call__(self, value):
+        return value
+
+    def const(self, init):
+        return Const(init, self.dest)
+
+    def from_bits(self, bits):
+        return bits
+
+    def format(self, value, format_desc):
+        return Value.cast(value)
+
+
+class MockValueCastableFormatWrong(ValueCastable):
+    def __init__(self, dest):
+        self.dest = dest
+
+    def shape(self):
+        return MockShapeCastableFormatWrong(Value.cast(self.dest).shape())
+
+    def as_value(self):
+        return self.dest
+
+
 class MockValueCastableChanges(ValueCastable):
     def __init__(self, width=0):
         self.width = width
@@ -1567,6 +1640,21 @@ class FormatTestCase(FHDLTestCase):
         fmt = Format("sub: {}, c: {:4x}", subfmt, c)
         self.assertRepr(fmt, "(format 'sub: a: {:2x}, b: {:3x}, c: {:4x}' (sig a) (sig b) (sig c))")
 
+    def test_construct_valuecastable(self):
+        a = Signal()
+        b = MockValueCastable(a)
+        fmt = Format("{:x}", b)
+        self.assertRepr(fmt, "(format '{:x}' (sig a))")
+        c = MockValueCastableFormat(a)
+        fmt = Format("{:meow}", c)
+        self.assertRepr(fmt, "(format '_{}_meow_' (sig a))")
+        d = MockValueCastableNoFormat(a)
+        fmt = Format("{:x}", d)
+        self.assertRepr(fmt, "(format '{:x}' (sig a))")
+        e = MockValueCastableFormat(a)
+        fmt = Format("{!v:x}", e)
+        self.assertRepr(fmt, "(format '{:x}' (sig a))")
+
     def test_construct_wrong(self):
         a = Signal()
         b = Signal(signed(16))
@@ -1576,9 +1664,6 @@ class FormatTestCase(FHDLTestCase):
         with self.assertRaisesRegex(ValueError,
                 r"^cannot switch from automatic field numbering to manual field specification$"):
             Format("{}, {1}", a, b)
-        with self.assertRaisesRegex(TypeError,
-                r"^'ValueCastable' formatting is not supported$"):
-            Format("{}", MockValueCastable(Const(0)))
         with self.assertRaisesRegex(ValueError,
                 r"^Format specifiers \('s'\) cannot be used for 'Format' objects$"):
             Format("{:s}", Format(""))
@@ -1622,6 +1707,14 @@ class FormatTestCase(FHDLTestCase):
                 r"^Cannot specify '_' with format specifier 'c'$"):
             Format("{a:_c}", a=a)
 
+    def test_construct_valuecastable_wrong(self):
+        a = Signal()
+        b = MockValueCastableFormatWrong(a)
+        with self.assertRaisesRegex(TypeError,
+                r"^`ShapeCastable.format` must return a 'Format' instance, "
+                r"not \(sig a\)$"):
+            fmt = Format("{:x}", b)
+
     def test_plus(self):
         a = Signal()
         b = Signal()
diff --git a/tests/test_hdl_ir.py b/tests/test_hdl_ir.py
index f54bc08..5a25320 100644
--- a/tests/test_hdl_ir.py
+++ b/tests/test_hdl_ir.py
@@ -460,6 +460,7 @@ class FragmentPortsTestCase(FHDLTestCase):
                     r" \(did you mean `ports=\(<signal>,\)`, rather than `ports=<signal>`\?\)$")):
             build_netlist(f, ports=Const(1))
 
+
 class FragmentDomainsTestCase(FHDLTestCase):
     def test_propagate_up(self):
         cd = ClockDomain()
@@ -981,6 +982,7 @@ class InstanceTestCase(FHDLTestCase):
         )
         """)
 
+
 class NamesTestCase(FHDLTestCase):
     def test_assign_names_to_signals(self):
         i = Signal()
@@ -1989,6 +1991,7 @@ class AssignTestCase(FHDLTestCase):
         )
         """)
 
+
 class RhsTestCase(FHDLTestCase):
     def test_const(self):
         o1 = Signal(8)
@@ -3173,6 +3176,7 @@ class RhsTestCase(FHDLTestCase):
         )
         """)
 
+
 class SwitchTestCase(FHDLTestCase):
     def test_comb(self):
         o1 = Signal(8)
@@ -3436,6 +3440,7 @@ class SwitchTestCase(FHDLTestCase):
         )
         """)
 
+
 class ConflictTestCase(FHDLTestCase):
     def test_domain_conflict(self):
         s = Signal()
@@ -3472,3 +3477,61 @@ class ConflictTestCase(FHDLTestCase):
                 r"^Bit 0 of signal \(sig s\) has multiple drivers: "
                 r".*test_hdl_ir.py:\d+ and .*test_hdl_ir.py:\d+$"):
             build_netlist(Fragment.get(m, None), [])
+
+
+class UndrivenTestCase(FHDLTestCase):
+    def test_undriven(self):
+        o = Signal(8)
+        m = Module()
+        nl = build_netlist(Fragment.get(m, None), [
+            ("o", o, PortDirection.Output),
+        ])
+        self.assertRepr(nl, """
+        (
+            (module 0 None ('top')
+                (output 'o' 8'd0)
+            )
+            (cell 0 0 (top
+                (output 'o' 8'd0)
+            ))
+        )
+        """)
+
+    def test_undef_to_ff(self):
+        o = Signal(8, init=0x55)
+        m = Module()
+        nl = build_netlist(Fragment.get(m, None), [
+            ("o", o, PortDirection.Output),
+        ], all_undef_to_ff=True)
+        self.assertRepr(nl, """
+        (
+            (module 0 None ('top')
+                (output 'o' 1.0:8)
+            )
+            (cell 0 0 (top
+                (output 'o' 1.0:8)
+            ))
+            (cell 1 0 (flipflop 1.0:8 85 pos 0 0))
+        )
+        """)
+
+    def test_undef_to_ff_partial(self):
+        o = Signal(8, init=0x55)
+        m = Module()
+        m.submodules.inst = Instance("t", o_o=o[2])
+        nl = build_netlist(Fragment.get(m, None), [
+            ("o", o, PortDirection.Output),
+        ], all_undef_to_ff=True)
+        self.assertRepr(nl, """
+        (
+            (module 0 None ('top')
+                (output 'o' (cat 2.0:2 1.0 3.0:5))
+            )
+            (cell 0 0 (top
+                (output 'o' (cat 2.0:2 1.0 3.0:5))
+            ))
+            (cell 1 0 (instance 't' 'inst' (output 'o' 0:1)))
+            (cell 2 0 (flipflop 2.0:2 1 pos 0 0))
+            (cell 3 0 (flipflop 3.0:5 10 pos 0 0))
+        )
+        """)