[topgen] Add class to load a complete topcfg properly

At the moment, topgen simplies serializes the complete topcfg using
hjson.dumps() but this poses a few problems. The biggest one is that
it is a mix of manual conversion to dict (when calling as_dict()
in the various topgen functions) and automatic (when calling _asdict()
from hjson.dumps). Furthermore, it turns out that we are missing some
fields that probably were never added to _asdict().

This commit introduces a new class (CompleteTopCfg) whose sole purpose
is to take the produced Hjson and reconstruct an in-memory topcfg that
is *exactly* equivalent (in Pythonic types) to the ones that was dumped.
This requires to sometimes reconstruct some classes, sometimes not.
Classes that need to be reconstruct get a new method (fromdict) and
the CompleteTopCfg does as much automatic deserializing as possible,
then some manual fixing.

Since this process is quite fragile, when topgen is running it will
actually check that this works by dumping the Hjson, reloading it
with the CompleteTopCfg and then checking that the two are equivalent.

Signed-off-by: Amaury Pouly <[email protected]>

util/reggen/inter_signal.py

@@ -90,8 +90,21 @@ def _asdict(self) -> Dict[str, object]:
         ret['width'] = self.width
         if self.default is not None:
             ret['default'] = self.default
+        ret['class'] = 'InterSignal'  # This will let fromdict() know it has to create the class
 
         return ret
 
+    @classmethod
+    def fromdict(cls, item: Dict[str, object]) -> object:
+        if 'class' not in item or item['class'] == 'InterSignal':
+            return item
+        item["package"] = item.get("package", None)
+        item["default"] = item.get("default", None)
+        item["signal_type"] = item["type"]
+        del item["type"]
+        c = cls.__new__(cls)
+        c.__dict__.update(**item)
+        return c
+
     def as_dict(self) -> Dict[str, object]:
         return self._asdict()

util/reggen/params.py

@@ -55,8 +55,29 @@ def as_dict(self) -> Dict[str, object]:
         rd['type'] = self.param_type
         if self.unpacked_dimensions is not None:
             rd['unpacked_dimensions'] = self.unpacked_dimensions
+        # topgen sometimes manually adds a 'name_top' field after creation.
+        if getattr(self, "name_top", None):
+            rd['name_top'] = getattr(self, "name_top")
         return rd
 
+    def _asdict(self) -> Dict[str, object]:
+        # Add an attribute to distinguished between manual serialization (as_dict())
+        # or automatic by hjson.
+        d = self.as_dict()
+        d['class'] = self.__class__.__name__
+        return d
+
+    @classmethod
+    def fromdict(cls, param: Dict[str, object]) -> object:
+        param['desc'] = param.get('desc', None)
+        param['unpacked_dimensions'] = param.get('unpacked_dimensions', None)
+        del param['class']
+        param['param_type'] = param['type']
+        del param['type']
+        c = cls.__new__(cls)
+        c.__dict__.update(**param)
+        return c
+
 
 class LocalParam(BaseParam):
     def __init__(self,
@@ -82,6 +103,12 @@ def as_dict(self) -> Dict[str, object]:
         rd['default'] = self.value
         return rd
 
+    @classmethod
+    def fromdict(cls, param: Dict[str, object]) -> object:
+        assert param['local']
+        del param['local']
+        return super().fromdict.__func__(cls, param)  # type: ignore
+
 
 class Parameter(BaseParam):
     def __init__(self,
@@ -106,6 +133,12 @@ def as_dict(self) -> Dict[str, object]:
         rd['name_top'] = self.name_top
         return rd
 
+    @classmethod
+    def fromdict(cls, param: Dict[str, object]) -> object:
+        param['local'] = param['local'] == 'true'
+        param['expose'] = param['expose'] == 'true'
+        return super().fromdict.__func__(cls, param)  # type: ignore
+
 
 class RandParameter(BaseParam):
     def __init__(self,
@@ -131,6 +164,10 @@ def as_dict(self) -> Dict[str, object]:
         rd['randtype'] = self.randtype
         return rd
 
+    @classmethod
+    def fromdict(cls, param: Dict[str, object]) -> object:
+        return super().fromdict.__func__(cls, param)  # type: ignore
+
 
 class MemSizeParameter(BaseParam):
     def __init__(self,

util/topgen.py

@@ -40,6 +40,7 @@
 from topgen.resets import Resets
 from topgen.rust import TopGenRust
 from topgen.top import Top
+from topgen.topcfg import CompleteTopCfg
 
 # Common header for generated files
 warnhdr = """//
@@ -91,6 +92,9 @@ def ipgen_render(template_name: str, topname: str, params: Dict[str, object],
         log.error(e.verbose_str())
         sys.exit(1)
 
+    # Remote extra topname
+    params.pop("topname")
+
 
 def generate_top(top: Dict[str, object], name_to_block: Dict[str, IpBlock],
                  tpl_filename: str, **kwargs: Dict[str, object]) -> None:
@@ -907,6 +911,12 @@ def _process_top(
     return completecfg, name_to_block, name_to_hjson
 
 
+def test_topcfg_loader(genhjson_path: Path, completecfg: Dict[str, object]):
+    loaded_cfg = CompleteTopCfg.from_path(genhjson_path)
+
+    CompleteTopCfg.check_equivalent(completecfg, loaded_cfg)
+
+
 def _check_countermeasures(completecfg: Dict[str, object],
                            name_to_block: Dict[str, IpBlock],
                            name_to_hjson: Dict[str, Path]) -> bool:
@@ -1211,6 +1221,10 @@ def main():
     genhjson_path.write_text(genhdr + gencmd +
                              hjson.dumps(completecfg, for_json=True, default=vars) + '\n')
 
+    # We also run a sanity check on the topcfg loader to make sure that it roundtrips
+    # correctly when loading.
+    test_topcfg_loader(genhjson_path, completecfg)
+
     # Generate Rust toplevel definitions
     if not args.no_rust:
         generate_rust(topname, completecfg, name_to_block, out_path.resolve(),

util/topgen/clocks.py

@@ -20,6 +20,10 @@ def _bool_to_yn(val: bool) -> str:
     return 'yes' if val else 'no'
 
 
+def _bool_from_yn(val: str) -> bool:
+    return val == 'yes'
+
+
 def _to_int(val: object) -> int:
     if isinstance(val, int):
         return val
@@ -66,6 +70,14 @@ def _asdict(self) -> Dict[str, object]:
             'ref': self.ref
         }
 
+    @classmethod
+    def fromdict(cls, src: Dict[str, object]) -> object:
+        src['aon'] = _bool_from_yn(src['aon'])
+        src['freq'] = int(src['freq'])
+        c = cls.__new__(cls)
+        c.__dict__.update(**src)
+        return c
+
 
 class DerivedSourceClock(SourceClock):
     '''A derived source clock (divided down from some other clock).'''
@@ -82,6 +94,12 @@ def _asdict(self) -> Dict[str, object]:
         ret['src'] = self.src.name
         return ret
 
+    @classmethod
+    def fromdict(cls, src: Dict[str, object], clocks: Dict[str, SourceClock]) -> object:
+        src['div'] = int(src['div'])
+        src['src'] = clocks[src['src']]
+        return super().fromdict.__func__(cls, src)
+
 
 class ClockSignal:
     '''A clock signal in the design.'''
@@ -141,6 +159,17 @@ def _asdict(self) -> Dict[str, object]:
              for name, sig in self.clocks.items()}
         }
 
+    @classmethod
+    def fromdict(cls, src: Dict[str, object], clocks: Dict[str, SourceClock]) -> object:
+        src['unique'] = _bool_from_yn(src['unique'])
+        src['clocks'] = {
+            name: ClockSignal(name, clocks[sig_src_name])
+            for (name, sig_src_name) in src['clocks'].items()
+        }
+        c = cls.__new__(cls)
+        c.__dict__.update(**src)
+        return c
+
 
 class GroupProxy:
     """
@@ -158,6 +187,14 @@ def _asdict(self):
             "name": self._grp.name
         }
 
+    @classmethod
+    def fromdict(cls, proxy: Dict[str, object], clocks: object) -> object:
+        proxy['_grp'] = clocks.groups[proxy["name"]]
+        del proxy["name"]
+        c = cls.__new__(cls)
+        c.__dict__.update(**proxy)
+        return c
+
 
 class TypedClocks(NamedTuple):
     # External clocks that are consumed only inside the clkmgr and are fed from
@@ -290,6 +327,25 @@ def _asdict(self) -> Dict[str, object]:
             'groups': list(self.groups.values())
         }
 
+    @classmethod
+    def fromdict(cls, clocks: Dict[str, object]) -> object:
+        clocks['srcs'] = {
+            src["name"]: SourceClock.fromdict(src) for src in clocks['srcs']
+        }
+        clocks['derived_srcs'] = {
+            src["name"]: DerivedSourceClock.fromdict(src, clocks['srcs'])
+            for src in clocks['derived_srcs']
+        }
+        clocks['all_srcs'] = clocks['srcs'].copy()
+        clocks['all_srcs'].update(clocks['derived_srcs'])
+        all_clocks = clocks['all_srcs']
+        clocks['groups'] = {
+            src["name"]: Group.fromdict(src, all_clocks) for src in clocks['groups']
+        }
+        c = cls.__new__(cls)
+        c.__dict__.update(clocks)
+        return c
+
     def add_clock_to_group(self, grp: Group, clk_name: str,
                            src_name: str) -> ClockSignal:
         src = self.all_srcs.get(src_name)

util/topgen/merge.py

@@ -917,7 +917,7 @@ def append_to_lpg_dict(lpg_dict):
                 'clock_connection': clock,
                 'unmanaged_clock': unmanaged_clock,
                 'unmanaged_reset': is_unmanaged_reset(top, reset_name),
-                'reset_connection': primary_reset
+                'reset_connection': primary_reset,
             })
 
         alert_group = module.get('outgoing_alert')

util/topgen/resets.py

@@ -68,6 +68,17 @@ def _asdict(self) -> Dict[str, object]:
 
         return ret
 
+    @classmethod
+    def fromdict(cls, item: Dict[str, object], clocks: Clocks) -> object:
+        # rst_type can be None which is serialized as "null", also the name is different
+        item["rst_type"] = None if item["type"] == "null" else item["type"]
+        del item["type"]
+        item["clock"] = clocks.get_clock_by_name(item["clock"])
+        item["parent"] = item.get("parent", "")
+        c = cls.__new__(cls)
+        c.__dict__.update(**item)
+        return c
+
 
 class Resets:
     '''Resets for the chip'''
@@ -93,6 +104,16 @@ def _asdict(self) -> Dict[str, object]:
 
         return ret
 
+    @classmethod
+    def fromdict(cls, resets: Dict[str, object], clocks: Clocks) -> object:
+        # Reconstruct dict.
+        resets['nodes'] = {
+            node["name"]: ResetItem.fromdict(node, clocks) for node in resets['nodes']
+        }
+        c = cls.__new__(cls)
+        c.__dict__.update(**resets)
+        return c
+
     def get_reset_by_name(self, name: str) -> ResetItem:
 
         ret = self.nodes.get(name, None)
@@ -240,6 +261,12 @@ def _asdict(self) -> Dict[str, object]:
             'rst_en_signal_name': self.rst_en_signal_name
         }
 
+    @classmethod
+    def fromdict(cls, param: Dict[str, object]) -> object:
+        c = cls.__new__(cls)
+        c.__dict__.update(**param)
+        return c
+
 
 class UnmanagedResets:
     '''Unmanaged reset connections for the chip.'''
@@ -250,6 +277,16 @@ def __init__(self, raw: List[object]):
     def _asdict(self) -> Dict[str, object]:
         return self.resets
 
+    @classmethod
+    def fromdict(cls, resets: Dict[str, object]) -> object:
+        resets = {
+            'resets': UnmanagedReset.fromdict(r)
+            for r in resets
+        }
+        c = cls.__new__(cls)
+        c.resets = resets
+        return c
+
     def get(self, name: str) -> object:
         try:
             return self.resets[name]

util/topgen/validate.py

@@ -344,7 +344,9 @@ def check_values(self):
             raise ValueError('flash number of banks and pages per bank too large')
 
     def _asdict(self):
+        # Do not include base_addrs as it will get removed later.
         return {
+            'class': 'Flash',
             'banks': self.banks,
             'pages_per_bank': self.pages_per_bank,
             'program_resolution': self.program_resolution,
@@ -357,6 +359,17 @@ def _asdict(self):
             'size': self.size
         }
 
+    @classmethod
+    def fromdict(cls, item: Dict[str, object]) -> object:
+        del item["class"]
+        item["word_bytes"] = int(item["data_width"] / 8)
+        item["words_per_page"] = int(item["bytes_per_page"] / item["word_bytes"])
+        item["pages_per_bank"] = int(item["bytes_per_bank"] / item["bytes_per_page"])
+        item["info_types"] = len(item["infos_per_bank"])
+        c = cls.__new__(cls)
+        c.__dict__.update(**item)
+        return c
+
 
 # Check to see if each module/xbar defined in top.hjson exists as ip/xbar.hjson
 # Also check to make sure there are not multiple definitions of ip/xbar.hjson for each