litesata/test/link_tb.py

   1 from litesata.common import *
   2 from litesata.core.link import LiteSATALink
   3
   4 from litesata.test.common import *
   5 from litesata.test.hdd import *
   6
   7 class LinkStreamer(PacketStreamer):
   8         def __init__(self):
   9                 PacketStreamer.__init__(self, link_description(32), LinkTXPacket)
  10
  11 class LinkLogger(PacketLogger):
  12         def __init__(self):
  13                 PacketLogger.__init__(self, link_description(32), LinkRXPacket)
  14
  15 class TB(Module):
  16         def __init__(self):
  17                 self.submodules.hdd = HDD(
  18                                 link_debug=False, link_random_level=50,
  19                                 transport_debug=False, transport_loopback=True)
  20                 self.submodules.link = InsertReset(LiteSATALink(self.hdd.phy, buffer_depth=512))
  21
  22                 self.submodules.streamer = LinkStreamer()
  23                 self.submodules.streamer_randomizer = Randomizer(link_description(32), level=50)
  24
  25                 self.submodules.logger_randomizer = Randomizer(link_description(32), level=50)
  26                 self.submodules.logger = LinkLogger()
  27
  28                 self.submodules.pipeline = Pipeline(
  29                         self.streamer,
  30                         self.streamer_randomizer,
  31                         self.link,
  32                         self.logger_randomizer,
  33                         self.logger
  34                 )
  35
  36         def gen_simulation(self, selfp):
  37                 for i in range(8):
  38                         streamer_packet = LinkTXPacket([i for i in range(64)])
  39                         yield from self.streamer.send(streamer_packet)
  40                         yield from self.logger.receive()
  41
  42                         # check results
  43                         s, l, e = check(streamer_packet, self.logger.packet)
  44                         print("shift "+ str(s) + " / length " + str(l) + " / errors " + str(e))
  45
  46
  47 if __name__ == "__main__":
  48         run_simulation(TB(), ncycles=2048, vcd_name="my.vcd", keep_files=True)