luna wrapper in LiteX

luna.py

#
# Copyright (c) 2025 Gwenhael Goavec-Merou <gwenhael.goavec-merou@trabucayre.com>
#
# SPDX-License-Identifier: BSD-2-Clause

import os

import migen

from litex.gen import *

from litex.soc.interconnect     import stream
from litex.soc.interconnect.csr import *

import amaranth

import luna.full_devices
from luna.gateware.architecture.car import PHYResetController

#from gateware.amaranth.amaranth2v_converter import Amaranth2VConverter
from litex.build.amaranth2v_converter import Amaranth2VConverter

# LunaUSBDevice ------------------------------------------------------------------------------------

class LunaUSBDevice(LiteXModule):
    def __init__(self, platform, data_width=8, pads=None, clock_domain="usb", sync_cd_name="usb"):
        self.source  = source = stream.Endpoint([("data", data_width)])
        self.sink    = sink   = stream.Endpoint([("data", data_width)])

        self.connect = migen.Signal()

        self.platform = platform
        self.pads     = pads

        self._cfg = CSRStorage(description="Configuration Register.", fields=[
            CSRField("ulpi_rst", size=1, offset=0, description="ULPI Reset."),
        ])

        # # #

        self.core_params = {}
        self.cd_list     = ["usb"]

        # CDC ACM clock domain converter -----------------------------------------------------------
        if clock_domain != sync_cd_name:
            self.tx_cdc = tx_cdc = stream.ClockDomainCrossing([("data", data_width)],
                cd_from = clock_domain,
                cd_to   = sync_cd_name,
            )
            self.rx_cdc = rx_cdc = stream.ClockDomainCrossing([("data", data_width)],
                cd_from = sync_cd_name,
                cd_to   = clock_domain,
            )
            self.comb += [
                sink.connect(tx_cdc.sink),
                rx_cdc.source.connect(source)
            ]
            sink, source = tx_cdc.source, rx_cdc.sink

        # Clk/Rst ----------------------------------------------------------------------------------
        ulpi_rst = migen.Signal()

        if hasattr(pads, 'rst'):
            self.comb += pads.rst.eq(ulpi_rst)
        elif hasattr(pads, 'rst_n'):
            self.comb += pads.rst_n.eq(~(ulpi_rst | self._cfg.fields.ulpi_rst))

        # Required to be re-added (or a test to check existance)
        if hasattr(pads, 'clk'):
            from migen.genlib.cdc import MultiReg
            cfg_ulpi_rst = Signal()
            self.specials += MultiReg(self._cfg.fields.ulpi_rst, cfg_ulpi_rst)

            self.comb += ResetSignal("usb").eq((cfg_ulpi_rst | ResetSignal(sync_cd_name)))

            # Amaranth Converter.
            if False:
                phy_reset = Signal()
                trigger   = Signal()

                self.phy_reset_controller = PHYResetController()
                self.reset_conv           = Amaranth2VConverter(self.platform,
                    name          = "phy_reset_controller",
                    modules       = self.phy_reset_controller,
                    core_params   = {
                        "i_trigger"   : (cfg_ulpi_rst | ResetSignal(sync_cd_name)),
                        "o_phy_reset" : phy_reset,
                        "i_sync_clk"  : migen.ClockSignal(sync_cd_name),
                        "i_sync_rst"  : migen.ResetSignal(sync_cd_name),
                    },
                    output_dir    = None,
                )

                self.core_params["i_usb_rst"] = phy_reset
            else:
                self.core_params["i_usb_rst"] = migen.ResetSignal("usb")

        # Signals ----------------------------------------------------------------------------------

        from amaranth.hdl.rec import DIR_FANIN, DIR_FANOUT, DIR_NONE
        if hasattr(pads, "data"):
            usb_type     = "ulpi"
            ulpi_data_w  = Signal(8)
            ulpi_data_oe = Signal()
            ulpi_data_r  = Signal(8)

            self.cd_list.append("fast")

            from litex.build.io import SDRTristate
            for i in range(8):
                self.specials += migen.fhdl.specials.Tristate(target=pads.data[i], i=ulpi_data_w[i], oe=ulpi_data_oe, o=ulpi_data_r[i])
            # Resources (ULPI).
            ulpi = amaranth.Record([
                ('data', [('i', 8, DIR_FANIN), ('o', 8, DIR_FANOUT), ('oe', 1, DIR_FANOUT)]),
                ('clk',  [('i', 1, DIR_FANIN)] if hasattr(pads, 'clk') else [('o', 1, DIR_FANOUT)]),
                ('stp',  [('o', 1, DIR_FANOUT)]),
                ('nxt',  [('i', 1, DIR_FANIN)]),
                ('dir',  [('i', 1, DIR_FANIN)]),
                ('rst',  [('o', 1, DIR_FANOUT)]),
            ])
        elif hasattr(pads, "d_p"):
            usb_type = "io"
            self.cd_list.append("usb_io")
            ulpi = amaranth.Record([
                ('d_p',    [('i', 1, DIR_FANIN), ('o', 1, DIR_FANOUT), ('oe', 1, DIR_FANOUT)]),
                ('d_n',    [('i', 1, DIR_FANIN), ('o', 1, DIR_FANOUT), ('oe', 1, DIR_FANOUT)]),
                ("pullup", [('o', 1, DIR_FANOUT)]),
            ])

            ulpi_d_p_w  = Signal()
            ulpi_d_p_oe = Signal()
            ulpi_d_p_r  = Signal()

            ulpi_d_n_w  = Signal()
            ulpi_d_n_oe = Signal()
            ulpi_d_n_r  = Signal()

            from litex.build.io import SDRTristate

            self.specials += [
                SDRTristate(io=pads.d_p, o=ulpi_d_p_r, oe=ulpi_d_p_oe, i=ulpi_d_p_w),
                SDRTristate(io=pads.d_n, o=ulpi_d_n_r, oe=ulpi_d_n_oe, i=ulpi_d_n_w),
            ]

        # Luna/USBSerialDevice ---------------------------------------------------------------------

        self.usb = usb = luna.full_devices.USBSerialDevice(bus=ulpi, idVendor=0x1209, idProduct=0x0001)

        # Connections ------------------------------------------------------------------------------

        # PHY/IOs.
        # --------
        if usb_type == "ulpi":
            self.ulpi_data_w  = ulpi_data_w
            self.ulpi_data_r  = ulpi_data_r
            self.ulpi_data_oe = ulpi_data_oe

            self.core_params.update({
                "i__bus_data.i"  : ulpi_data_w,
                "o__bus_data.o"  : ulpi_data_r,
                "o__bus_data.oe" : ulpi_data_oe,
                "o__bus_stp.o"   : pads.stp,
                "i__bus_nxt.i"   : pads.nxt,
                "i__bus_dir"     : pads.dir,
                "o__bus_rst_o"   : ulpi_rst,
                "o_usb_clk"      : migen.ClockSignal("usb"),
            })
        else:
            self.core_params.update({
                "i__bus_d_p_i"    : ulpi_d_p_w,
                "o__bus_d_p_o"    : ulpi_d_p_r,
                "o__bus_d_p_oe"   : ulpi_d_p_oe,
                "i__bus_d_n_i"    : ulpi_d_n_w,
                "o__bus_d_n_o"    : ulpi_d_n_r,
                "o__bus_d_n_oe"   : ulpi_d_n_oe,
                "o__bus_pullup_o" : pads.pullup,
                "i_usb_clk"       : migen.ClockSignal("usb"),
            })

        self.core_params.update({
            # Controls.
            # ---------
            "i_connect"    : self.connect,

            # Source.
            # -------
            "o_rx_valid"   : source.valid,
            "i_rx_ready"   : source.ready,
            "o_rx_last"    : source.last,
            "o_rx_first"   : source.first,
            "o_rx_payload" : source.data,

            # Sink.
            # -----
            "i_tx_valid"   : sink.valid,
            "o_tx_ready"   : sink.ready,
            "i_tx_last"    : sink.valid,
            "i_tx_first"   : sink.valid,
            "i_tx_payload" : sink.data,

            # Clk/Reset.
            # ----------
            "i_sync_clk"   : migen.ClockSignal(sync_cd_name),
            "i_sync_rst"   : migen.ResetSignal(sync_cd_name),
        })

        if usb_type == "ulpi":
            if hasattr(pads, 'clk'):
                self.core_params.update({
                    "i_fast_clk"   : migen.ClockSignal("fast"),
                    "i_fast_rst"   : migen.ResetSignal("fast"),
                    "i__bus_clk_i" : pads.clk,
                })
            else:
                self.core_params.update({
                    "o__bus_clk_o" : pads.clk_o,
                })
        else:
            self.core_params.update({
                "i_usb_io_clk" : migen.ClockSignal("usb_io"),
                "i_usb_io_rst" : migen.ResetSignal("usb_io"),
            })

    def do_finalize(self):
        # Amaranth Converter -----------------------------------------------------------------------
        self.converter = converter = Amaranth2VConverter(self.platform,
            name          = "usb_cdc_acm",
            module        = self.usb,
            core_params   = self.core_params,
            clock_domains = self.cd_list,
            output_dir    = None,
        )