comm_analyzer: analyze generic Wishbone PHYs

Add generic handler for Wishbone PHY devices, that will save the
bus transactions to a VCD file.

Signed-off-by: Drew Risinger <[email protected]>

Author: drewrisinger

artiq/coredevice/comm_analyzer.py

@@ -307,6 +307,83 @@ def process_read(self, address, data, read_slack):
         raise NotImplementedError
 
 
+class GenericWishboneHandler(WishboneHandler):
+    """Generic Wishbone Bus Data Handler.
+
+    Useful for testing out-of-tree PHY devices and viewing their bus transactions
+    in a VCD file.
+    """
+
+    comm_buses_and_width = {
+        "out_address": 8,
+        "out_data": 32,
+        "out_stb": 1,
+        "in_data": 14,
+        "in_stb": 1
+    }
+    def __init__(self, vcd_manager, name, channel):
+        """Create the Wishbone Data Handler.
+
+        Args:
+            vcd_manager (VCDManager): VCD Manager to add this device's channels to.
+            name (str): Name of the coredevice that communicates on this channel.
+        """
+        self.channels = {}
+        with vcd_manager.scope("bus_device/{}(chan{})".format(name, channel)):
+            for bus_name, bus_width in self.comm_buses_and_width.items():
+                self.channels[bus_name] = vcd_manager.get_channel(
+                    "{}/{}".format(name, bus_name),
+                    bus_width
+                )
+
+    def set_channel(self, channel_name, value):
+        """Set channel to a specific value.
+
+        Formats the data properly for interpretation by VCD viewer
+        (on a per-channel basis).
+
+        Args:
+            channel_name (str): Name of the channel to write
+            value (int): Value to be stored to VCD. Should be some sort of integer.
+        """
+        # format all outputs as binary of proper length
+        chan_fmt_string = "{{:0{}b}}".format(self.comm_buses_and_width[channel_name])
+        self.channels[channel_name].set_value(chan_fmt_string.format(value))
+
+    def process_message(self, message):
+        """Process a Wishbone message into VCD events.
+
+        Args:
+            message (typing.Union[InputMessage, OutputMessage]):
+                A message from the coredevice, either an InputMessage
+                (signal from PHY to coredevice) or an OutputMessage
+                (signal from coredevice to PHY).
+        """
+        # TODO: doesn't reset bus (addr/data) to 0 after bus transaction is done
+        # TODO: stb doesn't display width, but edges register when jumping through waveform
+        if isinstance(message, OutputMessage):
+            logger.debug(
+                "Wishbone out @%d adr=0x%02x data=0x%08x",
+                message.timestamp,
+                message.address,
+                message.data
+            )
+            self.set_channel("out_stb", 1)
+            self.set_channel("out_stb", 0)
+            self.set_channel("out_address", message.address)
+            self.set_channel("out_data", message.data)
+        elif isinstance(message, InputMessage):
+            logger.debug(
+                "Wishbone in @%d(ts=%d) data=0x%08x",
+                message.rtio_counter,
+                message.timestamp,
+                message.data
+            )
+            self.set_channel("in_data", message.data)
+            self.set_channel("in_stb", 1)
+            self.set_channel("in_stb", 0)
+
+
 class SPIMasterHandler(WishboneHandler):
     def __init__(self, vcd_manager, name):
         self.channels = {}