litex/misoclib/video/dvisampler/chansync.py

from migen.fhdl.std import *
from migen.genlib.cdc import MultiReg
from migen.genlib.fifo import _inc
from migen.genlib.record import Record, layout_len
from migen.genlib.misc import optree
from migen.bank.description import *

from misoclib.video.dvisampler.common import channel_layout

class _SyncBuffer(Module):
	def __init__(self, width, depth):
		self.din = Signal(width)
		self.dout = Signal(width)
		self.re = Signal()

		###

		produce = Signal(max=depth)
		consume = Signal(max=depth)
		storage = Memory(width, depth)
		self.specials += storage

		wrport = storage.get_port(write_capable=True)
		self.specials += wrport
		self.comb += [
			wrport.adr.eq(produce),
			wrport.dat_w.eq(self.din),
			wrport.we.eq(1)
		]
		self.sync += _inc(produce, depth)

		rdport = storage.get_port(async_read=True)
		self.specials += rdport
		self.comb += [
			rdport.adr.eq(consume),
			self.dout.eq(rdport.dat_r)
		]
		self.sync += If(self.re, _inc(consume, depth))

class ChanSync(Module, AutoCSR):
	def __init__(self, nchan=3, depth=8):
		self.valid_i = Signal()
		self.chan_synced = Signal()

		self._channels_synced = CSRStatus()

		lst_control = []
		all_control = Signal()
		for i in range(nchan):
			name = "data_in" + str(i)
			data_in = Record(channel_layout, name=name)
			setattr(self, name, data_in)
			name = "data_out" + str(i)
			data_out = Record(channel_layout, name=name)
			setattr(self, name, data_out)

			###

			syncbuffer = RenameClockDomains(_SyncBuffer(layout_len(channel_layout), depth), "pix")
			self.submodules += syncbuffer
			self.comb += [
				syncbuffer.din.eq(data_in.raw_bits()),
				data_out.raw_bits().eq(syncbuffer.dout)
			]
			is_control = Signal()
			self.comb += [
				is_control.eq(~data_out.de),
				syncbuffer.re.eq(~is_control | all_control)
			]
			lst_control.append(is_control)

		some_control = Signal()
		self.comb += [
			all_control.eq(optree("&", lst_control)),
			some_control.eq(optree("|", lst_control))
		]
		self.sync.pix += If(~self.valid_i,
				self.chan_synced.eq(0)
			).Else(
				If(some_control,
					If(all_control,
						self.chan_synced.eq(1)
					).Else(
						self.chan_synced.eq(0)
					)
				)
			)
		self.specials += MultiReg(self.chan_synced, self._channels_synced.status)

class _TB(Module):
	def __init__(self, test_seq_it):
		self.test_seq_it = test_seq_it

		self.submodules.chansync = RenameClockDomains(ChanSync(), {"pix": "sys"})
		self.comb += self.chansync.valid_i.eq(1)

	def do_simulation(self, selfp):
		try:
			de0, de1, de2 = next(self.test_seq_it)
		except StopIteration:
			raise StopSimulation

		selfp.chansync.data_in0.de = de0
		selfp.chansync.data_in1.de = de1
		selfp.chansync.data_in2.de = de2
		selfp.chansync.data_in0.d = selfp.simulator.cycle_counter
		selfp.chansync.data_in1.d = selfp.simulator.cycle_counter
		selfp.chansync.data_in2.d = selfp.simulator.cycle_counter

		out0 = selfp.chansync.data_out0.d
		out1 = selfp.chansync.data_out1.d
		out2 = selfp.chansync.data_out2.d

		print("{0:5} {1:5} {2:5}".format(out0, out1, out2))

if __name__ == "__main__":
	from migen.sim.generic import run_simulation

	test_seq = [
		(1, 1, 1),
		(1, 1, 0),
		(0, 0, 0),
		(0, 0, 0),
		(0, 0, 1),
		(1, 1, 1),
		(1, 1, 1),
	]
	tb = _TB(iter(test_seq*2))
	run_simulation(tb)