c4m/nmigen/jtag/tap.py

   1 #!/bin/env python3
   2 import os
   3
   4 from nmigen import *
   5 from nmigen.build import *
   6 from nmigen.lib.io import *
   7 from nmigen.tracer import get_var_name
   8
   9 from c4m_repo.nmigen.lib import Wishbone
  10
  11 from .bus import Interface
  12
  13 __all__ = [
  14     "TAP",
  15 ]
  16
  17
  18 class ShiftReg(Elaboratable):
  19     def __init__(self, ircodes, length, domain):
  20         # The sr record will be returned to user code
  21         self.sr = Record([("i", length), ("o", length), ("oe", len(ircodes)), ("ack", 1)])
  22         # The next attributes are for JTAG class usage only
  23         self.ir = None # made None as width is not known yet
  24         self.tdi = Signal()
  25         self.tdo = Signal()
  26         self.tdo_en = Signal()
  27         self.capture = Signal()
  28         self.shift = Signal()
  29         self.update = Signal()
  30         self.jtag_cd = None # The JTAG clock domain
  31
  32         ##
  33
  34         self._ircodes = ircodes
  35         self._domain = domain
  36
  37     def elaborate(self, platform):
  38         length = len(self.sr.o)
  39         domain = self._domain
  40
  41         m = Module()
  42
  43         m.domains.jtag = self.jtag_cd
  44
  45         sr_jtag = Signal(length)
  46
  47         assert isinstance(self.ir, Signal)
  48         isir = Signal(len(self._ircodes))
  49         capture = Signal()
  50         shift = Signal()
  51         update = Signal()
  52         m.d.comb += [
  53             isir.eq(Cat(self.ir == ircode for ircode in self._ircodes)),
  54             capture.eq((isir != 0) & self.capture),
  55             shift.eq((isir != 0) & self.shift),
  56             update.eq((isir != 0) & self.update),
  57         ]
  58
  59         # On update set o, oe and wait for ack
  60         # update signal is on JTAG clockdomain, latch it
  61         update_core = Signal()
  62         m.d[domain] += update_core.eq(update) # This is CDC from JTAG domain to given domain
  63         with m.FSM(domain=domain):
  64             with m.State("IDLE"):
  65                 m.d.comb += self.sr.oe.eq(0)
  66                 with m.If(update_core):
  67                     # Latch sr_jtag cross domain but it should be stable due to latching of update_core
  68                     m.d[domain] += self.sr.o.eq(sr_jtag)
  69                     # Wait one cycle to raise oe so sr.o has one more cycle to stabilize
  70                     m.next = "WAIT4ACK"
  71             with m.State("WAIT4ACK"):
  72                 m.d.comb += self.sr.oe.eq(isir)
  73                 with m.If(self.sr.ack):
  74                     m.next = "WAIT4END"
  75             with m.State("WAIT4END"):
  76                 m.d.comb += self.sr.oe.eq(0)
  77                 with m.If(~update_core):
  78                     m.next = "IDLE"
  79
  80         m.d.comb += [
  81             self.tdo.eq(sr_jtag[0]),
  82             self.tdo_en.eq(shift),
  83         ]
  84
  85         with m.If(shift):
  86             m.d.jtag += sr_jtag.eq(Cat(sr_jtag[1:], self.tdi))
  87         with m.If(capture):
  88             m.d.jtag += sr_jtag.eq(self.sr.i)
  89
  90         return m
  91
  92 class JTAGWishbone(Elaboratable):
  93     def __init__(self, sr_addr, sr_data, wb, domain):
  94         self._sr_addr = sr_addr
  95         self._sr_data = sr_data
  96         self._wb = wb
  97         self._domain = domain
  98
  99         # To be set by JTAG
 100         self._ir = None
 101
 102     def elaborate(self, platform):
 103         sr_addr = self._sr_addr
 104         sr_data = self._sr_data
 105         wb = self._wb
 106         domain = self._domain
 107         ir = self._ir
 108
 109         m = Module()
 110
 111         if hasattr(wb, "sel"):
 112             # Always selected
 113             m.d.comb += [s.eq(1) for s in wb.sel]
 114
 115         # Immediately ack oe
 116         m.d[domain] += [
 117             sr_addr.ack.eq(sr_addr.oe),
 118             sr_data.ack.eq(sr_data.oe != 0),
 119         ]
 120
 121         with m.FSM(domain=domain) as fsm:
 122             with m.State("IDLE"):
 123                 m.d.comb += [
 124                     wb.cyc.eq(0),
 125                     wb.stb.eq(0),
 126                     wb.we.eq(0),
 127                 ]
 128                 with m.If(sr_addr.oe): # WBADDR code
 129                     m.d[domain] += wb.addr.eq(sr_addr.o)
 130                     m.next = "READ"
 131                 with m.If(sr_data.oe[0]): # WBREAD code
 132                     m.d[domain] += wb.addr.eq(wb.addr + 1)
 133                     m.next = "READ"
 134                 with m.If(sr_data.oe[1]): # WBWRITE code
 135                     m.d[domain] += wb.dat_w.eq(sr_data.o)
 136                     m.next = "WRITEREAD"
 137             with m.State("READ"):
 138                 m.d.comb += [
 139                     wb.cyc.eq(1),
 140                     wb.stb.eq(1),
 141                     wb.we.eq(0),
 142                 ]
 143                 with m.If(~wb.stall):
 144                     m.next = "READACK"
 145             with m.State("READACK"):
 146                 m.d.comb += [
 147                     wb.cyc.eq(1),
 148                     wb.stb.eq(0),
 149                     wb.we.eq(0),
 150                 ]
 151                 with m.If(wb.ack):
 152                     m.d[domain] += sr_data.i.eq(wb.dat_r)
 153                     m.next = "IDLE"
 154             with m.State("WRITEREAD"):
 155                 m.d.comb += [
 156                     wb.cyc.eq(1),
 157                     wb.stb.eq(1),
 158                     wb.we.eq(1),
 159                 ]
 160                 with m.If(~wb.stall):
 161                     m.next = "WRITEREADACK"
 162             with m.State("WRITEREADACK"):
 163                 m.d.comb += [
 164                     wb.cyc.eq(1),
 165                     wb.stb.eq(0),
 166                     wb.we.eq(0),
 167                 ]
 168                 with m.If(wb.ack):
 169                     m.d[domain] += wb.addr.eq(wb.addr + 1)
 170                     m.next = "READ"
 171
 172         return m
 173
 174
 175 class TAP(Elaboratable):
 176     #TODO: Document TAP
 177     @staticmethod
 178     def _add_files(platform, prefix):
 179         d = os.path.realpath("{dir}{sep}{par}{sep}{par}{sep}vhdl{sep}jtag".format(
 180             dir=os.path.dirname(__file__), sep=os.path.sep, par=os.path.pardir
 181         )) + os.path.sep
 182         for fname in [
 183             "c4m_jtag_pkg.vhdl",
 184             "c4m_jtag_idblock.vhdl",
 185             "c4m_jtag_iocell.vhdl",
 186             "c4m_jtag_ioblock.vhdl",
 187             "c4m_jtag_irblock.vhdl",
 188             "c4m_jtag_tap_fsm.vhdl",
 189             "c4m_jtag_tap_controller.vhdl",
 190         ]:
 191             f = open(d + fname, "r")
 192             platform.add_file(prefix + fname, f)
 193             f.close()
 194
 195
 196     def __init__(
 197         self, io_count, *, with_reset=False, ir_width=None,
 198         manufacturer_id=Const(0b10001111111, 11), part_number=Const(1, 16),
 199         version=Const(0, 4),
 200         name=None, src_loc_at=0
 201     ):
 202         assert(isinstance(io_count, int) and io_count > 0)
 203         assert((ir_width is None) or (isinstance(ir_width, int) and ir_width >= 2))
 204         assert(len(version) == 4)
 205
 206         self.name = name if name is not None else get_var_name(depth=src_loc_at+2, default="TAP")
 207         self.bus = Interface(with_reset=with_reset, name=self.name+"_bus",
 208                              src_loc_at=src_loc_at+1)
 209
 210         # TODO: Handle IOs with different directions
 211         self.core = Array(Pin(1, "io") for _ in range(io_count)) # Signals to use for core
 212         self.pad  = Array(Pin(1, "io") for _ in range(io_count)) # Signals going to IO pads
 213
 214         self.jtag_cd = ClockDomain(name="jtag", local=True) # Own clock domain using TCK as clock signal
 215
 216         ##
 217
 218         self._io_count = io_count
 219         self._ir_width = ir_width
 220         self._manufacturer_id = manufacturer_id
 221         self._part_number = part_number
 222         self._version = version
 223
 224         self._ircodes = [0, 1, 2] # Already taken codes, all ones added at the end
 225         self._srs = []
 226
 227         self._wbs = []
 228
 229     def elaborate(self, platform):
 230         TAP._add_files(platform, "jtag" + os.path.sep)
 231
 232         m = Module()
 233
 234         tdo_jtag = Signal()
 235         reset = Signal()
 236         capture = Signal()
 237         shift = Signal()
 238         update = Signal()
 239
 240
 241         ir_max = max(self._ircodes) + 1 # One extra code needed with all ones
 242         ir_width = len("{:b}".format(ir_max))
 243         if self._ir_width is not None:
 244             assert self._ir_width >= ir_width, "Specified JTAG IR width not big enough for allocated shiift registers"
 245             ir_width = self._ir_width
 246         ir = Signal(ir_width)
 247
 248         core_i = Cat(pin.i for pin in self.core)
 249         core_o = Cat(pin.o for pin in self.core)
 250         core_oe = Cat(pin.oe for pin in self.core)
 251         pad_i = Cat(pin.i for pin in self.pad)
 252         pad_o = Cat(pin.o for pin in self.pad)
 253         pad_oe = Cat(pin.oe for pin in self.pad)
 254
 255         params = {
 256             "p_IOS": self._io_count,
 257             "p_IR_WIDTH": ir_width,
 258             "p_MANUFACTURER": self._manufacturer_id,
 259             "p_PART_NUMBER": self._part_number,
 260             "p_VERSION": self._version,
 261             "i_TCK": self.bus.tck,
 262             "i_TMS": self.bus.tms,
 263             "i_TDI": self.bus.tdi,
 264             "o_TDO": tdo_jtag,
 265             "i_TRST_N": Const(1),
 266             "o_RESET": reset,
 267             "o_DRCAPTURE": capture,
 268             "o_DRSHIFT": shift,
 269             "o_DRUPDATE": update,
 270             "o_IR": ir,
 271             "o_CORE_IN": core_i,
 272             "i_CORE_OUT": core_o,
 273             "i_CORE_EN": core_oe,
 274             "i_PAD_IN": pad_i,
 275             "o_PAD_OUT": pad_o,
 276             "o_PAD_EN": pad_oe,
 277         }
 278         m.submodules.tap = Instance("c4m_jtag_tap_controller", **params)
 279
 280         m.d.comb += [
 281             self.jtag_cd.clk.eq(self.bus.tck),
 282             self.jtag_cd.rst.eq(reset),
 283         ]
 284
 285         for i, sr in enumerate(self._srs):
 286             m.submodules["sr{}".format(i)] = sr
 287             sr.ir = ir
 288             m.d.comb += [
 289                 sr.tdi.eq(self.bus.tdi),
 290                 sr.capture.eq(capture),
 291                 sr.shift.eq(shift),
 292                 sr.update.eq(update),
 293             ]
 294
 295         if len(self._srs) > 0:
 296             first = True
 297             for sr in self._srs:
 298                 if first:
 299                     first = False
 300                     with m.If(sr.tdo_en):
 301                         m.d.comb += self.bus.tdo.eq(sr.tdo)
 302                 else:
 303                     with m.Elif(sr.tdo_en):
 304                         m.d.comb += self.bus.tdo.eq(sr.tdo)
 305             with m.Else():
 306                 m.d.comb += self.bus.tdo.eq(tdo_jtag)
 307         else:
 308             m.d.comb += self.bus.tdo.eq(tdo_jtag)
 309
 310         for i, wb in enumerate(self._wbs):
 311             m.submodules["wb{}".format(i)] = wb
 312             wb._ir = ir
 313
 314         return m
 315
 316
 317     def add_shiftreg(self, ircode, length, domain="sync"):
 318         """Add a shift register to the JTAG interface
 319
 320         Parameters:
 321         - ircode: code(s) for the IR; int or sequence of ints. In the latter case this
 322           shiftreg is shared between different IR codes.
 323         - length: the length of the shift register
 324         - domain: the domain on which the signal will be used"""
 325
 326         try:
 327             ir_it = iter(ircode)
 328             ircodes = ircode
 329         except TypeError:
 330             ir_it = ircodes = (ircode,)
 331         for _ircode in ir_it:
 332             assert(isinstance(_ircode, int) and _ircode > 0 and _ircode not in self._ircodes)
 333
 334         sr = ShiftReg(ircodes, length, domain)
 335         sr.jtag_cd = self.jtag_cd
 336         self._ircodes.extend(ircodes)
 337         self._srs.append(sr)
 338
 339         return sr.sr
 340
 341
 342     def add_wishbone(self, ircodes, address_width, data_width, sel_width=None, domain="sync"):
 343         """Add a wishbone interface
 344
 345         Parameters:
 346         - ircodes: sequence of three integer for the JTAG IR codes;
 347           they represent resp. WBADDR, WBREAD and WBREADWRITE. First code
 348           has a shift register of length 'address_width', the two other codes
 349           share a shift register of length data_width.
 350         - address_width: width of the address
 351         - data_width: width of the data"""
 352
 353         assert len(ircodes) == 3
 354
 355         sr_addr = self.add_shiftreg(ircodes[0], address_width, domain=domain)
 356         sr_data = self.add_shiftreg(ircodes[1:], data_width, domain=domain)
 357
 358         wb = Wishbone(data_width=data_width, address_width=address_width, sel_width=sel_width, master=True)
 359
 360         self._wbs.append(JTAGWishbone(sr_addr, sr_data, wb, domain))
 361
 362         return wb