litex

#
# This file is part of LiteX.
#
# Copyright (c) 2019-2026 Florent Kermarrec <florent@enjoy-digital.fr>
# SPDX-License-Identifier: BSD-2-Clause

import unittest

from migen import *

from litex.soc.interconnect import csr
from litex.soc.interconnect import csr_bus


def csr32_write(dut, adr, dat):
    for i in range(4):
        yield from dut.csr.write(adr + 3 - i, (dat >> 8*i) & 0xff)


def csr32_read(dut, adr):
    dat = 0
    for i in range(5):
        dat |= ((yield from dut.csr.read(adr + 3 - i)) << 8*i)
    return dat >> 8


class CSRModule(Module, csr.AutoCSR):
    def __init__(self):
        self._csr      = csr.CSR()
        self._constant = csr.CSRConstant(0x12345678)
        self._storage  = csr.CSRStorage(32, reset=0x12345678, write_from_dev=True)
        self._status   = csr.CSRStatus(32, reset=0x12345678)

        # # #

        # When csr is written:
        # - Set storage to 0xdeadbeef.
        # - Set status to storage value.
        self.comb += [
            If(self._csr.wr_stb,
                self._storage.we.eq(1),
                self._storage.dat_w.eq(0xdeadbeef)
            )
        ]
        self.sync += [
            If(self._csr.wr_stb,
                self._status.status.eq(self._storage.storage)
            )
        ]


class CSRDUT(Module):
    def address_map(self, name, memory):
            return {"csrmodule": 0}[name]

    def __init__(self):
        self.csr = csr_bus.Interface()
        self.submodules.csrmodule    = CSRModule()
        self.submodules.csrbankarray = csr_bus.CSRBankArray(
            source      = self,
            address_map = self.address_map,
            )
        self.submodules.csrcon = csr_bus.Interconnect(
            master = self.csr,
            slaves = self.csrbankarray.get_buses()
        )

class TestCSR(unittest.TestCase):
    def test_csr_signal_aliases(self):
        simple = csr.CSR(8, name="simple")
        self.assertIs(simple.wr_data, simple.r)
        self.assertIs(simple.wr_stb,  simple.re)
        self.assertIs(simple.rd_data, simple.w)
        self.assertIs(simple.rd_stb,  simple.we)

        status = csr.CSRStatus(8, name="status")
        self.assertIs(status.rd_stb, status.we)
        self.assertIs(status.wr_stb, status.re)

        pending = csr.CSRStatus(8, read_only=False, name="pending")
        self.assertIs(pending.wr_data, pending.r)
        self.assertIs(pending.rd_stb,  pending.we)
        self.assertIs(pending.wr_stb,  pending.re)

        storage = csr.CSRStorage(8, name="storage")
        self.assertIs(storage.wr_stb, storage.re)

    def test_csr_constant(self):
        def generator(dut):
            self.assertEqual(hex((yield from dut.csrmodule._constant.read())), hex(0x12345678))

        dut = CSRDUT()
        run_simulation(dut, generator(dut))

    def test_csr_storage(self):
        def generator(dut):
            # check init value
            self.assertEqual(hex((yield from csr32_read(dut, 5))), hex(0x12345678))

            # check writes
            yield from csr32_write(dut, 1, 0x5a5a5a5a)
            self.assertEqual(hex((yield from csr32_read(dut, 1))), hex(0x5a5a5a5a))
            yield from csr32_write(dut, 1, 0xa5a5a5a5)
            self.assertEqual(hex((yield from csr32_read(dut, 1))), hex(0xa5a5a5a5))

            # check update from dev
            yield from dut.csr.write(0, 1)
            self.assertEqual(hex((yield from csr32_read(dut, 1))), hex(0xdeadbeef))

        dut = CSRDUT()
        run_simulation(dut, generator(dut))

    def test_csr_status(self):
        def generator(dut):
            # check init value
            self.assertEqual(hex((yield from csr32_read(dut, 1))), hex(0x12345678))

            # check writes (no effect)
            yield from csr32_write(dut, 5, 0x5a5a5a5a)
            self.assertEqual(hex((yield from csr32_read(dut, 5))), hex(0x12345678))
            yield from csr32_write(dut, 5, 0xa5a5a5a5)
            self.assertEqual(hex((yield from csr32_read(dut, 5))), hex(0x12345678))

            # check update from dev
            yield from dut.csr.write(0, 1)
            yield from dut.csr.write(0, 1)
            self.assertEqual(hex((yield from csr32_read(dut, 5))), hex(0xdeadbeef))

        dut = CSRDUT()
        run_simulation(dut, generator(dut))

    def test_csr_fields(self):
        def generator(dut):
            # check reset values
            self.assertEqual((yield dut._storage.fields.foo), 0xa)
            self.assertEqual((yield dut._storage.fields.bar), 0x5a)
            self.assertEqual((yield dut._storage.storage), 0x5a000a)
            self.assertEqual((yield from dut._storage.read()), 0x5a000a)
            yield
            yield
            self.assertEqual((yield dut._status.fields.foo), 0xa)
            self.assertEqual((yield dut._status.fields.bar), 0x5a)
            self.assertEqual((yield from dut._status.read()), 0x5aa0)

        class DUT(Module):
            def __init__(self):
                self._storage = csr.CSRStorage(fields=[
                    csr.CSRField("foo", size=4, offset=0,  reset=0xa,  description="foo"),
                    csr.CSRField("bar", size=8, offset=16, reset=0x5a, description="bar")
                ])
                self._status = csr.CSRStatus(fields=[
                    csr.CSRField("foo", size=4, offset=4, description="foo"),
                    csr.CSRField("bar", size=8, offset=8, description="bar")
                ])
                self.comb += [
                    self._status.fields.foo.eq(self._storage.fields.foo),
                    self._status.fields.bar.eq(self._storage.fields.bar),
                ]
        dut = DUT()
        run_simulation(dut, generator(dut))

    def test_csrstatus_read_reflects_field_drivers(self):
        # Regression for issue #1247: CSRStatus.read() must return the live
        # values driven on fields, not the fields' reset values.
        class DUT(Module):
            def __init__(self):
                self.version = csr.CSRStatus(fields=[
                    csr.CSRField("version",  size=8, offset=0, reset=0x1),
                    csr.CSRField("revision", size=8, offset=8, reset=0xf),
                ])
                self.comb += [
                    self.version.fields.version.eq(0x1),
                    self.version.fields.revision.eq(0x5),
                ]

        dut = DUT()

        def gen():
            yield
            self.assertEqual((yield dut.version.fields.version),  0x1)
            self.assertEqual((yield dut.version.fields.revision), 0x5)
            self.assertEqual((yield from dut.version.read()),     0x501)

        run_simulation(dut, gen())

    def test_fixed_csr_locations(self):
        class DUT(Module, csr.AutoCSR):
            def __init__(self):
                self._auto  = csr.CSRStorage(name="auto")
                self._zero  = csr.CSRStorage(name="zero", n=0)
                self._three = csr.CSRStatus(name="three", n=3)

        dut = DUT()
        csrs = dut.get_csrs(sort=True)

        self.assertEqual([c.name for c in csrs], ["zero", "auto", "reserved2", "three"])
        self.assertIs(csrs[0], dut._zero)
        self.assertIs(csrs[1], dut._auto)
        self.assertIs(csrs[3], dut._three)

    def test_fixed_csr_location_conflict_rejected(self):
        class DUT(Module, csr.AutoCSR):
            def __init__(self):
                self._csr0 = csr.CSRStorage(name="csr0", n=0)
                self._csr1 = csr.CSRStorage(name="csr1", n=0)

        with self.assertRaisesRegex(ValueError, "CSR conflict"):
            DUT().get_csrs(sort=True)

    def test_fixed_csr_location_rejects_negative(self):
        with self.assertRaisesRegex(ValueError, "non-negative"):
            csr.CSRStorage(name="negative", n=-1)

    def test_fixed_csr_constant_sort_does_not_create_reserved_constants(self):
        class DUT(Module, csr.AutoCSR):
            def __init__(self):
                self._constant = csr.CSRConstant(0x12345678, name="constant", n=3)

        constants = DUT().get_constants(sort=True)

        self.assertEqual([c.name for c in constants], ["constant"])
        self.assertEqual(constants[0].constant, 0x12345678)

    # Additional focused tests on the CSRStorage / CSR* primitives themselves, without the CSR
    # bus plumbing of CSRDUT. These exercise behaviour that is directly observable via the
    # simulation-friendly `read()` / `write()` generators on each primitive.

    def test_csr_storage_reset_value(self):
        class DUT(Module):
            def __init__(self):
                self._r = csr.CSRStorage(32, reset=0xcafe_babe)
        dut = DUT()

        def gen():
            self.assertEqual((yield from dut._r.read()), 0xcafe_babe)
            yield from dut._r.write(0x1234_5678)
            self.assertEqual((yield from dut._r.read()), 0x1234_5678)
        run_simulation(dut, gen())

    def test_csr_storage_pulse_field(self):
        # A pulse field is only 1 for the cycle of the CSR write, then back to 0.
        class DUT(Module):
            def __init__(self):
                self._r = csr.CSRStorage(fields=[
                    csr.CSRField("trig", size=1, offset=0, pulse=True, description="trig"),
                    csr.CSRField("keep", size=1, offset=1, description="kept"),
                ])
        dut = DUT()

        def gen():
            # Write both bits high. `trig` pulses during the write cycle; `keep` latches.
            yield from dut._r.write(0b11)
            # Give one more cycle so the pulse reset in write()'s tail propagates.
            yield
            self.assertEqual((yield dut._r.fields.trig), 0)
            self.assertEqual((yield dut._r.fields.keep), 1)
        run_simulation(dut, gen())

    def test_csrstatus_read_reflects_driver(self):
        # CSRStatus.read() returns the current value of `.status` (async read).
        class DUT(Module):
            def __init__(self):
                self.src    = Signal(8)
                self._s     = csr.CSRStatus(8)
                self.comb  += self._s.status.eq(self.src)
        dut = DUT()

        def gen():
            yield dut.src.eq(0xA5)
            yield
            self.assertEqual((yield from dut._s.read()), 0xA5)
            yield dut.src.eq(0x42)
            yield
            self.assertEqual((yield from dut._s.read()), 0x42)
        run_simulation(dut, gen())

    def test_csr_storage_write_out_of_range_rejected(self):
        class DUT(Module):
            def __init__(self):
                self._r = csr.CSRStorage(4)
        dut = DUT()

        def gen():
            with self.assertRaises(ValueError):
                yield from dut._r.write(0x100)  # 9-bit value into 4-bit storage.
        run_simulation(dut, gen())

    def test_csr_storage_atomic_write(self):
        # A 32-bit CSRStorage(atomic_write=True) on an 8-bit bus splits into 4 simple CSRs.
        # With ordering="big" (default), the highest-index word lives at the lowest bus
        # address. The atomic-write protocol holds writes to the upper words in a backstore
        # signal and only commits to `storage` when the lowest word is written. We exercise
        # that explicitly here.
        class CSRModule(Module, csr.AutoCSR):
            def __init__(self):
                self._r = csr.CSRStorage(32, reset=0xDEAD_BEEF, atomic_write=True)

        class CSRDUT2(Module):
            def address_map(self, name, memory):
                return {"mod": 0}[name]

            def __init__(self):
                self.csr = csr_bus.Interface(data_width=8)
                self.submodules.mod = CSRModule()
                self.submodules.bankarray = csr_bus.CSRBankArray(
                    source      = self,
                    address_map = self.address_map,
                    data_width  = 8,
                )
                self.submodules.con = csr_bus.Interconnect(
                    master = self.csr,
                    slaves = self.bankarray.get_buses(),
                )

        dut = CSRDUT2()

        def gen():
            # Storage starts at the reset value.
            self.assertEqual((yield dut.mod._r.storage), 0xDEAD_BEEF)
            # Write the three upper bytes (addr 0..2 with ordering="big"). Storage must NOT
            # update yet — it stays at the reset value while backstore fills.
            yield from dut.csr.write(0, 0xCA)
            yield from dut.csr.write(1, 0xFE)
            yield from dut.csr.write(2, 0xBA)
            yield
            self.assertEqual((yield dut.mod._r.storage), 0xDEAD_BEEF)
            # Writing the lowest-address word last commits everything atomically.
            yield from dut.csr.write(3, 0xBE)
            yield
            yield
            self.assertEqual((yield dut.mod._r.storage), 0xCAFE_BABE)

        run_simulation(dut, gen())