splink/gen_liteeth.py

#!/usr/bin/env python3

#
# This file is part of LiteEth.
#
# Copyright (c) 2015-2022 Florent Kermarrec <florent@enjoy-digital.fr>
# Copyright (c) 2020 Xiretza <xiretza@xiretza.xyz>
# Copyright (c) 2020 Stefan Schrijvers <ximin@ximinity.net>
# SPDX-License-Identifier: BSD-2-Clause

import argparse
import os
import yaml

from migen import *

from litex.build.generic_platform import *
from litex.build.xilinx.platform import XilinxPlatform
from litex.build.lattice.platform import LatticePlatform

from litex.soc.interconnect import wishbone
from litex.soc.integration.soc_core import *
from litex.soc.integration.builder import *
from litex.soc.integration.soc import SoCRegion

from liteeth.common import *

from liteeth import phy as liteeth_phys
from liteeth.mac import LiteEthMAC
from liteeth.core import LiteEthUDPIPCore
from liteeth.core.udp import LiteEthUDP

# IOs ----------------------------------------------------------------------------------------------

MAC_ADDRESS = 0x00183e02a914
CLK_FREQ = int(80e6)

_io = [
    # Clk / Rst
    ("sys_clock", 0, Pins(1)),
    ("sys_reset", 1, Pins(1)),

    # IP/MAC Address.
    ("ip_address",  0, Pins(32)),

    # MII PHY Pads
    ("mii_eth_clocks", 0,
        Subsignal("tx", Pins(1)),
        Subsignal("rx", Pins(1)),
     ),
    ("mii_eth", 0,
        Subsignal("rst_n",   Pins(1)),
        Subsignal("mdio",    Pins(1)),
        Subsignal("mdc",     Pins(1)),
        Subsignal("rx_dv",   Pins(1)),
        Subsignal("rx_er",   Pins(1)),
        Subsignal("rx_data", Pins(4)),
        Subsignal("tx_en",   Pins(1)),
        Subsignal("tx_data", Pins(4)),
        Subsignal("col",     Pins(1)),
        Subsignal("crs",     Pins(1))
     ),
]


def get_udp_streamer_port_ios(name, data_width):
    return [
        (f"{name}", 0,
            Subsignal("ip_address", Pins(32)),

            # Sink.
            Subsignal("sink_valid", Pins(1)),
            Subsignal("sink_last",  Pins(1)),
            Subsignal("sink_ready", Pins(1)),
            Subsignal("sink_data",  Pins(data_width)),

            # Source.
            Subsignal("source_valid", Pins(1)),
            Subsignal("source_last",  Pins(1)),
            Subsignal("source_ready", Pins(1)),
            Subsignal("source_data",  Pins(data_width)),
         ),
    ]


def get_udp_port_ios(name, data_width):
    return [
        (f"{name}", 0,
            Subsignal("bind_port", Pins(16)),

            # Sink.
            Subsignal("sink_dst_ip_address", Pins(32)),
            Subsignal("sink_dst_port", Pins(16)),

            Subsignal("sink_length", Pins(16)),
            Subsignal("sink_valid", Pins(1)),
            Subsignal("sink_last",  Pins(1)),
            Subsignal("sink_last_be",  Pins(data_width // 8)),
            Subsignal("sink_ready", Pins(1)),
            Subsignal("sink_data",  Pins(data_width)),

            # Source.
            Subsignal("source_src_ip_address", Pins(32)),
            Subsignal("source_src_port", Pins(16)),

            Subsignal("source_length", Pins(16)),
            Subsignal("source_valid", Pins(1)),
            Subsignal("source_last",  Pins(1)),
            Subsignal("source_last_be",  Pins(data_width // 8)),
            Subsignal("source_ready", Pins(1)),
            Subsignal("source_data",  Pins(data_width)),
         ),
    ]


class PHYCore(SoCMini):
    def __init__(self, platform):
        super().__init__(platform, clk_freq=CLK_FREQ)

        # CRG --------------------------------------------------------------------------------------
        self.submodules.crg = CRG(
            platform.request("sys_clock"),
            platform.request("sys_reset")
        )

        # PHY --------------------------------------------------------------------------------------
        phy = liteeth_phys.LiteEthPHYMII
        ethphy = phy(
            clock_pads=platform.request("mii_eth_clocks"),
            pads=platform.request("mii_eth")
        )
        self.submodules.ethphy = ethphy

        # Timing constaints.
        # Generate timing constraints to ensure the "keep" attribute is properly set on the various
        # clocks. This also adds the constraints to the generated .xdc that can then be "imported"
        # in the project using the core.
        eth_rx_clk = getattr(ethphy, "crg", ethphy).cd_eth_rx.clk
        eth_tx_clk = getattr(ethphy, "crg", ethphy).cd_eth_tx.clk
        from liteeth.phy.model import LiteEthPHYModel
        if not isinstance(ethphy, LiteEthPHYModel):
            self.platform.add_period_constraint(
                eth_rx_clk,
                1e9/phy.rx_clk_freq
            )
            self.platform.add_period_constraint(
                eth_tx_clk,
                1e9/phy.tx_clk_freq
            )
            self.platform.add_false_path_constraints(
                self.crg.cd_sys.clk,
                eth_rx_clk,
                eth_tx_clk
            )


class UDPPort(Module, AutoCSR):
    def __init__(self, udp_core: LiteEthUDP, bind_port, data_width=8, cd="sys"):
        udp_port = udp_core.crossbar.get_port(bind_port, dw=data_width, cd=cd)
        self.sink = udp_port.sink
        self.comb += [
            self.sink.src_port.eq(bind_port),
        ]
        self.source = udp_port.source


class UDPCore(PHYCore):
    def __init__(self, platform):
        from liteeth.frontend.stream import LiteEthUDPStreamer

        super().__init__(platform)

        ip_address = platform.request("ip_address")

        # Core
        self.submodules.core = LiteEthUDPIPCore(
            self.ethphy,
            mac_address=MAC_ADDRESS,
            ip_address=ip_address,
            clk_freq=CLK_FREQ,
            dw=8,
            with_sys_datapath=False,
        )

        # DHCP port
        data_width = 32

        platform.add_extension(get_udp_streamer_port_ios(
            "dhcp",
            data_width=data_width,
        ))
        dhcp_ios = platform.request("dhcp")

        dhcp_streamer = LiteEthUDPStreamer(
            self.core.udp,
            ip_address=dhcp_ios.ip_address,
            udp_port=67,
            data_width=data_width,
            tx_fifo_depth=64,
            rx_fifo_depth=64
        )
        self.submodules += dhcp_streamer

        self.comb += [
            # Connect UDP Sink IOs to UDP Steamer.
            dhcp_streamer.sink.valid.eq(dhcp_ios.sink_valid),
            dhcp_streamer.sink.last.eq(dhcp_ios.sink_last),
            dhcp_ios.sink_ready.eq(dhcp_streamer.sink.ready),
            dhcp_streamer.sink.data.eq(dhcp_ios.sink_data),

            # Connect UDP Streamer to UDP Source IOs.
            dhcp_ios.source_valid.eq(dhcp_streamer.source.valid),
            dhcp_ios.source_last.eq(dhcp_streamer.source.last),
            dhcp_streamer.source.ready.eq(dhcp_ios.source_ready),
            dhcp_ios.source_data.eq(dhcp_streamer.source.data),
        ]

        # Pixel port
        data_width = 32

        platform.add_extension(get_udp_port_ios(
            "pixel",
            data_width=data_width,
        ))
        pixel_ios = platform.request("pixel")

        pixel_port = UDPPort(
            self.core.udp,
            bind_port=pixel_ios.bind_port,
            data_width=data_width,
        )
        self.submodules += pixel_port

        self.comb += [
            # Connect UDP Sink IOs to UDPPort
            pixel_port.sink.ip_address.eq(pixel_ios.sink_dst_ip_address),
            pixel_port.sink.dst_port.eq(pixel_ios.sink_dst_port),

            pixel_port.sink.length.eq(pixel_ios.sink_length),
            pixel_port.sink.valid.eq(pixel_ios.sink_valid),
            pixel_port.sink.last.eq(pixel_ios.sink_last),
            pixel_port.sink.last_be.eq(pixel_ios.sink_last_be),
            pixel_ios.sink_ready.eq(pixel_port.sink.ready),
            pixel_port.sink.data.eq(pixel_ios.sink_data),

            # Connect UDPPort to UDP Source IOs.
            pixel_ios.source_src_ip_address.eq(pixel_port.source.ip_address),
            pixel_ios.source_src_port.eq(pixel_port.source.src_port),

            pixel_ios.source_length.eq(pixel_port.source.length),
            pixel_ios.source_valid.eq(pixel_port.source.valid),
            pixel_ios.source_last.eq(pixel_port.source.last),
            pixel_ios.source_last_be.eq(pixel_port.source.last_be),
            pixel_port.source.ready.eq(pixel_ios.source_ready),
            pixel_ios.source_data.eq(pixel_port.source.data),
        ]

# Build --------------------------------------------------------------------------------------------


def main():
    parser = argparse.ArgumentParser(
        description="splink LiteEth core generator"
    )
    builder_args(parser)
    parser.set_defaults(output_dir="build")
    args = parser.parse_args()

    # Generate core --------------------------------------------------------------------------------
    platform = XilinxPlatform(
        "xc7a35ticsg324-1L",
        io=[],
        toolchain="yosys+nextpnr"
    )
    platform.add_extension(_io)

    soc = UDPCore(platform)

    builder_arguments = builder_argdict(args)
    builder_arguments["compile_gateware"] = False
    if builder_arguments["csr_csv"] is None:
        builder_arguments["csr_csv"] = os.path.join(
            builder_arguments["output_dir"],
            "csr.csv"
        )

    builder = Builder(soc, **builder_arguments)
    builder.build(build_name="liteeth_core")


if __name__ == "__main__":
    main()