[WIP] hdl.ast: add Display statement, a mixture of print() and format().

amaranth-lang · Dec 12, 2020 · fef97ae · fef97ae
1 parent 7dde2aa
commit fef97ae
Show file tree

Hide file tree

Showing 6 changed files with 134 additions and 32 deletions.
diff --git a/nmigen/__init__.py b/nmigen/__init__.py
@@ -17,6 +17,7 @@
 __all__ = [
     "Shape", "unsigned", "signed",
     "Value", "Const", "C", "Mux", "Cat", "Repl", "Array", "Signal", "ClockSignal", "ResetSignal",
+    "Display",
     "Module",
     "ClockDomain",
     "Elaboratable", "Fragment", "Instance",

diff --git a/nmigen/hdl/__init__.py b/nmigen/hdl/__init__.py
@@ -1,5 +1,6 @@
 from .ast import Shape, unsigned, signed
 from .ast import Value, Const, C, Mux, Cat, Repl, Array, Signal, ClockSignal, ResetSignal
+from .ast import Display
 from .dsl import Module
 from .cd import ClockDomain
 from .ir import Elaboratable, Fragment, Instance
@@ -11,6 +12,7 @@
 __all__ = [
     "Shape", "unsigned", "signed",
     "Value", "Const", "C", "Mux", "Cat", "Repl", "Array", "Signal", "ClockSignal", "ResetSignal",
+    "Display",
     "Module",
     "ClockDomain",
     "Elaboratable", "Fragment", "Instance",

diff --git a/nmigen/hdl/ast.py b/nmigen/hdl/ast.py
@@ -1,9 +1,11 @@
 from abc import ABCMeta, abstractmethod
-import traceback
+import typing
 import sys
 import warnings
-import typing
+import traceback
 import functools
+import re
+import string
 from collections import OrderedDict
 from collections.abc import Iterable, MutableMapping, MutableSet, MutableSequence
 from enum import Enum
@@ -20,8 +22,8 @@
     "Signal", "ClockSignal", "ResetSignal",
     "UserValue", "ValueCastable",
     "Sample", "Past", "Stable", "Rose", "Fell", "Initial",
-    "Statement", "Switch",
-    "Property", "Assign", "Assert", "Assume", "Cover",
+    "Statement", "Switch", "Assign",
+    "Display", "Property", "Assert", "Assume", "Cover",
     "ValueKey", "ValueDict", "ValueSet", "SignalKey", "SignalDict", "SignalSet",
 ]
 
@@ -1432,6 +1434,79 @@ def __repr__(self):
         return "(eq {!r} {!r})".format(self.lhs, self.rhs)
 
 
+class _DisplayFormatter(string.Formatter):
+    _ESCAPE_TRANS = str.maketrans({"{": "{{", "}": "}}"})
+
+    @classmethod
+    def escape(cls, string):
+        return string.translate(cls._ESCAPE_TRANS)
+
+    _FORMAT_RE = re.compile(r"""
+        ^
+        (?: (?P<fill> [ 0])? (?P<align> [<>=]) )?
+        (?P<sign> [ +-])?
+        (?P<prefix> \#)?
+        (?P<zero> 0)?
+        (?P<width> \d+)?
+        (?P<type> [bodx])?
+        $
+    """, re.X)
+
+    @classmethod
+    def _process_spec(cls, format_spec):
+        m = re.match(cls._FORMAT_RE, format_spec)
+        if m is None:
+            raise SyntaxError("Invalid Display format specifier {!r}".format(format_spec))
+        return format_spec
+
+    def __init__(self):
+        self.args = []
+
+    def format_field(self, value, format_spec):
+        if isinstance(value, (Value, ValueCastable)):
+            index = len(self.args)
+            self.args.append(Value.cast(value))
+            return "{{{}:{}}}".format(index, self._process_spec(format_spec))
+        else:
+            return self.escape(format(value, format_spec))
+
+    def convert_field(self, value, conversion):
+        if conversion is None:
+            return value
+        raise SyntaxError("Conversion specifiers are not supported in Display")
+
+    def parse(self, format_string):
+        for literal_text, field_name, format_spec, conversion in super().parse(format_string):
+            yield self.escape(literal_text), field_name, format_spec, conversion
+
+
+@final
+class Display(Statement):
+    def __init__(self, format_string, *args, end="\n", src_loc_at=0, _en=None, **kwargs):
+        super().__init__(src_loc_at=src_loc_at)
+
+        formatter = _DisplayFormatter()
+        self.format = formatter.vformat(format_string, args, kwargs) + formatter.escape(end)
+        self.args   = formatter.args
+        self._en    = _en
+        if self._en is None:
+            self._en = Signal(reset_less=True, name="$display$en")
+            self._en.src_loc = self.src_loc
+
+    def __repr__(self):
+        if self.args:
+            return "(display {!r} {})".format(self.format, " ".join(map(repr, self.args)))
+        else:
+            return "(display {!r})".format(self.format)
+
+    def _lhs_signals(self):
+        return SignalSet((self._en,))
+
+    def _rhs_signals(self):
+        return union((arg._rhs_signals() for arg in self.args),
+                     start=SignalSet())
+
+
 class UnusedProperty(UnusedMustUse):
     pass
 
@@ -1447,7 +1522,7 @@ def __init__(self, test, *, _check=None, _en=None, src_loc_at=0):
         if self._check is None:
             self._check = Signal(reset_less=True, name="${}$check".format(self._kind))
             self._check.src_loc = self.src_loc
-        if _en is None:
+        if self._en is None:
             self._en = Signal(reset_less=True, name="${}$en".format(self._kind))
             self._en.src_loc = self.src_loc
 

diff --git a/nmigen/hdl/dsl.py b/nmigen/hdl/dsl.py
@@ -484,9 +484,10 @@ def domain_name(domain):
             self._pop_ctrl()
 
         for stmt in Statement.cast(assigns):
-            if not compat_mode and not isinstance(stmt, (Assign, Assert, Assume, Cover)):
+            if not compat_mode and not isinstance(stmt, (Assign, Display, Assert, Assume, Cover)):
                 raise SyntaxError(
-                    "Only assignments and property checks may be appended to d.{}"
+                    "Only assignment, display, and property check statements may be appended "
+                    "to d.{}"
                     .format(domain_name(domain)))
 
             stmt._MustUse__used = True

diff --git a/nmigen/hdl/xfrm.py b/nmigen/hdl/xfrm.py
@@ -174,6 +174,10 @@ def on_Initial(self, value):
 
 
 class StatementVisitor(metaclass=ABCMeta):
+    @abstractmethod
+    def on_Display(self, stmt):
+        pass # :nocov:
+
     @abstractmethod
     def on_Assign(self, stmt):
         pass # :nocov:
@@ -207,6 +211,8 @@ def replace_statement_src_loc(self, stmt, new_stmt):
     def on_statement(self, stmt):
         if type(stmt) is Assign:
             new_stmt = self.on_Assign(stmt)
+        elif type(stmt) is Display:
+            new_stmt = self.on_Display(stmt)
         elif type(stmt) is Assert:
             new_stmt = self.on_Assert(stmt)
         elif type(stmt) is Assume:
@@ -239,6 +245,9 @@ def on_value(self, value):
     def on_Assign(self, stmt):
         return Assign(self.on_value(stmt.lhs), self.on_value(stmt.rhs))
 
+    def on_Display(self, stmt):
+        return Display(stmt.format, *map(self.on_value, stmt.args), end="", _en=stmt._en)
+
     def on_Assert(self, stmt):
         return Assert(self.on_value(stmt.test), _check=stmt._check, _en=stmt._en)
 
@@ -395,6 +404,10 @@ def on_Assign(self, stmt):
         self.on_value(stmt.lhs)
         self.on_value(stmt.rhs)
 
+    def on_Display(self, stmt):
+        for arg in stmt.args:
+            self.on_value(arg)
+
     def on_property(self, stmt):
         self.on_value(stmt.test)
 
@@ -653,14 +666,15 @@ def on_Assign(self, stmt):
         if lhs_signals:
             self.unify(*stmt._lhs_signals())
 
-    def on_property(self, stmt):
+    def on_cell(self, stmt):
         lhs_signals = stmt._lhs_signals()
         if lhs_signals:
             self.unify(*stmt._lhs_signals())
 
-    on_Assert = on_property
-    on_Assume = on_property
-    on_Cover  = on_property
+    on_Display = on_cell
+    on_Assert  = on_cell
+    on_Assume  = on_cell
+    on_Cover   = on_cell
 
     def on_Switch(self, stmt):
         for case_stmts in stmt.cases.values():
@@ -688,14 +702,15 @@ def on_Assign(self, stmt):
             if any_lhs_signal in self.signals:
                 return stmt
 
-    def on_property(self, stmt):
+    def on_cell(self, stmt):
         any_lhs_signal = next(iter(stmt._lhs_signals()))
         if any_lhs_signal in self.signals:
             return stmt
 
-    on_Assert = on_property
-    on_Assume = on_property
-    on_Cover  = on_property
+    on_Display = on_cell
+    on_Assert  = on_cell
+    on_Assume  = on_cell
+    on_Cover   = on_cell
 
 
 class _ControlInserter(FragmentTransformer):

diff --git a/nmigen/sim/_pyrtl.py b/nmigen/sim/_pyrtl.py
@@ -330,17 +330,28 @@ def __init__(self, state, emitter, *, inputs=None, outputs=None):
         self.rhs = _RHSValueCompiler(state, emitter, mode="curr", inputs=inputs)
         self.lhs = _LHSValueCompiler(state, emitter, rhs=self.rhs, outputs=outputs)
 
-    def on_statements(self, stmts):
-        for stmt in stmts:
-            self(stmt)
-        if not stmts:
-            self.emitter.append("pass")
+    def _prepare_rhs(self, value):
+        value_mask = (1 << len(value)) - 1
+        if value.shape().signed:
+            return f"sign({value_mask} & {self.rhs(value)}, {-1 << (len(value) - 1)})"
+        else: # unsigned
+            return f"({value_mask} & {self.rhs(value)})"
 
     def on_Assign(self, stmt):
-        gen_rhs = f"({(1 << len(stmt.rhs)) - 1} & {self.rhs(stmt.rhs)})"
-        if stmt.rhs.shape().signed:
-            gen_rhs = f"sign({gen_rhs}, {-1 << (len(stmt.rhs) - 1)})"
-        return self.lhs(stmt.lhs)(gen_rhs)
+        self.lhs(stmt.lhs)(self._prepare_rhs(stmt.rhs))
+
+    def on_Display(self, stmt):
+        gen_args = [self._prepare_rhs(arg) for arg in stmt.args]
+        self.emitter.append(f"print({stmt.format!r}.format({', '.join(gen_args)}), end='')")
+
+    def on_Assert(self, stmt):
+        raise NotImplementedError # :nocov:
+
+    def on_Assume(self, stmt):
+        raise NotImplementedError # :nocov:
+
+    def on_Cover(self, stmt):
+        raise NotImplementedError # :nocov:
 
     def on_Switch(self, stmt):
         gen_test = self.emitter.def_var("test",
@@ -365,14 +376,11 @@ def on_Switch(self, stmt):
             with self.emitter.indent():
                 self(stmts)
 
-    def on_Assert(self, stmt):
-        raise NotImplementedError # :nocov:
-
-    def on_Assume(self, stmt):
-        raise NotImplementedError # :nocov:
-
-    def on_Cover(self, stmt):
-        raise NotImplementedError # :nocov:
+    def on_statements(self, stmts):
+        for stmt in stmts:
+            self(stmt)
+        if not stmts:
+            self.emitter.append("pass")
 
     @classmethod
     def compile(cls, state, stmt):