diff --git a/.gitignore b/.gitignore index 610c399..11fffae 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,7 @@ *.pyc /*.egg-info /.eggs + +# tests +*.vcd +*.gtkw diff --git a/nmigen_stdio/spiflash.py b/nmigen_stdio/spiflash.py new file mode 100644 index 0000000..c2c737b --- /dev/null +++ b/nmigen_stdio/spiflash.py @@ -0,0 +1,410 @@ +from nmigen import * +from nmigen.lib.cdc import FFSynchronizer +from nmigen.lib.io import Pin +from nmigen.build.dsl import Subsignal, Pins +from nmigen.build.plat import Platform +from nmigen.utils import bits_for, log2_int + + +__all__ = ["SPIFlashSlowReader", "SPIFlashFastReader"] + + +class _SPIFlashReaderBase: + """A base class for an SPI flash controller that is to perform any READ operations + """ + CMD_WIDTH = 8 + CMD_DICT = { + "standard": None, + "dual" : None, + "quad" : None + } + + def _format_cmd(self): + """ + Returns the values on all the DQ lines that corresponds to the command. + + Since everything is transmitted on all DQ lines (command, address and data), + the input cmd_value is extended/interleaved to full DQ width + even if DQ1-DQ3 are "don't care" during the command phase: + + eg: 1 1 1 0 1 0 1 1 (standard SPI mode) + 11 11 11 10 11 10 11 11 (dual I/O SPI mode) + 1111 1111 1111 1110 1111 1110 1111 1111 (quad I/O SPI mode) + """ + cmd_value = type(self).CMD_DICT.get(self._protocol) + if cmd_value is None: + raise KeyError("Reader class {} does not support {} SPI protocol" + .format(type(self).__name__, self._protocol)) + fcmd = 2**(type(self).CMD_WIDTH*self.spi_width) - 1 + for bit in range(type(self).CMD_WIDTH): + if (cmd_value >> bit) % 2 == 0: + fcmd &= ~(1 << (bit*self.spi_width)) + return fcmd + + def __init__(self, *, protocol, data_width=32, + divisor=1, divisor_bits=None, pins=None): + if protocol not in ["standard", "dual", "quad"]: + raise ValueError("Invalid SPI protocol {!r}; must be one of " + "\"standard\", \"dual\", or \"quad\"" + .format(protocol)) + self._protocol = protocol + if not isinstance(data_width, int) or data_width < 1: + raise ValueError("Bus data width must be a positive integer, not {}" + .format(repr(data_width))) + if data_width < 8 or data_width % 8: + raise ValueError("Data width must be divisible by 8 ({} % 8 != 0)" + .format(data_width)) + self._data_width = data_width + + if divisor < 1: + raise ValueError("Invalid divisor value; must be at least 1" + .format(divisor)) + self.divisor = Signal(divisor_bits or bits_for(divisor), reset=divisor) + + self._pins = pins + + if self._protocol == "standard": + self.spi_width = 1 + elif self._protocol == "dual": + self.spi_width = 2 + elif self._protocol == "quad": + self.spi_width = 4 + + self.cs = Signal(reset=0) # Equivalent to ~CS_N + self.clk = Signal() + + if protocol == "standard": + self.mosi = Signal() + self.miso = Signal() + elif protocol == "dual": + self.dq = Pin(2, "io") + elif protocol == "quad": + self.dq = Pin(4, "io") + + self.rdy = Signal() # Internal rdy signal, set at idle state + self.ack = Signal() # External ack signal that begins a read + self.r_data = Signal(self._data_width) + self.r_rdy = Signal() # 1 if r_data is valid, 0 otherwise + + self.cmd_width = type(self).CMD_WIDTH + self._fcmd_width = self.cmd_width * self.spi_width + # A two-way register storing the current value on the DQ I/O pins + # (Note: DQ pins are both input/output only for Dual or Quad SPI protocols) + self.shreg = None + self.counter = Signal.like(self.divisor) + self.counter_en = Signal() + self.clk_posedge_next = Signal() + self.clk_negedge_next = Signal() + + def _add_spi_hardware_logic(self, platform, module): + """Add the fundamental hardware logic for controlling all SPI pins to be used + """ + if self.shreg is None: + raise NotImplementedError("Shift register shreg has not been defined!") + shreg = self.shreg + counter = self.counter + counter_en = self.counter_en + + if self._pins is not None: + module.d.comb += self._pins.cs.o.eq(self.cs) + if hasattr(self._pins, "wp"): + module.d.comb += self._pins.wp.eq(0) + if hasattr(self._pins, "hold"): + module.d.comb += self._pins.hold.eq(0) + # Platforms that require declaration of a user SPI clock signal + # (e.g. by instantiating a USRMCLK Instance) must NOT pass a CLK on the SPI flash + if hasattr(self._pins, "clk"): + module.d.comb += self._pins.clk.o.eq(self.clk) + if self._protocol == "standard": + module.submodules += FFSynchronizer(self._pins.miso.i, self.miso) + module.d.comb += self._pins.mosi.o.eq(self.mosi) + elif self._protocol in ["dual", "quad"]: + self.dq_oe = Signal() + self.dq = self._pins.dq + # If the user doesn't give pins, create dq Pins for Dual & Quad + else: + if self._protocol in ["dual", "quad"]: + self.dq_oe = Signal() + if self._protocol == "dual": + dq_pins = Record([ + ("dq", Pin(width=2, dir="io", xdr=0).layout) + ]) + elif self._protocol == "quad": + dq_pins = Record([ + ("dq", Pin(width=4, dir="io", xdr=0).layout) + ]) + tristate_submodule = Module() + tristate_submodule.submodules += Instance("$tribuf", + p_WIDTH=self.dq.width, + i_EN=self.dq.oe, + i_A=self.dq.o, + o_Y=dq_pins + ) + module.submodules.dq = tristate_submodule + + # r_data always reads from shreg + module.d.comb += self.r_data.eq(shreg) + + # Countdown logic for counter based on divisor + # Also implements MISO logic + dq_i = Signal(self.spi_width) + # When counter is enabled: + with module.If(counter_en): + # When countdown is half-way done, clock edge goes up (positive): + # - MISO latches bit/byte from slave + # - slave has been latching from MOSI mid-way (since previous negedge) + with module.If(counter == (self.divisor+1) >> 1): + # Enable SCLK only when CS: + with module.If(self.cs): + module.d.sync += self.clk.eq(1) + # Indicate imminent posedge + module.d.comb += self.clk_posedge_next.eq(1) + if self._protocol == "standard": + module.d.sync += dq_i.eq(self.miso) + elif self._protocol in ["dual", "quad"]: + module.d.sync += dq_i.eq(self.dq.i) + # When countdown reaches 0, clock edge goes down (negative): + # - MOSI latches from shreg by "pushing" old data out from the left of shreg; + # - slave has been outputting MISO mid-way (since previous posedge) + with module.If(counter == 0): + # Enable SCLK only when CS: + with module.If(self.cs): + module.d.sync += self.clk.eq(0) + module.d.sync += counter.eq(self.divisor) + # Indicate imminent negedge + module.d.comb += self.clk_negedge_next.eq(1) + module.d.sync += shreg.eq(Cat(dq_i, shreg[:-self.spi_width])) + # Normal countdown + with module.Else(): + module.d.sync += counter.eq(counter - 1) + # When counter is disabled, reset: + with module.Else(): + module.d.sync += counter.eq(self.divisor) + + # MOSI logic for Standard SPI protocol: + # MOSI always output the leftmost bit of shreg + if self._protocol == "standard": + module.d.comb += self.mosi.eq(shreg[-1]) + # MOSI logic for Dual and Quad SPI protocols: + # Whenever DQ output should be enabled, + # DQ always output the leftmost `spi_width`-wide bits of shreg + elif self._protocol in ["dual", "quad"]: + module.d.comb += [ + self.dq.o.eq(shreg[-self.spi_width:]), + self.dq.oe.eq(self.dq_oe) + ] + + # r_rdy is asserted for only 1 clock cycle + with module.If(self.r_rdy): + module.d.sync += self.r_rdy.eq(0) + + +class SPIFlashSlowReader(_SPIFlashReaderBase, Elaboratable): + """An SPI flash controller module for normal reading + (i.e. only available in Standard protocol and a lower frequency, + but no dummy cycles of waiting are required). + """ + CMD_DICT = { + "standard": 0x03, + "dual" : None, + "quad" : None + } + + def __init__(self, addr_width, *args, **kwargs): + super().__init__(*args, **kwargs) + self._addr_width = addr_width + self.addr = Signal(addr_width) + self.shreg = Signal(max(self._fcmd_width, self._data_width, self._addr_width)) + + def elaborate(self, platform): + m = Module() + self._add_spi_hardware_logic(platform, m) + + shreg = self.shreg + counter = self.counter + counter_en = self.counter_en + + # fcmd: get formatted command based on cmd_dict + fcmd = self._format_cmd() + # addr: convert bus address to byte-sized address + byte_addr = Cat(Repl(0, log2_int(self._data_width // 8)), self.addr) + # FSM + with m.FSM() as fsm: + state_durations = { + "SLOWREAD-CMD" : self.cmd_width//self.spi_width, + "SLOWREAD-ADDR": self._addr_width//self.spi_width, + "SLOWREAD-READ": self._data_width//self.spi_width + } + max_duration = max([dur for state, dur in state_durations.items()]) + # A "count-up" counter for each state of the command + state_counter = Signal(range(max_duration)) + # State: Idling + with m.State("SLOWREAD-IDLE"): + m.d.comb += self.rdy.eq(1) + with m.If(self.ack): + m.d.sync += [ + counter.eq(0), + counter_en.eq(1), + self.clk.eq(0) + ] + m.next = "SLOWREAD-CS" + # State: Chip select + with m.State("SLOWREAD-CS"): + m.d.sync += self.cs.eq(1) + with m.If(self.clk_negedge_next): + m.d.sync += [ + state_counter.eq(0), + shreg[-self.cmd_width:].eq(fcmd) + ] + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(1) + m.next = "SLOWREAD-CMD" + # State: Command, MOSI + with m.State("SLOWREAD-CMD"): + with m.If((state_counter == state_durations["SLOWREAD-CMD"] - 1) & + self.clk_negedge_next): + m.d.sync += [ + state_counter.eq(0), + shreg[-self._addr_width:].eq(byte_addr) + ] + m.next = "SLOWREAD-ADDR" + with m.Elif(self.clk_negedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Address, MOSI + with m.State("SLOWREAD-ADDR"): + with m.If((state_counter == state_durations["SLOWREAD-ADDR"] - 1) & + self.clk_posedge_next): + m.d.sync += state_counter.eq(0) + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(0) + m.next = "SLOWREAD-READ" + with m.Elif(self.clk_negedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Read, MISO + with m.State("SLOWREAD-READ"): + with m.If((state_counter == state_durations["SLOWREAD-READ"] - 1) & + self.clk_posedge_next): + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(0) + m.next = "SLOWREAD-RDYWAIT" + with m.Elif(self.clk_posedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Send r_rdy (for 1 clock period), and then Wait + with m.State("SLOWREAD-RDYWAIT"): + with m.If(self.clk_negedge_next): + m.d.sync += self.cs.eq(0) + with m.If(self.clk_posedge_next): + m.d.sync += [ + self.r_rdy.eq(1), + counter_en.eq(0) + ] + with m.If(self.r_rdy): + m.next = "SLOWREAD-IDLE" + + return m + + +class SPIFlashFastReader(_SPIFlashReaderBase, Elaboratable): + """An SPI flash controller module for fast-reading + (i.e. available in all SPI protocols and a higher frequency, + but dummy cycles of waiting are required) + """ + CMD_DICT = { + "standard": 0x0b, + "dual" : 0x3b, + "quad" : 0x6b + } + + def __init__(self, addr_width, dummy_cycles, *args, **kwargs): + super().__init__(*args, **kwargs) + self._addr_width = addr_width + self._dummy_cycles = dummy_cycles + self.addr = Signal(addr_width) + self.shreg = Signal(max(self._fcmd_width, self._data_width, self._addr_width)) + + def elaborate(self, platform): + m = Module() + self._add_spi_hardware_logic(platform, m) + + shreg = self.shreg + counter = self.counter + counter_en = self.counter_en + + # fcmd: get formatted command based on cmd_dict + fcmd = self._format_cmd() + # addr: convert bus address to byte-sized address + byte_addr = Cat(Repl(0, log2_int(self._data_width // 8)), self.addr) + # FSM + with m.FSM() as fsm: + state_durations = { + "FASTREAD-CMD" : self.cmd_width//self.spi_width, + "FASTREAD-ADDR" : self._addr_width//self.spi_width, + "FASTREAD-WAITREAD": self._dummy_cycles+self._data_width//self.spi_width + } + max_duration = max([dur for state, dur in state_durations.items()]) + # A "count-up" counter for each state of the command + state_counter = Signal(range(max_duration)) + # State: Idling + with m.State("FASTREAD-IDLE"): + m.d.comb += self.rdy.eq(1) + with m.If(self.ack): + m.d.sync += [ + counter.eq(0), + counter_en.eq(1), + self.clk.eq(0) + ] + m.next = "FASTREAD-CS" + # State: Chip select + with m.State("FASTREAD-CS"): + m.d.sync += self.cs.eq(1) + with m.If(self.clk_negedge_next): + m.d.sync += [ + state_counter.eq(0), + shreg[-self.cmd_width:].eq(fcmd) + ] + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(1) + m.next = "FASTREAD-CMD" + # State: Command, MOSI + with m.State("FASTREAD-CMD"): + with m.If((state_counter == state_durations["FASTREAD-CMD"] - 1) & + self.clk_negedge_next): + m.d.sync += [ + state_counter.eq(0), + shreg[-self._addr_width:].eq(byte_addr) + ] + m.next = "FASTREAD-ADDR" + with m.Elif(self.clk_negedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Address, MOSI + with m.State("FASTREAD-ADDR"): + with m.If((state_counter == state_durations["FASTREAD-ADDR"] - 1) & + self.clk_posedge_next): + m.d.sync += state_counter.eq(0) + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(0) + m.next = "FASTREAD-WAITREAD" + with m.Elif(self.clk_negedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Dummy cycles (waiting), and then Read (MISO) + with m.State("FASTREAD-WAITREAD"): + with m.If((state_counter == state_durations["FASTREAD-WAITREAD"] - 1) & + self.clk_posedge_next): + if self._protocol in ["dual", "quad"]: + m.d.sync += self.dq_oe.eq(0) + m.next = "FASTREAD-RDYWAIT" + with m.Elif(self.clk_posedge_next): + m.d.sync += state_counter.eq(state_counter + 1) + # State: Send r_rdy (for 1 clock period), and then Wait + with m.State("FASTREAD-RDYWAIT"): + with m.If(self.clk_negedge_next): + m.d.sync += self.cs.eq(0) + with m.If(self.clk_posedge_next): + m.d.sync += [ + self.r_rdy.eq(1), + counter_en.eq(0) + ] + with m.If(self.r_rdy): + m.next = "FASTREAD-IDLE" + + return m diff --git a/nmigen_stdio/test/test_spiflash.py b/nmigen_stdio/test/test_spiflash.py new file mode 100644 index 0000000..2bac691 --- /dev/null +++ b/nmigen_stdio/test/test_spiflash.py @@ -0,0 +1,165 @@ +import unittest +from nmigen import * +from nmigen.lib.fifo import SyncFIFO +from nmigen.back.pysim import * + +from ..spiflash import * + + +def simulation_test(dut, process): + with Simulator(dut, vcd_file=open("test.vcd", "w")) as sim: + sim.add_clock(1e-6) + sim.add_sync_process(process) + sim.run() + + +class SPIFlashFastReadTestCase(unittest.TestCase): + def divisor_period(self): + for _ in range((yield self.divisor) + 1): + yield + + class SuperNaiveFlash(Elaboratable): + def __init__(self, *, dut, data, data_size, dummy_cycles): + self.dut = dut + self.data = data + self.data_size = data_size + self.dummy_cycles = dummy_cycles + + def elaborate(self, platform): + m = Module() + data_sig = Signal(self.data_size) + recv_data = SyncFIFO(width=self.dut.spi_width, + depth=(self.dut.cmd_width+self.dut._addr_width) // + self.dut.spi_width) + stored_data = SyncFIFO(width=self.dut.spi_width, + depth=self.dut._data_width // self.dut.spi_width) + stored_data_num_left = Signal(range(stored_data.depth+1), + reset=stored_data.depth) + dummy_counter = Signal(range(self.dummy_cycles), + reset=self.dummy_cycles - 1) + m.submodules.recv_data = recv_data + m.submodules.stored_data = stored_data + if self.dut.spi_width == 1: + m.d.comb += recv_data.w_data.eq(self.dut.mosi) + else: + m.d.comb += recv_data.w_data.eq(self.dut.dq.o) + with m.FSM() as fsm: + with m.State("INIT"): + m.d.sync += data_sig.eq(self.data) + with m.If(self.dut.cs): + with m.If(self.dut.clk_posedge_next): + m.d.comb += recv_data.w_en.eq(1) + with m.Else(): + m.d.comb += recv_data.w_en.eq(0) + with m.If(~recv_data.w_rdy & stored_data.w_rdy & self.dut.clk_negedge_next): + m.next = "PUT-DATA" + with m.State("PUT-DATA"): + with m.If((dummy_counter != 0) & self.dut.clk_negedge_next): + m.d.sync += dummy_counter.eq(dummy_counter - 1) + with m.Elif((dummy_counter == 0) & self.dut.clk_negedge_next): + m.d.comb += stored_data.r_en.eq(1) + m.next = "RETURN-DATA" + with m.State("RETURN-DATA"): + with m.If(self.dut.counter == 0): + m.d.comb += stored_data.r_en.eq(1) + with m.Else(): + m.d.comb += stored_data.r_en.eq(0) + # Keep getting data to the FIFO while returning bytes out of it + with m.If(stored_data_num_left != 0): + m.d.comb += [ + stored_data.w_en.eq(1), + stored_data.w_data.eq(data_sig[-self.dut.spi_width:]) + ] + m.d.sync += [ + stored_data_num_left.eq(stored_data_num_left-1), + data_sig.eq(Cat(Repl(0, self.dut.spi_width), + data_sig[:-self.dut.spi_width])) + ] + with m.Else(): + m.d.comb += stored_data.w_en.eq(0) + with m.If((stored_data_num_left != 0) & ~fsm.ongoing("INIT")): + m.d.comb += [ + stored_data.w_en.eq(1), + stored_data.w_data.eq(data_sig[-self.dut.spi_width:]) + ] + m.d.sync += [ + stored_data_num_left.eq(stored_data_num_left-1), + data_sig.eq(Cat(Repl(0, self.dut.spi_width), + data_sig[:-self.dut.spi_width])) + ] + with m.If(stored_data.r_rdy & stored_data.r_en): + if self.dut.spi_width == 1: + m.d.sync += self.dut.miso.eq(stored_data.r_data) + else: + m.d.sync += self.dut.dq.i.eq(stored_data.r_data) + with m.Elif(~fsm.ongoing("RETURN-DATA")): + if self.dut.spi_width == 1: + m.d.sync += self.dut.miso.eq(0) + else: + m.d.sync += self.dut.dq.i.eq(0) + return m + + def test_standard(self): + self.divisor = 9 + self.dut = SPIFlashFastReader(protocol="standard", + addr_width=24, + data_width=32, + divisor=self.divisor, + dummy_cycles=15) + self.flash = (SPIFlashFastReadTestCase. + SuperNaiveFlash(dut=self.dut, + data=0xAABBCCDD, + data_size=32, + dummy_cycles=15)) + self.simple_test() + + def test_dual(self): + self.divisor = 9 + self.dut = SPIFlashFastReader(protocol="dual", + addr_width=24, + data_width=32, + divisor=self.divisor, + dummy_cycles=15) + self.flash = (SPIFlashFastReadTestCase. + SuperNaiveFlash(dut=self.dut, + data=0xAABBCCDD, + data_size=32, + dummy_cycles=15)) + self.simple_test() + + def test_quad(self): + self.divisor = 9 + self.dut = SPIFlashFastReader(protocol="quad", + addr_width=24, + data_width=32, + divisor=self.divisor, + dummy_cycles=15) + self.flash = (SPIFlashFastReadTestCase. + SuperNaiveFlash(dut=self.dut, + data=0xAABBCCDD, + data_size=32, + dummy_cycles=15)) + self.simple_test() + + def simple_test(self): + m = Module() + m.submodules.master = self.dut + m.submodules.slave = self.flash + + def process(): + yield self.dut.addr.eq(0x123456 >> 2) + yield self.dut.ack.eq(1) + while (yield self.dut.rdy): + yield # Wait until it enters CMD state + yield self.dut.ack.eq(0) + while not (yield self.dut.r_rdy): + yield # Wait until it enters RDYWAIT state + self.assertEqual((yield self.dut.r_data), self.flash.data) + # simulate continuous reading, informally + while not (yield self.dut.rdy): + yield + yield self.dut.ack.eq(1) + while not (yield self.dut.cs): + yield # Test fails if simulation got stuck here + + simulation_test(m, process) diff --git a/setup.py b/setup.py index f83315c..5a30776 100644 --- a/setup.py +++ b/setup.py @@ -23,10 +23,11 @@ def local_scheme(version): #long_description="""TODO""", license="BSD", setup_requires=["setuptools_scm"], - install_requires=["nmigen"], + install_requires=["nmigen~=0.1.rc1"], packages=find_packages(), project_urls={ "Source Code": "https://github.com/m-labs/nmigen-stdio", "Bug Tracker": "https://github.com/m-labs/nmigen-stdio/issues", }, ) +