-import random
-
from migen.fhdl.std import *
from migen.genlib.fsm import *
from migen.actorlib.fifo import *
from migen.flow.actor import EndpointDescription
+from migen.actorlib.packet import Arbiter, Dispatcher
user_layout = EndpointDescription(
[("dst", 8),
("length", 4*8),
("error", 1),
- ("d", 8)
+ ("data", 8)
],
packetized=True
)
-phy_layout = [("d", 8)]
+phy_layout = [("data", 8)]
class LiteUSBPipe:
self.sink = Sink(layout)
self.source = Source(layout)
-
-class LiteUSBTimeout(Module):
- def __init__(self, clk_freq, length):
- cnt_max = int(clk_freq*length)
- width = bits_for(cnt_max)
-
- self.clear = Signal()
- self.done = Signal()
-
- cnt = Signal(width)
- self.sync += \
- If(self.clear,
- cnt.eq(0)
- ).Elif(~self.done,
- cnt.eq(cnt+1)
- )
- self.comb += self.done.eq(cnt == cnt_max)
-
-
-#
-# TB
-#
-def randn(max_n):
- return random.randint(0, max_n-1)
-
-
-class RandRun:
- def __init__(self, level=0):
- self.run = True
- self.level = level
-
- def do_simulation(self, selfp):
- self.run = True
- n = randn(100)
- if n < self.level:
- self.run = False
next CRC value.
"""
def __init__(self, dat_width, width, polynom):
- self.d = Signal(dat_width)
+ self.data = Signal(dat_width)
self.last = Signal(width)
self.next = Signal(width)
if t == "state":
xors += [self.last[n]]
elif t == "din":
- xors += [self.d[n]]
+ xors += [self.data[n]]
self.comb += self.next[i].eq(optree("^", xors))
check = 0xC704DD7B
def __init__(self, dat_width):
- self.d = Signal(dat_width)
+ self.data = Signal(dat_width)
self.value = Signal(self.width)
self.error = Signal()
reg = Signal(self.width, reset=self.init)
self.sync += reg.eq(self.engine.next)
self.comb += [
- self.engine.d.eq(self.d),
+ self.engine.data.eq(self.data),
self.engine.last.eq(reg),
self.value.eq(~reg[::-1]),
# # #
- dw = flen(sink.d)
+ dw = flen(sink.data)
crc = crc_class(dw)
fsm = FSM(reset_state="IDLE")
self.submodules += crc, fsm
)
fsm.act("COPY",
crc.ce.eq(sink.stb & source.ack),
- crc.d.eq(sink.d),
+ crc.data.eq(sink.data),
Record.connect(sink, source),
source.eop.eq(0),
If(sink.stb & sink.eop & source.ack,
cnt_done = Signal()
fsm.act("INSERT",
source.stb.eq(1),
- chooser(crc.value, cnt, source.d, reverse=True),
+ chooser(crc.value, cnt, source.data, reverse=True),
If(cnt_done,
source.eop.eq(1),
If(source.ack, NextState("IDLE"))
fsm.act("INSERT",
source.stb.eq(1),
source.eop.eq(1),
- source.d.eq(crc.value),
+ source.data.eq(crc.value),
If(source.ack, NextState("IDLE"))
)
self.comb += self.busy.eq(~fsm.ongoing("IDLE"))
# # #
- dw = flen(sink.d)
+ dw = flen(sink.data)
crc = crc_class(dw)
self.submodules += crc
ratio = crc.width//dw
NextState("IDLE"),
)
fsm.act("IDLE",
- crc.d.eq(sink.d),
+ crc.data.eq(sink.data),
If(sink.stb & sink.sop & sink.ack,
crc.ce.eq(1),
NextState("COPY")
)
)
fsm.act("COPY",
- crc.d.eq(sink.d),
+ crc.data.eq(sink.data),
If(sink.stb & sink.ack,
crc.ce.eq(1),
If(sink.eop,
from migen.fhdl.std import *
from migen.actorlib.structuring import *
from migen.genlib.fsm import FSM, NextState
+from migen.genlib.misc import Timeout
from misoclib.com.liteusb.common import *
for i, byte in enumerate(header):
chunk = getattr(header_pack.source.payload, "chunk" + str(i))
- self.comb += byte.eq(chunk.d)
+ self.comb += byte.eq(chunk.data)
fsm = FSM()
self.submodules += fsm
- self.comb += preamble[0].eq(sink.d)
+ self.comb += preamble[0].eq(sink.data)
for i in range(1, 4):
self.sync += If(sink.stb & sink.ack,
preamble[i].eq(preamble[i-1])
header_pack.source.ack.eq(1),
)
- self.submodules.timeout = LiteUSBTimeout(60000000, timeout)
- self.comb += self.timeout.clear.eq(fsm.ongoing("WAIT_SOP"))
+ self.submodules.timeout = Timeout(60000000*timeout)
+ self.comb += [
+ self.timeout.reset.eq(fsm.ongoing("WAIT_SOP")),
+ self.timeout.ce.eq(1)
+ ]
fsm.act("RECEIVE_HEADER",
header_pack.sink.stb.eq(sink.stb),
header_pack.sink.payload.eq(sink.payload),
- If(self.timeout.done, NextState("WAIT_SOP"))
+ If(self.timeout.reached, NextState("WAIT_SOP"))
.Elif(header_pack.source.stb, NextState("RECEIVE_PAYLOAD"))
.Else(sink.ack.eq(1))
)
- self.comb += header_pack.reset.eq(self.timeout.done)
+ self.comb += header_pack.reset.eq(self.timeout.reached)
sop = Signal()
eop = Signal()
source.stb.eq(sink.stb),
source.sop.eq(sop),
source.eop.eq(eop),
- source.d.eq(sink.d),
+ source.data.eq(sink.data),
sink.ack.eq(source.ack),
- If((eop & sink.stb & source.ack) | self.timeout.done,
+ If((eop & sink.stb & source.ack) | self.timeout.reached,
NextState("WAIT_SOP")
)
)
self._cnt += 1
selfp.stb = self._stb
- selfp.d = self.data[self._cnt]
+ selfp.data = self.data[self._cnt]
if self._cnt == len(self.data)-1:
raise StopSimulation
for i, byte in enumerate(header):
chunk = getattr(header_unpack.sink.payload, "chunk" + str(i))
- self.comb += chunk.d.eq(byte)
+ self.comb += chunk.data.eq(byte)
fsm = FSM()
self.submodules += fsm
fsm.act("SEND_HEADER",
header_unpack.sink.stb.eq(1),
source.stb.eq(1),
- source.d.eq(header_unpack.source.d),
+ source.data.eq(header_unpack.source.data),
header_unpack.source.ack.eq(source.ack),
If(header_unpack.sink.ack, NextState("SEND_DATA"))
)
fsm.act("SEND_DATA",
source.stb.eq(sink.stb),
- source.d.eq(sink.d),
+ source.data.eq(sink.data),
sink.ack.eq(source.ack),
If(source.ack & sink.eop, NextState("WAIT_SOP"))
)
selfp.eop = self._eop & self._stb
selfp.dst = dst
selfp.length = length
- selfp.d = payload[self._payload_cnt]
+ selfp.data = payload[self._payload_cnt]
if self._frame_cnt == len(self.data):
raise StopSimulation
self.slave = slave
# masters --> slave arbitration
- self.submodules.rr = RoundRobin(len(masters))
- cases = {}
- for i, m in enumerate(masters):
- sop = Signal()
- eop = Signal()
- ongoing = Signal()
- self.comb += [
- sop.eq(m.source.stb & m.source.sop),
- eop.eq(m.source.stb & m.source.eop & m.source.ack),
- ]
- self.sync += ongoing.eq((sop | ongoing) & ~eop)
- self.comb += self.rr.request[i].eq(sop | ongoing)
-
- cases[i] = [Record.connect(masters[i].source, slave.source)]
- self.comb += Case(self.rr.grant, cases)
+ sources = [master.source for master in masters]
+ self.submodules.arbiter = Arbiter(sources, slave.source)
# slave --> master demux
cases = {}
source.sop.eq(cnt == 0),
source.eop.eq(cnt == (self.dma.length*pack_factor-1)),
source.length.eq(self.dma.length*pack_factor+4),
- source.d.eq(unpack.source.d),
+ source.data.eq(unpack.source.data),
source.dst.eq(tag),
unpack.source.ack.eq(source.ack)
]
self.sync += \
If(tx_start,
source.stb.eq(1),
- source.d.eq(self._rxtx.r),
+ source.data.eq(self._rxtx.r),
).Elif(tx_done,
source.stb.eq(0)
)
self.comb += [
rx_fifo.we.eq(sink.stb),
sink.ack.eq(sink.stb & rx_fifo.writable),
- rx_fifo.din.eq(sink.d),
+ rx_fifo.din.eq(sink.data),
rx_available.eq(rx_fifo.readable),
rx_fifo.re.eq(self.ev.rx.clear),
from misoclib.com.liteusb.common import *
-class FT2232HPHY(Module):
+class FT2232HPHYSynchronous(Module):
def __init__(self, pads, fifo_depth=32, read_time=16, write_time=16):
dw = flen(pads.data)
data_w_accepted = Signal(reset=1)
- fsm = FSM()
+ fsm = FSM(reset_state="READ")
self.submodules += RenameClockDomains(fsm, {"sys": "ftdi"})
fsm.act("READ",
pads.wr_n.eq(1)
),
read_buffer.sink.stb.eq(~pads.rd_n & ~rxf_n),
- read_buffer.sink.d.eq(data_r),
+ read_buffer.sink.data.eq(data_r),
If(~txe_n & data_w_accepted,
- data_w.eq(write_fifo.source.d)
+ data_w.eq(write_fifo.source.data)
)
]
self.debug = Signal(8)
self.comb += self.debug.eq(data_r)
-
-
-#
-# TB
-#
-class FT2232HModel(Module, RandRun):
- def __init__(self, rd_data):
- RandRun.__init__(self, 50)
- self.rd_data = [0] + rd_data
- self.rd_idx = 0
-
- # pads
- self.data = Signal(8)
- self.rxf_n = Signal(reset=1)
- self.txe_n = Signal(reset=1)
- self.rd_n = Signal(reset=1)
- self.wr_n = Signal(reset=1)
- self.oe_n = Signal(reset=1)
- self.siwua = Signal()
- self.pwren_n = Signal(reset=1)
-
- self.init = True
- self.wr_data = []
- self.wait_wr_n = False
- self.rd_done = 0
-
- def wr_sim(self, selfp):
- if not selfp.wr_n and not selfp.txe_n:
- self.wr_data.append(selfp.data)
- self.wait_wr_n = False
-
- if not self.wait_wr_n:
- if self.run:
- selfp.txe_n = 1
- else:
- if selfp.txe_n:
- self.wait_wr_n = True
- selfp.txe_n = 0
-
- def rd_sim(self, selfp):
- rxf_n = selfp.rxf_n
- if self.run:
- if self.rd_idx < len(self.rd_data)-1:
- self.rd_done = selfp.rxf_n
- selfp.rxf_n = 0
- else:
- selfp.rxf_n = self.rd_done
- else:
- selfp.rxf_n = self.rd_done
-
- if not selfp.rd_n and not selfp.oe_n:
- if self.rd_idx < len(self.rd_data)-1:
- self.rd_idx += not rxf_n
- selfp.data = self.rd_data[self.rd_idx]
- self.rd_done = 1
-
- def do_simulation(self, selfp):
- RandRun.do_simulation(self, selfp)
- if self.init:
- selfp.rxf_n = 0
- self.wr_data = []
- self.init = False
- self.wr_sim(selfp)
- self.rd_sim(selfp)
-
-
-class UserModel(Module, RandRun):
- def __init__(self, wr_data):
- RandRun.__init__(self, 50)
- self.wr_data = wr_data
- self.wr_data_idx = 0
-
- self.sink = Sink(phy_layout)
- self.source = Source(phy_layout)
-
- self.rd_data = []
-
- def wr_sim(self, selfp):
- auth = True
- if selfp.source.stb and not selfp.source.ack:
- auth = False
- if auth:
- if self.wr_data_idx < len(self.wr_data):
- if self.run:
- selfp.source.d = self.wr_data[self.wr_data_idx]
- selfp.source.stb = 1
- self.wr_data_idx += 1
- else:
- selfp.source.stb = 0
- else:
- self.source.stb = 0
-
- def rd_sim(self, selfp):
- if self.run:
- selfp.sink.ack = 1
- else:
- selfp.sink.ack = 0
- if selfp.sink.stb & selfp.sink.ack:
- self.rd_data.append(selfp.sink.d)
-
- def do_simulation(self, selfp):
- RandRun.do_simulation(self, selfp)
- self.wr_sim(selfp)
- self.rd_sim(selfp)
-
-
-LENGTH = 512
-model_rd_data = [i%256 for i in range(LENGTH)][::-1]
-user_wr_data = [i%256 for i in range(LENGTH)]
-
-
-class TB(Module):
- def __init__(self):
- self.submodules.model = FT2232HModel(model_rd_data)
- self.submodules.phy = FT2232HPHY(self.model)
-
- self.submodules.user = UserModel(user_wr_data)
-
- self.comb += [
- self.user.source.connect(self.phy.sink),
- self.phy.source.connect(self.user.sink)
- ]
-
- # Use sys_clk as ftdi_clk in simulation
- self.comb += [
- ClockSignal("ftdi").eq(ClockSignal()),
- ResetSignal("ftdi").eq(ResetSignal())
- ]
-
-
-def print_results(s, l1, l2):
- def comp(l1, l2):
- r = True
- try:
- for i, val in enumerate(l1):
- if val != l2[i]:
- print(s + " : val : {:02X}, exp : {:02X}".format(val, l2[i]))
- r = False
- except:
- return r
- return r
-
- c = comp(l1, l2)
- r = s + " "
- if c:
- r += "[OK]"
- else:
- r += "[KO]"
- print(r)
-
-
-def main():
- from migen.sim.generic import run_simulation
- tb = TB()
- run_simulation(tb, ncycles=8000, vcd_name="tb_phy.vcd")
-
- # print(tb.user.rd_data)
- # print(tb.model.wr_data)
- # print(len(tb.user.rd_data))
- # print(len(tb.model.wr_data))
-
- print_results("F2232HModel --> UserModel", model_rd_data, tb.user.rd_data)
- print_results("UserModel --> FT2232HModel", user_wr_data, tb.model.wr_data)
-
-if __name__ == "__main__":
- main()
--- /dev/null
+MSCDIR = ../../
+PYTHON = python3
+
+CMD = PYTHONPATH=$(MSCDIR) $(PYTHON)
+
+ft2232h_tb:
+ $(CMD) ft2232h_tb.py
--- /dev/null
+import random
+
+def randn(max_n):
+ return random.randint(0, max_n-1)
+
+
+class RandRun:
+ def __init__(self, level=0):
+ self.run = True
+ self.level = level
+
+ def do_simulation(self, selfp):
+ self.run = True
+ n = randn(100)
+ if n < self.level:
+ self.run = False
--- /dev/null
+from migen.fhdl.std import *
+from migen.flow.actor import *
+from migen.fhdl.specials import *
+
+from migen.sim.generic import run_simulation
+
+from misoclib.com.liteusb.common import *
+from misoclib.com.liteusb.phy.ft2232h import FT2232HPHYSynchronous
+from misoclib.com.liteusb.test.common import *
+
+# XXX for now use it from liteeth to avoid duplication
+from misoclib.com.liteeth.test.common import *
+
+
+def phy_description():
+ payload_layout = [("data", 8)]
+ return EndpointDescription(payload_layout, packetized=True)
+
+
+class FT2232HSynchronousModel(Module, RandRun):
+ def __init__(self, rd_data):
+ RandRun.__init__(self, 10)
+ self.rd_data = [0] + rd_data
+ self.rd_idx = 0
+
+ # pads
+ self.data = Signal(8)
+ self.rxf_n = Signal(reset=1)
+ self.txe_n = Signal(reset=1)
+ self.rd_n = Signal(reset=1)
+ self.wr_n = Signal(reset=1)
+ self.oe_n = Signal(reset=1)
+ self.siwua = Signal()
+ self.pwren_n = Signal(reset=1)
+
+ self.init = True
+ self.wr_data = []
+ self.wait_wr_n = False
+ self.rd_done = 0
+
+
+ self.data_w = Signal(8)
+ self.data_r = Signal(8)
+
+ self.specials += Tristate(self.data, self.data_r, ~self.oe_n, self.data_w)
+
+ def wr_sim(self, selfp):
+ if not selfp.wr_n and not selfp.txe_n:
+ self.wr_data.append(selfp.data_w)
+ self.wait_wr_n = False
+
+ if not self.wait_wr_n:
+ if self.run:
+ selfp.txe_n = 1
+ else:
+ if selfp.txe_n:
+ self.wait_wr_n = True
+ selfp.txe_n = 0
+
+ def rd_sim(self, selfp):
+ rxf_n = selfp.rxf_n
+ if self.run:
+ if self.rd_idx < len(self.rd_data)-1:
+ self.rd_done = selfp.rxf_n
+ selfp.rxf_n = 0
+ else:
+ selfp.rxf_n = self.rd_done
+ else:
+ selfp.rxf_n = self.rd_done
+
+ if not selfp.rd_n and not selfp.oe_n:
+ if self.rd_idx < len(self.rd_data)-1:
+ self.rd_idx += not rxf_n
+ selfp.data_r = self.rd_data[self.rd_idx]
+ self.rd_done = 1
+
+ def do_simulation(self, selfp):
+ RandRun.do_simulation(self, selfp)
+ if self.init:
+ selfp.rxf_n = 0
+ self.wr_data = []
+ self.init = False
+ self.wr_sim(selfp)
+ self.rd_sim(selfp)
+
+test_packet = [i%256 for i in range(512)]
+
+
+class TB(Module):
+ def __init__(self):
+ self.submodules.model = FT2232HSynchronousModel(test_packet)
+ self.submodules.phy = FT2232HPHYSynchronous(self.model)
+
+ self.submodules.streamer = PacketStreamer(phy_description())
+ self.submodules.streamer_randomizer = AckRandomizer(phy_description(), level=10)
+
+ self.submodules.logger_randomizer = AckRandomizer(phy_description(), level=10)
+ self.submodules.logger = PacketLogger(phy_description())
+
+ self.comb += [
+ Record.connect(self.streamer.source, self.streamer_randomizer.sink),
+ self.phy.sink.stb.eq(self.streamer_randomizer.source.stb),
+ self.phy.sink.data.eq(self.streamer_randomizer.source.data),
+ self.streamer_randomizer.source.ack.eq(self.phy.sink.ack),
+
+ self.logger_randomizer.sink.stb.eq(self.phy.source.stb),
+ self.logger_randomizer.sink.data.eq(self.phy.source.data),
+ self.phy.source.ack.eq(self.logger_randomizer.sink.ack),
+ Record.connect(self.logger_randomizer.source, self.logger.sink)
+ ]
+
+ # Use sys_clk as ftdi_clk in simulation
+ self.comb += [
+ ClockSignal("ftdi").eq(ClockSignal()),
+ ResetSignal("ftdi").eq(ResetSignal())
+ ]
+
+ def gen_simulation(self, selfp):
+ yield from self.streamer.send(Packet(test_packet))
+ for i in range(2000):
+ yield
+ s, l, e = check(test_packet, self.model.wr_data)
+ print("shift " + str(s) + " / length " + str(l) + " / errors " + str(e))
+
+ s, l, e = check(test_packet, self.logger.packet[1:])
+ print("shift " + str(s) + " / length " + str(l) + " / errors " + str(e))
+
+
+def main():
+ run_simulation(TB(), ncycles=8000, vcd_name="my.vcd", keep_files=True)
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
projects / litex.git / commitdiff