diff --git a/amaranth/back/rtlil.py b/amaranth/back/rtlil.py index e824919..bd6a3f8 100644 --- a/amaranth/back/rtlil.py +++ b/amaranth/back/rtlil.py @@ -5,7 +5,7 @@ import warnings import re from .._utils import bits_for, flatten -from ..hdl import ast, ir, mem, xfrm +from ..hdl import ast, ir, mem, xfrm, _repr from ..lib import wiring @@ -337,12 +337,14 @@ class _ValueCompilerState: wire_name = signal.name is_sync_driven = signal in self.driven and self.driven[signal] - + attrs = dict(signal.attrs) - if signal._enum_class is not None: - attrs["enum_base_type"] = signal._enum_class.__name__ - for value in signal._enum_class: - attrs["enum_value_{:0{}b}".format(value.value, signal.width)] = value.name + for repr in signal._value_repr: + if repr.path == () and isinstance(repr.format, _repr.FormatEnum): + enum = repr.format.enum + attrs["enum_base_type"] = enum.__name__ + for value in enum: + attrs["enum_value_{:0{}b}".format(value.value, signal.width)] = value.name # For every signal in the sync domain, assign \sig's initial value (using the \init reg # attribute) to the reset value. @@ -872,7 +874,7 @@ def _convert_fragment(builder, fragment, name_map, hierarchy): if sub_type == "$mem_v2" and "MEMID" not in sub_params: sub_params["MEMID"] = builder._make_name(sub_name, local=False) - + sub_ports = OrderedDict() for port, value in sub_port_map.items(): if not isinstance(subfragment, ir.Instance): @@ -917,7 +919,7 @@ def _convert_fragment(builder, fragment, name_map, hierarchy): stmt_compiler._has_rhs = False stmt_compiler._wrap_assign = False stmt_compiler(group_stmts) - + # For every driven signal in the sync domain, create a flop of appropriate type. Which type # is appropriate depends on the domain: for domains with sync reset, it is a $dff, for # domains with async reset it is an $adff. The latter is directly provided with the reset @@ -930,7 +932,7 @@ def _convert_fragment(builder, fragment, name_map, hierarchy): wire_curr, wire_next = compiler_state.resolve(signal) if not cd.async_reset: - # For sync reset flops, the reset value comes from logic inserted by + # For sync reset flops, the reset value comes from logic inserted by # `hdl.xfrm.DomainLowerer`. module.cell("$dff", ports={ "\\CLK": wire_clk, diff --git a/amaranth/hdl/_repr.py b/amaranth/hdl/_repr.py new file mode 100644 index 0000000..adecc87 --- /dev/null +++ b/amaranth/hdl/_repr.py @@ -0,0 +1,46 @@ +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}" + + +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}" + + +class FormatCustom(Format): + def __init__(self, formatter): + self.formatter = formatter + + def format(self, value): + return self.formatter(value) + + +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 diff --git a/amaranth/hdl/ast.py b/amaranth/hdl/ast.py index 461345e..6d5107e 100644 --- a/amaranth/hdl/ast.py +++ b/amaranth/hdl/ast.py @@ -7,6 +7,7 @@ from collections.abc import Iterable, MutableMapping, MutableSet, MutableSequenc from enum import Enum from itertools import chain +from ._repr import * from .. import tracer from .._utils import * from .._utils import _ignore_deprecated @@ -53,6 +54,9 @@ class ShapeCastable: raise TypeError(f"Class '{cls.__name__}' deriving from `ShapeCastable` must override " f"the `const` method") + def _value_repr(self, value): + return (Repr(FormatInt(), value),) + class Shape: """Bit width and signedness of a value. @@ -1127,19 +1131,51 @@ class Signal(Value, DUID, metaclass=_SignalMeta): self.attrs = OrderedDict(() if attrs is None else attrs) - if decoder is None and isinstance(orig_shape, type) and issubclass(orig_shape, Enum): - decoder = orig_shape - if isinstance(decoder, type) and issubclass(decoder, Enum): + 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. The setter sets + # `self._value_repr` as well as the compatibility `self.decoder`. + self.decoder = decoder + 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): + self.decoder = orig_shape + else: + self.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(FormatEnum(orig_shape), self),) + else: + # Any other case is formatted as a plain integer. + self._value_repr = (Repr(FormatInt(), self),) + + @property + def decoder(self): + return self._decoder + + @decoder.setter + def decoder(self, decoder): + # Compute the value representation that will be used by Amaranth. + if decoder is None: + self._value_repr = (Repr(FormatInt(), self),) + self._decoder = None + elif not (isinstance(decoder, type) and issubclass(decoder, Enum)): + self._value_repr = (Repr(FormatCustom(decoder), self),) + self._decoder = decoder + else: # Violence. In the name of backwards compatibility! + self._value_repr = (Repr(FormatEnum(decoder), self),) def enum_decoder(value): try: return "{0.name:}/{0.value:}".format(decoder(value)) except ValueError: return str(value) - self.decoder = enum_decoder - self._enum_class = decoder - else: - self.decoder = decoder - self._enum_class = None + self._decoder = enum_decoder # Not a @classmethod because amaranth.compat requires it. @staticmethod @@ -1914,3 +1950,6 @@ class SignalDict(_MappedKeyDict): class SignalSet(_MappedKeySet): _map_key = SignalKey _unmap_key = lambda self, key: key.signal + + +from ._repr import * diff --git a/amaranth/lib/data.py b/amaranth/lib/data.py index 7c23836..151696e 100644 --- a/amaranth/lib/data.py +++ b/amaranth/lib/data.py @@ -1,8 +1,10 @@ from abc import ABCMeta, abstractmethod +from enum import Enum from collections.abc import Mapping, Sequence import warnings from amaranth.hdl import * +from amaranth.hdl._repr import * from amaranth.hdl.ast import ShapeCastable, ValueCastable @@ -231,6 +233,21 @@ class Layout(ShapeCastable, metaclass=ABCMeta): int_value |= (key_value.value << field.offset) & mask return View(self, Const(int_value, self.as_shape())) + 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. @@ -774,6 +791,9 @@ class _AggregateMeta(ShapeCastable, type): fields.update(init or {}) return cls.as_shape().const(fields) + def _value_repr(cls, value): + return cls.__layout._value_repr(value) + class Struct(View, metaclass=_AggregateMeta): """Structures defined with annotations. diff --git a/amaranth/lib/enum.py b/amaranth/lib/enum.py index 7800375..02e9dfe 100644 --- a/amaranth/lib/enum.py +++ b/amaranth/lib/enum.py @@ -2,6 +2,7 @@ import enum as py_enum import warnings from ..hdl.ast import Value, Shape, ShapeCastable, Const +from ..hdl._repr import * __all__ = py_enum.__all__ @@ -150,6 +151,9 @@ class EnumMeta(ShapeCastable, py_enum.EnumMeta): member = cls(init) return Const(member.value, cls.as_shape()) + def _value_repr(cls, value): + yield Repr(FormatEnum(cls), value) + class Enum(py_enum.Enum): """Subclass of the standard :class:`enum.Enum` that has :class:`EnumMeta` as diff --git a/amaranth/sim/pysim.py b/amaranth/sim/pysim.py index 7cfe340..f6d3e5c 100644 --- a/amaranth/sim/pysim.py +++ b/amaranth/sim/pysim.py @@ -5,7 +5,8 @@ from vcd import VCDWriter from vcd.gtkw import GTKWSave from ..hdl import * -from ..hdl.ast import SignalDict +from ..hdl._repr import * +from ..hdl.ast import SignalDict, Slice, Operator from ._base import * from ._pyrtl import _FragmentCompiler from ._pycoro import PyCoroProcess @@ -17,8 +18,24 @@ __all__ = ["PySimEngine"] class _VCDWriter: @staticmethod - def decode_to_vcd(signal, value): - return signal.decoder(value).expandtabs().replace(" ", "_") + 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 def __init__(self, fragment, *, vcd_file, gtkw_file=None, traces=()): if isinstance(vcd_file, str): @@ -55,53 +72,59 @@ class _VCDWriter: return for signal, names in itertools.chain(signal_names.items(), trace_names.items()): - if signal.decoder: - var_type = "string" - var_size = 1 - var_init = self.decode_to_vcd(signal, signal.reset) - else: - var_type = "wire" - var_size = signal.width - var_init = signal.reset + fields = [] + self.vcd_vars[signal] = [] + self.gtkw_names[signal] = [] + for repr in signal._value_repr: + var_init = self.eval_field(repr.value, signal, signal.reset) + 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) - for (*var_scope, var_name) in names: - if re.search(r"[ \t\r\n]", var_name): - raise NameError("Signal '{}.{}' contains a whitespace character" - .format(".".join(var_scope), var_name)) + vcd_var = None + for (*var_scope, var_name) in names: + if re.search(r"[ \t\r\n]", var_name): + raise NameError("Signal '{}.{}' contains a whitespace character" + .format(".".join(var_scope), var_name)) - suffix = None - while True: - try: - if suffix is None: - var_name_suffix = var_name + field_name = var_name + for item in repr.path: + if isinstance(item, int): + field_name += f"[{item}]" else: - var_name_suffix = f"{var_name}${suffix}" - if signal not in self.vcd_vars: - vcd_var = self.vcd_writer.register_var( - scope=var_scope, name=var_name_suffix, - var_type=var_type, size=var_size, init=var_init) - self.vcd_vars[signal] = vcd_var - else: - self.vcd_writer.register_alias( - scope=var_scope, name=var_name_suffix, - var=self.vcd_vars[signal]) - break - except KeyError: - suffix = (suffix or 0) + 1 + field_name += f".{item}" + + if vcd_var is None: + vcd_var = self.vcd_writer.register_var( + scope=var_scope, name=field_name, + var_type=var_type, size=var_size, init=var_init) + if var_size > 1: + suffix = f"[{var_size - 1}:0]" + else: + suffix = "" + if repr.path: + gtkw_field_name = '\\' + field_name + else: + gtkw_field_name = field_name + self.gtkw_names[signal].append(".".join((*var_scope, gtkw_field_name)) + suffix) + else: + self.vcd_writer.register_alias( + scope=var_scope, name=field_name, + var=vcd_var) + + self.vcd_vars[signal].append((vcd_var, repr)) - if signal not in self.gtkw_names: - self.gtkw_names[signal] = (*var_scope, var_name_suffix) def update(self, timestamp, signal, value): - vcd_var = self.vcd_vars.get(signal) - if vcd_var is None: - return - - if signal.decoder: - var_value = self.decode_to_vcd(signal, value) - else: - var_value = value - self.vcd_writer.change(vcd_var, timestamp, var_value) + for (vcd_var, repr) in self.vcd_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) + self.vcd_writer.change(vcd_var, timestamp, var_value) def close(self, timestamp): if self.vcd_writer is not None: @@ -113,11 +136,8 @@ class _VCDWriter: self.gtkw_save.treeopen("top") for signal in self.traces: - if len(signal) > 1 and not signal.decoder: - suffix = f"[{len(signal) - 1}:0]" - else: - suffix = "" - self.gtkw_save.trace(".".join(self.gtkw_names[signal]) + suffix) + for name in self.gtkw_names[signal]: + self.gtkw_save.trace(name) if self.vcd_file is not None: self.vcd_file.close()