import operator import collections import inspect from migen.fhdl.structure import * from migen.fhdl.structure import (_Value, _Statement, _Operator, _Slice, _ArrayProxy, _Assign, _Fragment) from migen.fhdl.bitcontainer import value_bits_sign from migen.fhdl.tools import list_signals, list_targets, insert_resets from migen.fhdl.simplify import MemoryToArray from migen.fhdl.specials import _MemoryLocation from migen.sim.vcd import VCDWriter, DummyVCDWriter class ClockState: def __init__(self, high, half_period, time_before_trans): self.high = high self.half_period = half_period self.time_before_trans = time_before_trans class TimeManager: def __init__(self, description): self.clocks = dict() for k, period_phase in description.items(): if isinstance(period_phase, tuple): period, phase = period_phase else: period = period_phase phase = 0 half_period = period//2 if phase >= half_period: phase -= half_period high = True else: high = False self.clocks[k] = ClockState(high, half_period, half_period - phase) def tick(self): rising = set() falling = set() dt = min(cs.time_before_trans for cs in self.clocks.values()) for k, cs in self.clocks.items(): if cs.time_before_trans == dt: cs.high = not cs.high if cs.high: rising.add(k) else: falling.add(k) cs.time_before_trans -= dt if not cs.time_before_trans: cs.time_before_trans += cs.half_period return dt, rising, falling str2op = { "~": operator.invert, "+": operator.add, "-": operator.sub, "*": operator.mul, ">>>": operator.rshift, "<<<": operator.lshift, "&": operator.and_, "^": operator.xor, "|": operator.or_, "<": operator.lt, "<=": operator.le, "==": operator.eq, "!=": operator.ne, ">": operator.gt, ">=": operator.ge, } def _truncate(value, nbits, signed): value = value & (2**nbits - 1) if signed and (value & 2**(nbits - 1)): value -= 2**nbits return value class Evaluator: def __init__(self, clock_domains, replaced_memories): self.clock_domains = clock_domains self.replaced_memories = replaced_memories self.signal_values = dict() self.modifications = dict() def commit(self): r = set() for k, v in self.modifications.items(): if k not in self.signal_values or self.signal_values[k] != v: self.signal_values[k] = v r.add(k) self.modifications.clear() return r def eval(self, node, postcommit=False): if isinstance(node, Constant): return node.value elif isinstance(node, Signal): if postcommit: try: return self.modifications[node] except KeyError: pass try: return self.signal_values[node] except KeyError: return node.reset.value elif isinstance(node, _Operator): operands = [self.eval(o, postcommit) for o in node.operands] if node.op == "-": if len(operands) == 1: return -operands[0] else: return operands[0] - operands[1] elif node.op == "m": return operands[1] if operands[0] else operands[2] else: return str2op[node.op](*operands) elif isinstance(node, _Slice): v = self.eval(node.value, postcommit) idx = range(node.start, node.stop) return sum(((v >> i) & 1) << j for j, i in enumerate(idx)) elif isinstance(node, Cat): shift = 0 r = 0 for element in node.l: nbits = len(element) # make value always positive r |= (self.eval(element, postcommit) & (2**nbits-1)) << shift shift += nbits return r elif isinstance(node, _ArrayProxy): return self.eval(node.choices[self.eval(node.key, postcommit)], postcommit) elif isinstance(node, _MemoryLocation): array = self.replaced_memories[node.memory] return self.eval(array[self.eval(node.index, postcommit)], postcommit) elif isinstance(node, ClockSignal): return self.eval(self.clock_domains[node.cd].clk, postcommit) elif isinstance(node, ResetSignal): rst = self.clock_domains[node.cd].rst if rst is None: if node.allow_reset_less: return 0 else: raise ValueError("Attempted to get reset signal of resetless" " domain '{}'".format(node.cd)) else: return self.eval(rst, postcommit) else: raise NotImplementedError def assign(self, node, value): if isinstance(node, Signal): assert not node.variable self.modifications[node] = _truncate(value, node.nbits, node.signed) elif isinstance(node, Cat): for element in node.l: nbits = len(element) self.assign(element, value & (2**nbits-1)) value >>= nbits elif isinstance(node, _Slice): full_value = self.eval(node.value, True) # clear bits assigned to by the slice full_value &= ~((2**node.stop-1) - (2**node.start-1)) # set them to the new value value &= 2**(node.stop - node.start)-1 full_value |= value << node.start self.assign(node.value, full_value) elif isinstance(node, _ArrayProxy): self.assign(node.choices[self.eval(node.key)], value) elif isinstance(node, _MemoryLocation): array = self.replaced_memories[node.memory] self.assign(array[self.eval(node.index)], value) else: raise NotImplementedError def execute(self, statements): for s in statements: if isinstance(s, _Assign): self.assign(s.l, self.eval(s.r)) elif isinstance(s, If): if self.eval(s.cond) & (2**len(s.cond) - 1): self.execute(s.t) else: self.execute(s.f) elif isinstance(s, Case): nbits, signed = value_bits_sign(s.test) test = _truncate(self.eval(s.test), nbits, signed) found = False for k, v in s.cases.items(): if isinstance(k, Constant) and k.value == test: self.execute(v) found = True break if not found and "default" in s.cases: self.execute(s.cases["default"]) elif isinstance(s, collections.Iterable): self.execute(s) else: raise NotImplementedError # TODO: instances via Iverilog/VPI class Simulator: def __init__(self, fragment_or_module, generators, clocks={"sys": 10}, vcd_name=None): if isinstance(fragment_or_module, _Fragment): self.fragment = fragment_or_module else: self.fragment = fragment_or_module.get_fragment() if not isinstance(generators, dict): generators = {"sys": generators} self.generators = dict() for k, v in generators.items(): if (isinstance(v, collections.Iterable) and not inspect.isgenerator(v)): self.generators[k] = list(v) else: self.generators[k] = [v] self.time = TimeManager(clocks) for clock in clocks.keys(): if clock not in self.fragment.clock_domains: cd = ClockDomain(name=clock, reset_less=True) cd.clk.reset = C(self.time.clocks[clock].high) self.fragment.clock_domains.append(cd) mta = MemoryToArray() mta.transform_fragment(None, self.fragment) insert_resets(self.fragment) # comb signals return to their reset value if nothing assigns them self.fragment.comb[0:0] = [s.eq(s.reset) for s in list_targets(self.fragment.comb)] self.evaluator = Evaluator(self.fragment.clock_domains, mta.replacements) if vcd_name is None: self.vcd = DummyVCDWriter() else: signals = list_signals(self.fragment) for cd in self.fragment.clock_domains: signals.add(cd.clk) if cd.rst is not None: signals.add(cd.rst) for memory_array in mta.replacements.values(): signals |= set(memory_array) signals = sorted(signals, key=lambda x: x.duid) self.vcd = VCDWriter(vcd_name, signals) def __enter__(self): return self def __exit__(self, type, value, traceback): self.close() def close(self): self.vcd.close() def _commit_and_comb_propagate(self): # TODO: optimize all_modified = set() modified = self.evaluator.commit() all_modified |= modified while modified: self.evaluator.execute(self.fragment.comb) modified = self.evaluator.commit() all_modified |= modified for signal in all_modified: self.vcd.set(signal, self.evaluator.signal_values[signal]) def _evalexec_nested_lists(self, x): if isinstance(x, list): return [self._evalexec_nested_lists(e) for e in x] elif isinstance(x, _Value): return self.evaluator.eval(x) elif isinstance(x, _Statement): self.evaluator.execute([x]) return None else: raise ValueError def _process_generators(self, cd): exhausted = [] for generator in self.generators[cd]: reply = None while True: try: request = generator.send(reply) if request is None: break # next cycle else: reply = self._evalexec_nested_lists(request) except StopIteration: exhausted.append(generator) break for generator in exhausted: self.generators[cd].remove(generator) def _continue_simulation(self): # TODO: passive generators return any(self.generators.values()) def run(self): self.evaluator.execute(self.fragment.comb) self._commit_and_comb_propagate() while True: dt, rising, falling = self.time.tick() self.vcd.delay(dt) for cd in rising: self.evaluator.assign(self.fragment.clock_domains[cd].clk, 1) if cd in self.fragment.sync: self.evaluator.execute(self.fragment.sync[cd]) if cd in self.generators: self._process_generators(cd) for cd in falling: self.evaluator.assign(self.fragment.clock_domains[cd].clk, 0) self._commit_and_comb_propagate() if not self._continue_simulation(): break def run_simulation(*args, **kwargs): with Simulator(*args, **kwargs) as s: s.run()