sim: use Format.* for VCD output, remove hdl._repr.
This also changes `decoder` a bit: when an enum is used as a decoder, it is converted to a `Format.Enum` instead. The original enum is still stored on the `decoder` attribute, so that it can be propagated on `Signal.like`.
This commit is contained in:
Wanda
Catherine
parent
122be7849c
commit
8bf4f77616
|
|
@ -5,7 +5,7 @@ import io
|
|||
from ..utils import bits_for
|
||||
from .._utils import to_binary
|
||||
from ..lib import wiring
|
||||
from ..hdl import _repr, _ast, _ir, _nir
|
||||
from ..hdl import _ast, _ir, _nir
|
||||
|
||||
|
||||
__all__ = ["convert", "convert_fragment"]
|
||||
|
|
|
|||
|
|
@ -406,10 +406,6 @@ class ShapeCastable:
|
|||
"""
|
||||
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),)
|
||||
|
||||
|
||||
class _ShapeLikeMeta(type):
|
||||
def __subclasscheck__(cls, subclass):
|
||||
|
|
@ -2097,46 +2093,15 @@ class Signal(Value, DUID, metaclass=_SignalMeta):
|
|||
|
||||
if isinstance(orig_shape, ShapeCastable):
|
||||
self._format = orig_shape.format(orig_shape(self), "")
|
||||
elif isinstance(orig_shape, type) and issubclass(orig_shape, Enum):
|
||||
self._format = Format.Enum(self, orig_shape, name=orig_shape.__name__)
|
||||
else:
|
||||
self._format = Format("{}", self)
|
||||
|
||||
if decoder is not None:
|
||||
# The value representation is specified explicitly. Since we do not expose `hdl._repr`,
|
||||
# this is the only way to add a custom filter to the signal right now.
|
||||
if isinstance(decoder, type) and issubclass(decoder, Enum):
|
||||
self._value_repr = (_repr.Repr(_repr.FormatEnum(decoder), self),)
|
||||
else:
|
||||
self._value_repr = (_repr.Repr(_repr.FormatCustom(decoder), self),)
|
||||
else:
|
||||
# If it's an enum, expose it via `self.decoder` for compatibility, whether it's a Python
|
||||
# enum or an Amaranth enum. This also sets the value representation, even for custom
|
||||
# shape-castables that implement their own `_value_repr`.
|
||||
if isinstance(orig_shape, type) and issubclass(orig_shape, Enum):
|
||||
decoder = orig_shape
|
||||
else:
|
||||
decoder = None
|
||||
# The value representation is specified implicitly in the shape of the signal.
|
||||
if isinstance(orig_shape, ShapeCastable):
|
||||
# A custom shape-castable always has a `_value_repr`, at least the default one.
|
||||
self._value_repr = tuple(orig_shape._value_repr(self))
|
||||
elif isinstance(orig_shape, type) and issubclass(orig_shape, Enum):
|
||||
# A non-Amaranth enum needs a value repr constructed for it.
|
||||
self._value_repr = (_repr.Repr(_repr.FormatEnum(orig_shape), self),)
|
||||
else:
|
||||
# Any other case is formatted as a plain integer.
|
||||
self._value_repr = (_repr.Repr(_repr.FormatInt(), self),)
|
||||
|
||||
# Compute the value representation that will be used by Amaranth.
|
||||
if isinstance(decoder, type) and issubclass(decoder, Enum):
|
||||
# Violence. In the name of backwards compatibility!
|
||||
def enum_decoder(value):
|
||||
try:
|
||||
return "{0.name:}/{0.value:}".format(decoder(value))
|
||||
except ValueError:
|
||||
return str(value)
|
||||
self._decoder = enum_decoder
|
||||
else:
|
||||
self._decoder = decoder
|
||||
self._format = Format.Enum(self, decoder, name=decoder.__name__)
|
||||
|
||||
self._decoder = decoder
|
||||
|
||||
def shape(self):
|
||||
return Shape(self._width, self._signed)
|
||||
|
|
@ -3348,6 +3313,3 @@ class SignalDict(_MappedKeyDict):
|
|||
class SignalSet(_MappedKeySet):
|
||||
_map_key = SignalKey
|
||||
_unmap_key = lambda self, key: key.signal
|
||||
|
||||
|
||||
from . import _repr
|
||||
|
|
|
|||
|
|
@ -1,58 +0,0 @@
|
|||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
|
||||
__all__ = ["Format", "FormatInt", "FormatEnum", "FormatCustom", "Repr"]
|
||||
|
||||
|
||||
class Format(metaclass=ABCMeta):
|
||||
@abstractmethod
|
||||
def format(self, value):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class FormatInt(Format):
|
||||
def format(self, value):
|
||||
return f"{value:d}"
|
||||
|
||||
def __repr__(self):
|
||||
return f"FormatInt()"
|
||||
|
||||
|
||||
class FormatEnum(Format):
|
||||
def __init__(self, enum):
|
||||
self.enum = enum
|
||||
|
||||
def format(self, value):
|
||||
try:
|
||||
return f"{self.enum(value).name}/{value:d}"
|
||||
except ValueError:
|
||||
return f"?/{value:d}"
|
||||
|
||||
def __repr__(self):
|
||||
return f"FormatEnum({self.enum.__name__})"
|
||||
|
||||
|
||||
class FormatCustom(Format):
|
||||
def __init__(self, formatter):
|
||||
self.formatter = formatter
|
||||
|
||||
def format(self, value):
|
||||
return self.formatter(value)
|
||||
|
||||
def __repr__(self):
|
||||
return f"FormatCustom({self.formatter})"
|
||||
|
||||
|
||||
class Repr:
|
||||
def __init__(self, format, value, *, path=()):
|
||||
from ._ast import Value # avoid a circular dependency
|
||||
assert isinstance(format, Format)
|
||||
assert isinstance(value, Value)
|
||||
assert isinstance(path, tuple) and all(isinstance(part, (str, int)) for part in path)
|
||||
|
||||
self.format = format
|
||||
self.value = value
|
||||
self.path = path
|
||||
|
||||
def __repr__(self):
|
||||
return f"Repr({self.format!r}, {self.value!r}, {self.path!r})"
|
||||
|
|
@ -6,7 +6,6 @@ import operator
|
|||
|
||||
from amaranth._utils import final
|
||||
from amaranth.hdl import *
|
||||
from amaranth.hdl._repr import Repr, FormatInt, FormatEnum
|
||||
from amaranth import hdl
|
||||
|
||||
|
||||
|
|
@ -264,21 +263,6 @@ class Layout(ShapeCastable, metaclass=ABCMeta):
|
|||
fields[str(key)] = Format("{}", field_value)
|
||||
return Format.Struct(value, fields)
|
||||
|
||||
def _value_repr(self, value):
|
||||
yield Repr(FormatInt(), value)
|
||||
for key, field in self:
|
||||
shape = Shape.cast(field.shape)
|
||||
field_value = value[field.offset:field.offset+shape.width]
|
||||
if shape.signed:
|
||||
field_value = field_value.as_signed()
|
||||
if isinstance(field.shape, ShapeCastable):
|
||||
for repr in field.shape._value_repr(field_value):
|
||||
yield Repr(repr.format, repr.value, path=(key,) + repr.path)
|
||||
elif isinstance(field.shape, type) and issubclass(field.shape, Enum):
|
||||
yield Repr(FormatEnum(field.shape), field_value, path=(key,))
|
||||
else:
|
||||
yield Repr(FormatInt(), field_value, path=(key,))
|
||||
|
||||
|
||||
class StructLayout(Layout):
|
||||
"""Description of a structure layout.
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ import warnings
|
|||
import operator
|
||||
|
||||
from ..hdl import Value, ValueCastable, Shape, ShapeCastable, Const, SyntaxWarning, Format
|
||||
from ..hdl._repr import Repr, FormatEnum
|
||||
|
||||
|
||||
__all__ = py_enum.__all__ + ["EnumView", "FlagView"]
|
||||
|
|
@ -181,9 +180,6 @@ class EnumType(ShapeCastable, py_enum.EnumMeta):
|
|||
raise ValueError(f"Format specifier {format_spec!r} is not supported for enums")
|
||||
return Format.Enum(value, cls, name=cls.__name__)
|
||||
|
||||
def _value_repr(cls, value):
|
||||
yield Repr(FormatEnum(cls), value)
|
||||
|
||||
|
||||
# In 3.11, Python renamed EnumMeta to EnumType. Like Python itself, we support both for
|
||||
# compatibility.
|
||||
|
|
|
|||
|
|
@ -2,12 +2,12 @@ from contextlib import contextmanager
|
|||
import itertools
|
||||
import re
|
||||
import os.path
|
||||
import enum as py_enum
|
||||
|
||||
from ..hdl import *
|
||||
from ..hdl._repr import *
|
||||
from ..hdl._mem import MemoryInstance
|
||||
from ..hdl._ast import SignalDict, Slice, Operator
|
||||
from ..hdl._ast import SignalDict
|
||||
from ._base import *
|
||||
from ._pyeval import eval_format, eval_value
|
||||
from ._pyrtl import _FragmentCompiler
|
||||
from ._pycoro import PyCoroProcess
|
||||
from ._pyclock import PyClockProcess
|
||||
|
|
@ -21,23 +21,8 @@ class _VCDWriter:
|
|||
def decode_to_vcd(format, value):
|
||||
return format.format(value).expandtabs().replace(" ", "_")
|
||||
|
||||
@staticmethod
|
||||
def eval_field(field, signal, value):
|
||||
if isinstance(field, Signal):
|
||||
assert field is signal
|
||||
return value
|
||||
elif isinstance(field, Const):
|
||||
return field.value
|
||||
elif isinstance(field, Slice):
|
||||
sub = _VCDWriter.eval_field(field.value, signal, value)
|
||||
return (sub >> field.start) & ((1 << (field.stop - field.start)) - 1)
|
||||
elif isinstance(field, Operator) and field.operator in ('s', 'u'):
|
||||
sub = _VCDWriter.eval_field(field.operands[0], signal, value)
|
||||
return Const(sub, field.shape()).value
|
||||
else:
|
||||
raise NotImplementedError # :nocov:
|
||||
|
||||
def __init__(self, design, *, vcd_file, gtkw_file=None, traces=(), fs_per_delta=0, processes=()):
|
||||
def __init__(self, state, design, *, vcd_file, gtkw_file=None, traces=(), fs_per_delta=0, processes=()):
|
||||
self.state = state
|
||||
self.fs_per_delta = fs_per_delta
|
||||
|
||||
# Although pyvcd is a mandatory dependency, be resilient and import it as needed, so that
|
||||
|
|
@ -123,16 +108,8 @@ class _VCDWriter:
|
|||
for signal, names in itertools.chain(signal_names.items(), trace_names.items()):
|
||||
self.vcd_signal_vars[signal] = []
|
||||
self.gtkw_signal_names[signal] = []
|
||||
for repr in signal._value_repr:
|
||||
var_init = self.eval_field(repr.value, signal, signal.init)
|
||||
if isinstance(repr.format, FormatInt):
|
||||
var_type = "wire"
|
||||
var_size = repr.value.shape().width
|
||||
else:
|
||||
var_type = "string"
|
||||
var_size = 1
|
||||
var_init = self.decode_to_vcd(repr.format, var_init)
|
||||
|
||||
def add_var(path, var_type, var_size, var_init, value):
|
||||
vcd_var = None
|
||||
for (*var_scope, var_name) in names:
|
||||
if re.search(r"[ \t\r\n]", var_name):
|
||||
|
|
@ -140,12 +117,12 @@ class _VCDWriter:
|
|||
.format(".".join(var_scope), var_name))
|
||||
|
||||
field_name = var_name
|
||||
for item in repr.path:
|
||||
for item in path:
|
||||
if isinstance(item, int):
|
||||
field_name += f"[{item}]"
|
||||
else:
|
||||
field_name += f".{item}"
|
||||
if repr.path:
|
||||
if path:
|
||||
field_name = "\\" + field_name
|
||||
|
||||
if vcd_var is None:
|
||||
|
|
@ -162,7 +139,35 @@ class _VCDWriter:
|
|||
scope=var_scope, name=field_name,
|
||||
var=vcd_var)
|
||||
|
||||
self.vcd_signal_vars[signal].append((vcd_var, repr))
|
||||
self.vcd_signal_vars[signal].append((vcd_var, value))
|
||||
|
||||
def add_wire_var(path, value):
|
||||
add_var(path, "wire", len(value), eval_value(self.state, value), value)
|
||||
|
||||
def add_format_var(path, fmt):
|
||||
add_var(path, "string", 1, eval_format(self.state, fmt), fmt)
|
||||
|
||||
def add_format(path, fmt):
|
||||
if isinstance(fmt, Format.Struct):
|
||||
add_wire_var(path, fmt._value)
|
||||
for name, subfmt in fmt._fields.items():
|
||||
add_format(path + (name,), subfmt)
|
||||
elif isinstance(fmt, Format.Array):
|
||||
add_wire_var(path, fmt._value)
|
||||
for idx, subfmt in enumerate(fmt._fields):
|
||||
add_format(path + (idx,), subfmt)
|
||||
elif (isinstance(fmt, Format) and
|
||||
len(fmt._chunks) == 1 and
|
||||
isinstance(fmt._chunks[0], tuple) and
|
||||
fmt._chunks[0][1] == ""):
|
||||
add_wire_var(path, fmt._chunks[0][0])
|
||||
else:
|
||||
add_format_var(path, fmt)
|
||||
|
||||
if signal._decoder is not None and not isinstance(signal._decoder, py_enum.EnumMeta):
|
||||
add_var((), "string", 1, signal._decoder(signal._init), signal._decoder)
|
||||
else:
|
||||
add_format((), signal._format)
|
||||
|
||||
for memory, memory_name in memories.items():
|
||||
self.vcd_memory_vars[memory] = vcd_vars = []
|
||||
|
|
@ -205,11 +210,15 @@ class _VCDWriter:
|
|||
except KeyError:
|
||||
pass # try another name
|
||||
|
||||
def update_signal(self, timestamp, signal, value):
|
||||
def update_signal(self, timestamp, signal):
|
||||
for (vcd_var, repr) in self.vcd_signal_vars.get(signal, ()):
|
||||
var_value = self.eval_field(repr.value, signal, value)
|
||||
if not isinstance(repr.format, FormatInt):
|
||||
var_value = self.decode_to_vcd(repr.format, var_value)
|
||||
if isinstance(repr, Value):
|
||||
var_value = eval_value(self.state, repr)
|
||||
elif isinstance(repr, (Format, Format.Enum)):
|
||||
var_value = eval_format(self.state, repr)
|
||||
else:
|
||||
# decoder
|
||||
var_value = repr(eval_value(self.state, signal))
|
||||
self.vcd_writer.change(vcd_var, timestamp, var_value)
|
||||
|
||||
def update_memory(self, timestamp, memory, addr, value):
|
||||
|
|
@ -512,7 +521,7 @@ class PySimEngine(BaseEngine):
|
|||
if isinstance(change, _PySignalState):
|
||||
signal_state = change
|
||||
vcd_writer.update_signal(now_plus_deltas,
|
||||
signal_state.signal, signal_state.curr)
|
||||
signal_state.signal)
|
||||
elif isinstance(change, _PyMemoryChange):
|
||||
vcd_writer.update_memory(now_plus_deltas, change.state.memory,
|
||||
change.addr, change.state.data[change.addr])
|
||||
|
|
@ -559,7 +568,7 @@ class PySimEngine(BaseEngine):
|
|||
|
||||
@contextmanager
|
||||
def write_vcd(self, *, vcd_file, gtkw_file, traces, fs_per_delta):
|
||||
vcd_writer = _VCDWriter(self._design,
|
||||
vcd_writer = _VCDWriter(self._state, self._design,
|
||||
vcd_file=vcd_file, gtkw_file=gtkw_file, traces=traces, fs_per_delta=fs_per_delta,
|
||||
processes=self._testbenches)
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -61,6 +61,7 @@ Implemented RFCs
|
|||
.. _RFC 59: https://amaranth-lang.org/rfcs/0059-no-domain-upwards-propagation.html
|
||||
.. _RFC 62: https://amaranth-lang.org/rfcs/0062-memory-data.html
|
||||
.. _RFC 63: https://amaranth-lang.org/rfcs/0063-remove-lib-coding.html
|
||||
.. _RFC 65: https://amaranth-lang.org/rfcs/0065-format-struct-enum.html
|
||||
|
||||
* `RFC 17`_: Remove ``log2_int``
|
||||
* `RFC 27`_: Testbench processes for the simulator
|
||||
|
|
@ -75,6 +76,7 @@ Implemented RFCs
|
|||
* `RFC 59`_: Get rid of upwards propagation of clock domains
|
||||
* `RFC 62`_: The ``MemoryData`` class
|
||||
* `RFC 63`_: Remove ``amaranth.lib.coding``
|
||||
* `RFC 65`_: Special formatting for structures and enums
|
||||
|
||||
|
||||
Language changes
|
||||
|
|
|
|||
|
|
@ -1326,17 +1326,15 @@ class SignalTestCase(FHDLTestCase):
|
|||
RED = 1
|
||||
BLUE = 2
|
||||
s = Signal(decoder=Color)
|
||||
self.assertEqual(s.decoder(1), "RED/1")
|
||||
self.assertEqual(s.decoder(3), "3")
|
||||
self.assertRepr(s._value_repr, "(Repr(FormatEnum(Color), (sig s), ()),)")
|
||||
self.assertEqual(s.decoder, Color)
|
||||
self.assertRepr(s._format, "(format-enum (sig s) 'Color' (1 'RED') (2 'BLUE'))")
|
||||
|
||||
def test_enum(self):
|
||||
s1 = Signal(UnsignedEnum)
|
||||
self.assertEqual(s1.shape(), unsigned(2))
|
||||
s2 = Signal(SignedEnum)
|
||||
self.assertEqual(s2.shape(), signed(2))
|
||||
self.assertEqual(s2.decoder(SignedEnum.FOO), "FOO/-1")
|
||||
self.assertRepr(s2._value_repr, "(Repr(FormatEnum(SignedEnum), (sig s2), ()),)")
|
||||
self.assertRepr(s2._format, "(format-enum (sig s2) 'SignedEnum' (-1 'FOO') (0 'BAR') (1 'BAZ'))")
|
||||
|
||||
def test_const_wrong(self):
|
||||
s1 = Signal()
|
||||
|
|
@ -1344,9 +1342,9 @@ class SignalTestCase(FHDLTestCase):
|
|||
r"^Value \(sig s1\) cannot be converted to an Amaranth constant$"):
|
||||
Const.cast(s1)
|
||||
|
||||
def test_value_repr(self):
|
||||
def test_format_simple(self):
|
||||
s = Signal()
|
||||
self.assertRepr(s._value_repr, "(Repr(FormatInt(), (sig s), ()),)")
|
||||
self.assertRepr(s._format, "(format '{}' (sig s))")
|
||||
|
||||
|
||||
class ClockSignalTestCase(FHDLTestCase):
|
||||
|
|
|
|||
|
|
@ -203,7 +203,7 @@ class RecordTestCase(FHDLTestCase):
|
|||
|
||||
def test_enum_decoder(self):
|
||||
r1 = Record([("a", UnsignedEnum)])
|
||||
self.assertEqual(r1.a.decoder(UnsignedEnum.FOO), "FOO/1")
|
||||
self.assertRepr(r1.a._format, "(format-enum (sig r1__a) 'UnsignedEnum' (1 'FOO') (2 'BAR') (3 'BAZ'))")
|
||||
|
||||
def test_operators(self):
|
||||
r1 = Record([("a", 1)])
|
||||
|
|
|
|||
|
|
@ -531,10 +531,11 @@ class ViewTestCase(FHDLTestCase):
|
|||
self.assertIsInstance(cv, Signal)
|
||||
self.assertEqual(cv.shape(), unsigned(3))
|
||||
self.assertEqual(cv.name, "v")
|
||||
self.assertRepr(cv._value_repr, """
|
||||
(Repr(FormatInt(), (sig v), ()),
|
||||
Repr(FormatInt(), (slice (sig v) 0:1), ('a',)),
|
||||
Repr(FormatInt(), (slice (sig v) 1:3), ('b',)))
|
||||
self.assertRepr(cv._format, """
|
||||
(format-struct (sig v)
|
||||
('a' (format '{}' (slice (sig v) 0:1)))
|
||||
('b' (format '{}' (slice (sig v) 1:3)))
|
||||
)
|
||||
""")
|
||||
|
||||
def test_construct_signal_init(self):
|
||||
|
|
|
|||
|
|
@ -138,7 +138,7 @@ class EnumTestCase(FHDLTestCase):
|
|||
B = -3
|
||||
a = Signal(EnumA)
|
||||
self.assertRepr(a, "(sig a)")
|
||||
self.assertRepr(a._value_repr, "(Repr(FormatEnum(EnumA), (sig a), ()),)")
|
||||
self.assertRepr(a._format, "(format-enum (sig a) 'EnumA' (0 'A') (-3 'B'))")
|
||||
|
||||
def test_enum_view(self):
|
||||
class EnumA(Enum, shape=signed(4)):
|
||||
|
|
|
|||
|
|
@ -469,10 +469,10 @@ class SimulatorUnitTestCase(FHDLTestCase):
|
|||
|
||||
class SimulatorIntegrationTestCase(FHDLTestCase):
|
||||
@contextmanager
|
||||
def assertSimulation(self, module, *, deadline=None):
|
||||
def assertSimulation(self, module, *, deadline=None, traces=[]):
|
||||
sim = Simulator(module)
|
||||
yield sim
|
||||
with sim.write_vcd("test.vcd", "test.gtkw"):
|
||||
with sim.write_vcd("test.vcd", "test.gtkw", traces=traces):
|
||||
if deadline is None:
|
||||
sim.run()
|
||||
else:
|
||||
|
|
@ -1336,10 +1336,25 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
|
|||
self.assertEqual(eval_format(state, Format("{:s}", sig5)), "ABCD")
|
||||
self.assertEqual(eval_format(state, Format("{:<5s}", sig5)), "ABCD ")
|
||||
|
||||
sim = Simulator(Module())
|
||||
sim.add_testbench(testbench)
|
||||
with sim.write_vcd("test.vcd", fs_per_delta=1):
|
||||
sim.run()
|
||||
with self.assertSimulation(Module(), traces=[sig, sig2, sig3, sig4, sig5]) as sim:
|
||||
sim.add_testbench(testbench)
|
||||
|
||||
def test_decoder(self):
|
||||
def decoder(val):
|
||||
return f".oO{val}Oo."
|
||||
|
||||
sig = Signal(2, decoder=decoder)
|
||||
|
||||
def testbench():
|
||||
yield Delay(1e-6)
|
||||
yield sig.eq(1)
|
||||
yield Delay(1e-6)
|
||||
yield sig.eq(2)
|
||||
yield Delay(1e-6)
|
||||
|
||||
with self.assertSimulation(Module(), traces=[sig]) as sim:
|
||||
sim.add_testbench(testbench)
|
||||
|
||||
|
||||
class SimulatorRegressionTestCase(FHDLTestCase):
|
||||
def test_bug_325(self):
|
||||
|
|
|
|||
Loading…
Reference in a new issue