move sources to src, add package info and release 0.0.1

This commit is contained in:
Stefan Holst committed 2020-12-04 17:58:33 +09:00
1 parent 765cb70ca3
commit 08a22cf251
16 files changed
+79 -8

No files matched your search

+108
View File
@@ -0,0 +1,108 @@
"""This package provides tools for high-performance processing and validation
of non-hierarchical VLSI circuits to aid rapid prototyping of research code
in the fields of VLSI test, diagnosis and reliability.
"""
import time
import importlib.util
class Log:
def __init__(self):
self.start = time.perf_counter()
self.logfile = None
def log(self, level, message):
t = time.perf_counter() - self.start
if self.logfile is None:
print(f'{t:011.3f} {level} {message}')
else:
self.logfile.write(f'{t:011.3f} {level} {message}\n')
self.logfile.flush()
def info(self, message): self.log('-', message)
def warn(self, message): self.log('W', message)
def error(self, message): self.log('E', message)
log = Log()
class MockNumba:
@staticmethod
def njit(func):
def inner(*args, **kwargs):
return func(*args, **kwargs)
return inner
class MockCuda:
def __init__(self):
self.x = 0
self.y = 0
def jit(self, device=False):
_ = device # silence "not used" warning
outer = self
def make_launcher(func):
class Launcher(object):
def __init__(self, funcc):
self.func = funcc
def __call__(self, *args, **kwargs):
# print(f'device func call {self.func.__name__}')
return self.func(*args, **kwargs)
def __getitem__(self, item):
grid_dim, block_dim = item
# print(f'kernel call {self.func.__name__} grid_dim:{grid_dim} block_dim:{block_dim}')
def inner(*args, **kwargs):
for grid_x in range(grid_dim[0]):
for grid_y in range(grid_dim[1]):
for block_x in range(block_dim[0]):
for block_y in range(block_dim[1]):
outer.x = grid_x * block_dim[0] + block_x
outer.y = grid_y * block_dim[1] + block_y
self.func(*args, **kwargs)
return inner
return Launcher(func)
return make_launcher
@staticmethod
def to_device(array, to=None):
if to is not None:
to[...] = array
return to
return array.copy()
def synchronize(self):
pass
def grid(self, dims):
_ = dims # silence "not used" warning
return self.x, self.y
if importlib.util.find_spec('numba') is not None:
import numba
import numba.cuda
from numba.cuda.cudadrv.error import CudaSupportError
try:
list(numba.cuda.gpus)
from numba import cuda
except CudaSupportError:
log.warn('Cuda unavailable. Falling back to pure python')
cuda = MockCuda()
else:
numba = MockNumba()
cuda = MockCuda()
log.warn('Numba unavailable. Falling back to pure python')
+43
View File
@@ -0,0 +1,43 @@
from lark import Lark, Transformer
from .circuit import Circuit, Node, Line
class BenchTransformer(Transformer):
def __init__(self, name):
super().__init__()
self.c = Circuit(name)
def start(self, _): return self.c
def parameters(self, args): return [self.c.get_or_add_fork(name) for name in args]
def interface(self, args): self.c.interface.extend(args[0])
def assignment(self, args):
name, cell_type, drivers = args
cell = Node(self.c, str(name), str(cell_type))
Line(self.c, cell, self.c.get_or_add_fork(str(name)))
[Line(self.c, d, cell) for d in drivers]
def parse(bench):
grammar = r"""
start: (statement)*
statement: input | output | assignment
input: ("INPUT" | "input") parameters -> interface
output: ("OUTPUT" | "output") parameters -> interface
assignment: NAME "=" NAME parameters
parameters: "(" [ NAME ( "," NAME )* ] ")"
NAME: /[-_a-z0-9]+/i
%ignore ( /\r?\n/ | "#" /[^\n]*/ | /[\t\f ]/ )+
"""
name = None
if '(' not in str(bench): # No parentheses?: Assuming it is a file name.
name = str(bench).replace('.bench', '')
with open(bench, 'r') as f:
text = f.read()
else:
text = bench
return Lark(grammar, parser="lalr", transformer=BenchTransformer(name)).parse(text)
+23
View File
@@ -0,0 +1,23 @@
import numpy as np
import importlib.util
if importlib.util.find_spec('numba') is not None:
import numba
else:
from . import numba
print('Numba unavailable. Falling back to pure python')
_pop_count_lut = np.asarray([bin(x).count('1') for x in range(256)])
def popcount(a):
return np.sum(_pop_count_lut[a])
_bit_in_lut = np.array([2 ** x for x in range(7, -1, -1)], dtype='uint8')
@numba.njit
def bit_in(a, pos):
return a[pos >> 3] & _bit_in_lut[pos & 7]
+236
View File
@@ -0,0 +1,236 @@
from collections import deque
class GrowingList(list):
def __setitem__(self, index, value):
if index >= len(self):
self.extend([None] * (index + 1 - len(self)))
super().__setitem__(index, value)
class IndexList(list):
def __delitem__(self, index):
if index == len(self) - 1:
super().__delitem__(index)
else:
replacement = self.pop()
replacement.index = index
super().__setitem__(index, replacement)
class Node:
"""A Node is a named entity in a circuit (e.g. a gate, a standard cell,
a named signal, or a fan-out point) that has connections to other nodes.
Each node contains:
* `self.index`: a circuit-unique integer index.
* `self.kind`: a type describing its function (e.g. 'AND', 'NOR').
The type '__fork__' is special. It signifies a named signal
or a fan-out in the circuit. Any other type is considered a physical cell.
* `self.name`: a name. Names must be unique among all forks and all cells
in the circuit. However, a fork (`self.kind=='__fork__'`) and a cell with
the same name may coexist.
* `self.ins`: a list of input connections (objects of class `Line`)
* `self.outs`: a list of output connections (objects of class `Line`).
"""
def __init__(self, circuit, name, kind='__fork__'):
if kind == '__fork__':
if name in circuit.forks:
raise ValueError(f'fork of name {name} already exists.')
circuit.forks[name] = self
else:
if name in circuit.cells:
raise ValueError(f'cell of name {name} already exists.')
circuit.cells[name] = self
self.index = len(circuit.nodes)
circuit.nodes.append(self)
self.circuit = circuit
self.name = name
self.kind = kind
self.ins = GrowingList()
self.outs = GrowingList()
def __repr__(self):
ins = ' '.join([f'<{line.index}' if line is not None else '<None' for line in self.ins])
outs = ' '.join([f'>{line.index}' if line is not None else '>None' for line in self.outs])
return f'{self.index}:{self.kind}"{self.name}" {ins} {outs}'
def remove(self):
if self.circuit is not None:
del self.circuit.nodes[self.index]
if self.kind == '__fork__':
del self.circuit.forks[self.name]
else:
del self.circuit.cells[self.name]
self.circuit = None
class Line:
"""A Line is a directional 1:1 connection between two Nodes. It always
connects an output of a node (called `driver`) to an input of a node
(called `reader`) and has a circuit-unique index (`self.index`).
Furthermore, `self.driver_pin` and `self.reader_pin` are the
integer indices of the connected pins of the nodes. They always correspond
to the positions of the line in the connection lists of the nodes:
* `self.driver.outs[self.driver_pin] == self`
* `self.reader.ins[self.reader_pin] == self`
A Line always connects a single driver to a single reader. If a signal fans out to
multiple readers, a '__fork__' Node needs to be added.
"""
def __init__(self, circuit, driver, reader):
self.index = len(circuit.lines)
circuit.lines.append(self)
if type(driver) is Node:
self.driver = driver
self.driver_pin = len(driver.outs)
for pin, line in enumerate(driver.outs):
if line is None:
self.driver_pin = pin
break
else:
self.driver, self.driver_pin = driver
if type(reader) is Node:
self.reader = reader
self.reader_pin = len(reader.ins)
for pin, line in enumerate(reader.ins):
if line is None:
self.reader_pin = pin
break
else:
self.reader, self.reader_pin = reader
self.driver.outs[self.driver_pin] = self
self.reader.ins[self.reader_pin] = self
def remove(self):
circuit = None
if self.driver is not None:
self.driver.outs[self.driver_pin] = None
circuit = self.driver.circuit
if self.reader is not None:
self.reader.ins[self.reader_pin] = None
circuit = self.reader.circuit
if circuit is not None:
del circuit.lines[self.index]
self.driver = None
self.reader = None
def __repr__(self):
return f'{self.index}'
def __lt__(self, other):
return self.index < other.index
class Circuit:
"""A Circuit is a container for interconnected nodes and lines.
All contained lines have unique indices, so have all contained nodes.
These indices can be used to store additional data about nodes or lines
by allocating an array `my_data` of length `len(self.nodes)` and then
accessing it by `my_data[n.index]`. The indices may change iff lines or
nodes are removed from the circuit.
Nodes come in two flavors (cells and forks, see `Node`). The names of
these nodes are kept unique within these two flavors.
"""
def __init__(self, name=None):
self.name = name
self.nodes = IndexList()
self.lines = IndexList()
self.interface = GrowingList()
self.cells = {}
self.forks = {}
def get_or_add_fork(self, name):
return self.forks[name] if name in self.forks else Node(self, name)
def copy(self):
c = Circuit(self.name)
for node in self.nodes:
Node(c, node.name, node.kind)
for line in self.lines:
d = c.forks[line.driver.name] if line.driver.kind == '__fork__' else c.cells[line.driver.name]
r = c.forks[line.reader.name] if line.reader.kind == '__fork__' else c.cells[line.reader.name]
Line(c, (d, line.driver_pin), (r, line.reader_pin))
for node in self.interface:
if node.kind == '__fork__':
n = c.forks[node.name]
else:
n = c.cells[node.name]
c.interface.append(n)
return c
def dump(self):
header = f'{self.name}({",".join([str(n.index) for n in self.interface])})\n'
return header + '\n'.join([str(n) for n in self.nodes])
def __repr__(self):
name = f" '{self.name}'" if self.name else ''
return f'<Circuit{name} with {len(self.nodes)} nodes, {len(self.lines)} lines, {len(self.interface)} ports>'
def topological_order(self):
visit_count = [0] * len(self.nodes)
queue = deque(n for n in self.nodes if len(n.ins) == 0 or 'DFF' in n.kind)
while len(queue) > 0:
n = queue.popleft()
for line in n.outs:
if line is None: continue
succ = line.reader
visit_count[succ.index] += 1
if visit_count[succ.index] == len(succ.ins) and 'DFF' not in succ.kind:
queue.append(succ)
yield n
def topological_line_order(self):
for n in self.topological_order():
for line in n.outs:
if line is not None:
yield line
def reversed_topological_order(self):
visit_count = [0] * len(self.nodes)
queue = deque(n for n in self.nodes if len(n.outs) == 0 or 'DFF' in n.kind)
while len(queue) > 0:
n = queue.popleft()
for line in n.ins:
pred = line.driver
visit_count[pred.index] += 1
if visit_count[pred.index] == len(pred.outs) and 'DFF' not in pred.kind:
queue.append(pred)
yield n
def fanin(self, origin_nodes):
marks = [False] * len(self.nodes)
for n in origin_nodes:
marks[n.index] = True
for n in self.reversed_topological_order():
if not marks[n.index]:
for line in n.outs:
if line is not None:
marks[n.index] |= marks[line.reader.index]
if marks[n.index]:
yield n
def fanout_free_regions(self):
for stem in self.reversed_topological_order():
if len(stem.outs) == 1 and 'DFF' not in stem.kind: continue
region = []
if 'DFF' in stem.kind:
n = stem.ins[0]
if len(n.driver.outs) == 1 and 'DFF' not in n.driver.kind:
queue = deque([n.driver])
else:
queue = deque()
else:
queue = deque(n.driver for n in stem.ins
if len(n.driver.outs) == 1 and 'DFF' not in n.driver.kind)
while len(queue) > 0:
n = queue.popleft()
preds = [pred.driver for pred in n.ins
if len(pred.driver.outs) == 1 and 'DFF' not in pred.driver.kind]
queue.extend(preds)
region.append(n)
yield stem, region
+421
View File
@@ -0,0 +1,421 @@
import numpy as np
from . import packed_vectors
class LogicSim:
"""A bit-parallel naive combinational logic simulator supporting 1, 4, or 8-valued logics.
"""
def __init__(self, circuit, nvectors=1, vdim=1):
self.circuit = circuit
self.nvectors = nvectors
nbytes = (nvectors - 1) // 8 + 1
self.interface = list(circuit.interface) + [n for n in circuit.nodes if 'dff' in n.kind.lower()]
self.state = np.zeros((len(circuit.lines), vdim, nbytes), dtype='uint8')
self.state_epoch = np.zeros(len(circuit.nodes), dtype='int8') - 1
self.tmp = np.zeros((5, vdim, nbytes), dtype='uint8')
self.zero = np.zeros((vdim, nbytes), dtype='uint8')
if vdim > 1:
self.zero[1] = 255
self.epoch = 0
self.fork_vd1 = self.fork_vdx
self.const0_vd1 = self.const0_vdx
self.input_vd1 = self.fork_vd1
self.output_vd1 = self.fork_vd1
self.inv_vd1 = self.not_vd1
self.ibuff_vd1 = self.not_vd1
self.nbuff_vd1 = self.fork_vd1
self.xor2_vd1 = self.xor_vd1
self.fork_vd2 = self.fork_vdx
self.const0_vd2 = self.const0_vdx
self.input_vd2 = self.fork_vd2
self.output_vd2 = self.fork_vd2
self.inv_vd2 = self.not_vd2
self.ibuff_vd2 = self.not_vd2
self.nbuff_vd2 = self.fork_vd2
self.xor2_vd2 = self.xor_vd2
self.fork_vd3 = self.fork_vdx
self.const0_vd3 = self.const0_vdx
self.input_vd3 = self.fork_vd3
self.output_vd3 = self.fork_vd3
self.inv_vd3 = self.not_vd3
self.ibuff_vd3 = self.not_vd3
self.nbuff_vd3 = self.fork_vd3
self.xor2_vd3 = self.xor_vd3
known_fct = [(f[:-4], getattr(self, f)) for f in dir(self) if f.endswith(f'_vd{vdim}')]
self.node_fct = []
for n in circuit.nodes:
t = n.kind.lower().replace('__fork__', 'fork')
t = t.replace('__const0__', 'const0')
t = t.replace('__const1__', 'const1')
t = t.replace('tieh', 'const1')
# t = t.replace('xor', 'or').replace('xnor', 'nor')
fcts = [f for n, f in known_fct if t.startswith(n)]
if len(fcts) < 1:
raise ValueError(f'Unknown node kind {n.kind}')
self.node_fct.append(fcts[0])
def assign(self, stimuli):
if isinstance(stimuli, packed_vectors.PackedVectors):
stimuli = stimuli.bits
for (stim, node) in zip(stimuli, self.interface):
if len(node.outs) == 0: continue
outputs = [self.state[line.index] if line else self.tmp[3] for line in node.outs]
self.node_fct[node.index]([stim], outputs)
for line in node.outs:
if line:
self.state_epoch[line.reader.index] = self.epoch
for n in self.circuit.nodes:
if (n.kind == '__const1__') or (n.kind == '__const0__'):
outputs = [self.state[line.index] if line else self.tmp[3] for line in n.outs]
self.node_fct[n.index]([], outputs)
# print('assign const')
for line in n.outs:
if line:
self.state_epoch[line.reader.index] = self.epoch
def capture(self, responses):
if isinstance(responses, packed_vectors.PackedVectors):
responses = responses.bits
for (resp, node) in zip(responses, self.interface):
if len(node.ins) == 0: continue
resp[...] = self.state[node.ins[0].index]
def propagate(self):
for node in self.circuit.topological_order():
if self.state_epoch[node.index] != self.epoch: continue
inputs = [self.state[line.index] if line else self.zero for line in node.ins]
outputs = [self.state[line.index] if line else self.tmp[3] for line in node.outs]
# print('sim', node)
self.node_fct[node.index](inputs, outputs)
for line in node.outs:
self.state_epoch[line.reader.index] = self.epoch
self.epoch = (self.epoch + 1) % 128
@staticmethod
def fork_vdx(inputs, outputs):
for o in outputs: o[...] = inputs[0]
def const0_vdx(self, _, outputs):
for o in outputs: o[...] = self.zero
# 2-valued simulation
@staticmethod
def not_vd1(inputs, outputs):
outputs[0][0] = ~inputs[0][0]
def const1_vd1(self, _, outputs):
for o in outputs: o[...] = self.zero
self.not_vd1(outputs, outputs)
@staticmethod
def and_vd1(inputs, outputs):
o = outputs[0]
o[0] = inputs[0][0]
for i in inputs[1:]: o[0] &= i[0]
@staticmethod
def or_vd1(inputs, outputs):
o = outputs[0]
o[0] = inputs[0][0]
for i in inputs[1:]: o[0] |= i[0]
@staticmethod
def xor_vd1(inputs, outputs):
o = outputs[0]
o[0] = inputs[0][0]
for i in inputs[1:]: o[0] ^= i[0]
@staticmethod
def sdff_vd1(inputs, outputs):
outputs[0][0] = inputs[0][0]
if len(outputs) > 1:
outputs[1][0] = ~inputs[0][0]
@staticmethod
def dff_vd1(inputs, outputs):
outputs[0][0] = inputs[0][0]
if len(outputs) > 1:
outputs[1][0] = ~inputs[0][0]
def nand_vd1(self, inputs, outputs):
self.and_vd1(inputs, outputs)
self.not_vd1(outputs, outputs)
def nor_vd1(self, inputs, outputs):
self.or_vd1(inputs, outputs)
self.not_vd1(outputs, outputs)
def xnor_vd1(self, inputs, outputs):
self.xor_vd1(inputs, outputs)
self.not_vd1(outputs, outputs)
# 4-valued simulation
# sym [0] [1] (value, care)
# 0 0 1
# 1 1 1
# - 0 0
# X 1 0
@staticmethod
def not_vd2(inputs, outputs):
# 4-valued not:
# i: 0 1 - X
# o: 1 0 X X
# o0 1 0 1 1
# o1 1 1 0 0
outputs[0][0] = ~inputs[0][0] | ~inputs[0][1] # value = 0 or DC
outputs[0][1] = inputs[0][1] # care = C
def and_vd2(self, inputs, outputs):
# 4-valued: o[0]: o[1]:
# 0 1 - X 0 1 - X 0 1 - X
# 0 0 0 0 0 0 0 0 0 1 1 1 1
# 1 0 1 X X 0 1 1 1 1 1 0 0
# - 0 X X X 0 1 1 1 1 0 0 0
# X 0 X X X 0 1 1 1 1 0 0 0
i = inputs[0]
any0 = self.tmp[0]
anyd = self.tmp[1]
any0[0] = ~i[0] & i[1]
anyd[0] = ~i[1]
for i in inputs[1:]:
any0[0] |= ~i[0] & i[1]
anyd[0] |= ~i[1]
o = outputs[0]
o[0] = ~any0[0] # value = no0
o[1] = any0[0] | ~anyd[0] # care = any0 or noDC
def or_vd2(self, inputs, outputs):
# 4-valued: o[0]: o[1]:
# 0 1 - X 0 1 - X 0 1 - X
# 0 0 1 X X 0 1 1 1 1 1 0 0
# 1 1 1 1 1 1 1 1 1 1 1 1 1
# - X 1 X X 1 1 1 1 0 1 0 0
# X X 1 X X 1 1 1 1 0 1 0 0
i = inputs[0]
any1 = self.tmp[0]
anyd = self.tmp[1]
any1[0] = i[0] & i[1]
anyd[0] = ~i[1]
for i in inputs[1:]:
any1[0] |= i[0] & i[1]
anyd[0] |= ~i[1]
o = outputs[0]
o[0] = any1[0] | anyd[0] # value = any1 or anyDC
o[1] = any1[0] | ~anyd[0] # care = any1 or noDC
def xor_vd2(self, inputs, outputs):
# 4-valued: o[0]: o[1]:
# 0 1 - X 0 1 - X 0 1 - X
# 0 0 1 X X 0 1 1 1 1 1 0 0
# 1 1 0 X X 1 0 1 1 1 1 0 0
# - X X X X 1 1 1 1 0 0 0 0
# X X X X X 1 1 1 1 0 0 0 0
i = inputs[0]
odd1 = self.tmp[0]
anyd = self.tmp[1]
odd1[0] = i[0] & i[1]
anyd[0] = ~i[1]
for i in inputs[1:]:
odd1[0] ^= i[0] & i[1]
anyd[0] |= ~i[1]
o = outputs[0]
o[0] = odd1[0] | anyd[0] # value = odd1 or anyDC
o[1] = ~anyd[0] # care = noDC
def sdff_vd2(self, inputs, outputs):
self.dff_vd2(inputs, outputs)
if len(outputs) > 1:
outputs[1][0] = ~inputs[0][0] | ~inputs[0][1] # value = 0 or DC
outputs[1][1] = inputs[0][1] # care = C
@staticmethod
def dff_vd2(inputs, outputs):
outputs[0][0] = inputs[0][0] | ~inputs[0][1] # value = 1 or DC
outputs[0][1] = inputs[0][1] # care = C
def nand_vd2(self, inputs, outputs):
self.and_vd2(inputs, outputs)
self.not_vd2(outputs, outputs)
def nor_vd2(self, inputs, outputs):
self.or_vd2(inputs, outputs)
self.not_vd2(outputs, outputs)
def xnor_vd2(self, inputs, outputs):
self.xor_vd2(inputs, outputs)
self.not_vd2(outputs, outputs)
def const1_vd2(self, _, outputs):
for o in outputs: o[...] = self.zero
self.not_vd2(outputs, outputs)
# 8-valued simulation
# sym [0] [1] [2] (initial value, ~final value, toggles present?)
# 0 0 1 0
# 1 1 0 0
# - 0 0 0
# X 1 1 0
# R 0 0 1 _/"
# F 1 1 1 "\_
# P 0 1 1 _/\_
# N 1 0 1 "\/"
def not_vd3(self, inputs, outputs):
# 8-valued not:
# i: 0 1 - X R F P N
# i0 0 1 0 1 0 1 0 1
# i1 1 0 0 1 0 1 1 0
# i2 0 0 0 0 1 1 1 1
# o: 1 0 X X F R N P
# o0 1 0 1 1 1 0 1 0
# o1 0 1 1 1 1 0 0 1
# o2 0 0 0 0 1 1 1 1
i = inputs[0]
dc = self.tmp[0]
dc[0] = ~(i[0] ^ i[1]) & ~i[2]
dc = self.tmp[0]
outputs[0][0] = ~i[0] | dc[0] # init.v = ~i0 or DC
outputs[0][1] = ~i[1] | dc[0] # init.v = ~i1 or DC
outputs[0][2] = i[2] # toggles = i2
def and_vd3(self, inputs, outputs):
# 8-valued: o[0]: o[1]: o[2]:
# 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N
# 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# 1 0 1 X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 0 0 0 0 1 1 1 1
# - 0 X X X X X X X 0 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# X 0 X X X X X X X 0 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# R 0 R X X R R P R 0 0 1 1 0 0 0 0 1 0 1 1 0 0 1 0 0 1 0 0 1 1 1 1
# F 0 F X X R F P F 0 1 1 1 0 1 0 1 1 1 1 1 0 1 1 1 0 1 0 0 1 1 1 1
# P 0 P X X P P P P 0 0 1 1 0 0 0 0 1 1 1 1 1 1 1 1 0 1 0 0 1 1 1 1
# N 0 N X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 0 1 0 0 1 1 1 1
i = inputs[0]
anyi0 = self.tmp[0]
anyf0 = self.tmp[1]
anyd = self.tmp[2]
any0 = self.tmp[3]
any_t = self.tmp[4]
anyd[0] = ~(i[0] ^ i[1]) & ~i[2]
anyi0[0] = ~i[0] & ~anyd[0]
anyf0[0] = i[1] & ~anyd[0]
any_t[0] = i[2]
any0[0] = anyi0[0] & anyf0[0] & ~i[2]
for i in inputs[1:]:
dc = ~(i[0] ^ i[1]) & ~i[2]
anyd[0] |= dc
anyi0[0] |= ~i[0] & ~dc
anyf0[0] |= i[1] & ~dc
any_t[0] |= i[2]
any0[0] |= ~i[0] & ~dc & i[1] & ~i[2]
o = outputs[0]
o[0] = (~anyi0[0] | anyd[0]) & ~any0[0] # initial = no_i0 or DC
o[1] = anyf0[0] | anyd[0] # ~final = ~no_f0 or DC
o[2] = any_t[0] & ~(anyd[0] | any0[0]) # toggle = anyT and noDC and no0
def or_vd3(self, inputs, outputs):
# 8-valued: o[0]: o[1]: o[2]:
# 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N
# 0 0 1 X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 0 0 0 0 1 1 1 1
# 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
# - X 1 X X X X X X 1 1 1 1 1 1 1 1 1 0 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# X X 1 X X X X X X 1 1 1 1 1 1 1 1 1 0 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# R R 1 X X R N R R 0 1 1 1 0 1 0 0 0 0 1 1 0 0 0 0 1 0 0 0 1 1 1 1
# F F 1 X X N F F F 1 1 1 1 1 1 1 1 1 0 1 1 0 1 1 1 1 0 0 0 1 1 1 1
# P P 1 X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 1 0 0 0 1 1 1 1
# N N 1 X X R F N N 1 1 1 1 0 1 1 1 0 0 1 1 0 1 0 0 1 0 0 0 1 1 1 1
i = inputs[0]
anyi1 = self.tmp[0]
anyf1 = self.tmp[1]
anyd = self.tmp[2]
any1 = self.tmp[3]
any_t = self.tmp[4]
anyd[0] = ~(i[0] ^ i[1]) & ~i[2]
anyi1[0] = i[0] & ~anyd[0]
anyf1[0] = ~i[1] & ~anyd[0]
any_t[0] = i[2]
any1[0] = (anyi1[0] & anyf1[0]) & ~i[2]
for i in inputs[1:]:
dc = ~(i[0] ^ i[1]) & ~i[2]
anyd[0] |= dc
anyi1[0] |= i[0] & ~dc
anyf1[0] |= ~i[1] & ~dc
any_t[0] |= i[2]
any1[0] |= i[0] & ~dc & ~i[1] & ~i[2]
o = outputs[0]
o[0] = anyi1[0] | anyd[0] # initial = i1 or DC
o[1] = (~anyf1[0] | anyd[0]) & ~any1[0] # ~final = f1 or DC
o[2] = any_t[0] & ~(anyd[0] | any1[0]) # toggle = anyT and no(DC or 1)
def xor_vd3(self, inputs, outputs):
# 8-valued: o[0]: o[1]: o[2]:
# 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N 0 1 - X R F P N
# 0 0 1 X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 0 0 0 0 1 1 1 1
# 1 1 0 X X F R N P 1 0 1 1 1 0 1 0 0 1 1 1 1 0 0 1 0 0 0 0 1 1 1 1
# - X X X X X X X X 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# X X X X X X X X X 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0
# R R F X X P N R F 0 1 1 1 0 1 0 1 0 1 1 1 1 0 0 1 1 1 0 0 1 1 1 1
# F F R X X N P F R 1 0 1 1 1 0 1 0 1 0 1 1 0 1 1 0 1 1 0 0 1 1 1 1
# P P N X X R F P N 0 1 1 1 0 1 0 1 1 0 1 1 0 1 1 0 1 1 0 0 1 1 1 1
# N N P X X F R N P 1 0 1 1 1 0 1 0 0 1 1 1 1 0 0 1 1 1 0 0 1 1 1 1
i = inputs[0]
odd0 = self.tmp[0]
odd1 = self.tmp[1]
anyd = self.tmp[2]
anyt = self.tmp[3]
odd0[0] = i[0]
odd1[0] = i[1]
anyd[0] = ~(i[0] ^ i[1]) & ~i[2]
anyt[0] = i[2]
for i in inputs[1:]:
odd0[0] ^= i[0]
odd1[0] ^= i[1]
anyd[0] |= ~(i[0] ^ i[1]) & ~i[2]
anyt[0] |= i[2]
o = outputs[0]
o[0] = odd0[0] | anyd[0]
o[1] = ~odd1[0] | anyd[0]
o[2] = anyt[0] & ~anyd[0]
def sdff_vd3(self, inputs, outputs):
self.dff_vd3(inputs, outputs)
if len(outputs) > 1:
i = inputs[0]
dc = self.tmp[0]
dc[0] = ~(i[0] ^ i[1]) & ~i[2]
outputs[1][0] = ~i[0] | dc[0] # value = 1 or DC
outputs[1][1] = ~i[1] | dc[0] # value = 1 or DC
outputs[1][2] = i[2] # toggle = T
def dff_vd3(self, inputs, outputs):
i = inputs[0]
dc = self.tmp[0]
dc[0] = ~(i[0] ^ i[1]) & ~i[2]
outputs[0][0] = i[0] | dc[0] # value = 1 or DC
outputs[0][1] = i[1] | dc[0] # value = 1 or DC
outputs[0][2] = i[2] # toggle = T
def nand_vd3(self, inputs, outputs):
self.and_vd3(inputs, outputs)
self.not_vd3(outputs, outputs)
def nor_vd3(self, inputs, outputs):
self.or_vd3(inputs, outputs)
self.not_vd3(outputs, outputs)
def xnor_vd3(self, inputs, outputs):
self.xor_vd3(inputs, outputs)
self.not_vd3(outputs, outputs)
def const1_vd3(self, _, outputs):
for o in outputs: o[...] = self.zero
self.not_vd3(outputs, outputs)
+299
View File
@@ -0,0 +1,299 @@
import numpy as np
from .bittools import popcount, bit_in
class PackedVectors:
def __init__(self, nvectors=8, width=1, vdim=1, from_cache=None):
if from_cache is not None:
self.bits = np.array(from_cache)
self.width, self.vdim, nbytes = self.bits.shape
else:
self.bits = np.zeros((width, vdim, (nvectors - 1) // 8 + 1), dtype='uint8')
self.vdim = vdim
self.width = width
self.nvectors = nvectors
m1 = np.array([2 ** x for x in range(7, -1, -1)], dtype='uint8')
m0 = ~m1
self.mask = np.rollaxis(np.vstack((m0, m1)), 1)
@classmethod
def from_pair(cls, init, final):
assert init.nvectors == final.nvectors
assert len(init.bits) == len(final.bits)
init_v = init.bits[:, 0]
if init.vdim == 3:
init_c = (init.bits[:, 0] ^ init.bits[:, 1]) | init.bits[:, 2]
elif init.vdim == 2:
init_c = init.bits[:, 1]
else:
init_c = ~np.zeros_like(init.bits[:, 0])
final_v = final.bits[:, 0]
if final.vdim == 3:
final_c = (final.bits[:, 0] ^ final.bits[:, 1]) | final.bits[:, 2]
final_v = ~final.bits[:, 1]
elif final.vdim == 2:
final_c = final.bits[:, 1]
else:
final_c = ~np.zeros_like(final.bits[:, 0])
c = init_c & final_c
a0 = init_v & c
a1 = ~final_v & c
a2 = (init_v ^ final_v) & c
p = PackedVectors(init.nvectors, len(init.bits), 3)
p.bits[:, 0] = a0
p.bits[:, 1] = a1
p.bits[:, 2] = a2
return p
def transition_vectors(self):
a = PackedVectors(self.nvectors-1, self.width, 3)
for pos in range(self.width):
for vidx in range(self.nvectors-1):
tr = self.get_value(vidx, pos) + self.get_value(vidx+1, pos)
if tr == '00':
a.set_value(vidx, pos, '0')
elif tr == '11':
a.set_value(vidx, pos, '1')
elif tr == '01':
a.set_value(vidx, pos, 'R')
elif tr == '10':
a.set_value(vidx, pos, 'F')
elif tr == '--':
a.set_value(vidx, pos, '-')
else:
a.set_value(vidx, pos, 'X')
return a
def __add__(self, other):
a = PackedVectors(self.nvectors + other.nvectors, self.width, max(self.vdim, other.vdim))
# a.bits[:self.bits.shape[0], 0] = self.bits[:, 0]
# if self.vdim == 2:
# a.bits[:self.bits.shape[0], 1] = self.care_bits
# elif self.vdim == 3:
# a.bits[:self.bits.shape[0], 1] = ~self.value_bits
# a.bits[:self.bits.shape[0], 2] = self.toggle_bits
for i in range(self.nvectors):
a[i] = self[i]
for i in range(len(other)):
a[self.nvectors+i] = other[i]
return a
def __len__(self):
return self.nvectors
def randomize(self, one_probability=0.5):
for data in self.bits:
data[0] = np.packbits((np.random.rand(self.nvectors) < one_probability).astype(int))
if self.vdim == 2:
data[1] = 255
elif self.vdim == 3:
data[1] = ~np.packbits((np.random.rand(self.nvectors) < one_probability).astype(int))
data[2] = data[0] ^ ~data[1]
def copy(self, selection_mask=None):
if selection_mask is not None:
cpy = PackedVectors(popcount(selection_mask), len(self.bits), self.vdim)
cur = 0
for vidx in range(self.nvectors):
if bit_in(selection_mask, vidx):
cpy[cur] = self[vidx]
cur += 1
else:
cpy = PackedVectors(self.nvectors, len(self.bits), self.vdim)
np.copyto(cpy.bits, self.bits)
return cpy
@property
def care_bits(self):
if self.vdim == 1:
return self.bits[:, 0] | 255
elif self.vdim == 2:
return self.bits[:, 1]
elif self.vdim == 3:
return (self.bits[:, 0] ^ self.bits[:, 1]) | self.bits[:, 2]
@property
def initial_bits(self):
return self.bits[:, 0]
@property
def value_bits(self):
if self.vdim == 3:
return ~self.bits[:, 1]
else:
return self.bits[:, 0]
@property
def toggle_bits(self):
if self.vdim == 3:
return self.bits[:, 2]
else:
return self.bits[:, 0] & 0
def get_value(self, vector, position):
if vector >= self.nvectors:
raise IndexError(f'vector out of range: {vector} >= {self.nvectors}')
a = self.bits[position, :, vector // 8]
m = self.mask[vector % 8]
if self.vdim == 1:
return '1' if a[0] & m[1] else '0'
elif self.vdim == 2:
if a[0] & m[1]:
return '1' if a[1] & m[1] else 'X'
else:
return '0' if a[1] & m[1] else '-'
elif self.vdim == 3:
if a[2] & m[1]:
if a[0] & m[1]:
return 'F' if a[1] & m[1] else 'N'
else:
return 'P' if a[1] & m[1] else 'R'
else:
if a[0] & m[1]:
return 'X' if a[1] & m[1] else '1'
else:
return '0' if a[1] & m[1] else '-'
def get_values_for_position(self, position):
return ''.join(self.get_value(x, position) for x in range(self.nvectors))
def set_value(self, vector, position, v):
if vector >= self.nvectors:
raise IndexError(f'vector out of range: {vector} >= {self.nvectors}')
a = self.bits[position, :, vector // 8]
m = self.mask[vector % 8]
if self.vdim == 1:
self._set_value_vd1(a, m, v)
elif self.vdim == 2:
self._set_value_vd2(a, m, v)
elif self.vdim == 3:
self._set_value_vd3(a, m, v)
def set_values(self, vector, v, mapping=None, inversions=None):
if vector >= self.nvectors:
raise IndexError(f'vector out of range: {vector} >= {self.nvectors}')
if not mapping:
mapping = [y for y in range(len(v))]
if inversions is None:
inversions = [False] * len(v)
for i, c in enumerate(v):
if inversions[i]:
if c == '1':
c = '0'
elif c == '0':
c = '1'
elif c == 'H':
c = 'L'
elif c == 'L':
c = 'H'
elif c == 'R':
c = 'F'
elif c == 'F':
c = 'R'
self.set_value(vector, mapping[i], c)
def set_values_for_position(self, position, values):
for i, v in enumerate(values):
self.set_value(i, position, v)
def __setitem__(self, vector, value):
for i, c in enumerate(value):
self.set_value(vector, i, c)
def __getitem__(self, vector):
if isinstance(vector, slice):
first = self.get_values_for_position(0)[vector]
ret = PackedVectors(len(first), self.width, self.vdim)
ret.set_values_for_position(0, first)
for pos in range(1, self.width):
ret.set_values_for_position(pos, self.get_values_for_position(pos)[vector])
return ret
return ''.join(self.get_value(vector, pos) for pos in range(len(self.bits)))
@staticmethod
def _set_value_vd1(a, m, v):
if v in [True, 1, '1', 'H', 'h']:
a[0] |= m[1]
else:
a[0] &= m[0]
@staticmethod
def _set_value_vd2(a, m, v):
if v in [True, 1, '1', 'H', 'h']:
a[0] |= m[1]
a[1] |= m[1]
elif v in [False, 0, '0', 'L', 'l']:
a[0] &= m[0]
a[1] |= m[1]
elif v in ['X', 'x']:
a[0] |= m[1]
a[1] &= m[0]
else:
a[0] &= m[0]
a[1] &= m[0]
# i fb act
# a 0 1 2
# - 0 0 0 None, '-'
# 0 0 1 0 False, 0, '0', 'l', 'L'
# 1 1 0 0 True, 1, '1', 'h', 'H'
# X 1 1 0 'x', 'X'
# / 0 0 1 '/', 'r', 'R'
# ^ 0 1 1 '^', 'p', 'P'
# v 1 0 1 'v', 'n', 'N'
# \ 1 1 1 '\', 'f', 'F'
@staticmethod
def _set_value_vd3(a, m, v):
if v in [False, 0, '0', 'L', 'l']:
a[0] &= m[0]
a[1] |= m[1]
a[2] &= m[0]
elif v in [True, 1, '1', 'H', 'h']:
a[0] |= m[1]
a[1] &= m[0]
a[2] &= m[0]
elif v in ['X', 'x']:
a[0] |= m[1]
a[1] |= m[1]
a[2] &= m[0]
elif v in ['/', 'r', 'R']:
a[0] &= m[0]
a[1] &= m[0]
a[2] |= m[1]
elif v in ['^', 'p', 'P']:
a[0] &= m[0]
a[1] |= m[1]
a[2] |= m[1]
elif v in ['v', 'n', 'N']:
a[0] |= m[1]
a[1] &= m[0]
a[2] |= m[1]
elif v in ['\\', 'f', 'F']:
a[0] |= m[1]
a[1] |= m[1]
a[2] |= m[1]
else:
a[0] &= m[0]
a[1] &= m[0]
a[2] &= m[0]
def __repr__(self):
return f'<PackedVectors nvectors={self.nvectors}, width={self.width}, vdim={self.vdim}>'
def __str__(self):
lst = []
for p in range(self.nvectors):
lst.append(''.join(self.get_value(p, w) for w in range(len(self.bits))))
if len(lst) == 0: return ''
if len(lst[0]) > 64:
lst = [s[:32] + '...' + s[-32:] for s in lst]
if len(lst) <= 16:
return '\n'.join(lst)
else:
return '\n'.join(lst[:8]) + '\n...\n' + '\n'.join(lst[-8:])
def diff(self, other, out=None):
if out is None:
out = np.zeros((self.width, self.bits.shape[-1]), dtype='uint8')
out[...] = (self.value_bits ^ other.value_bits) & self.care_bits & other.care_bits
return out
+289
View File
@@ -0,0 +1,289 @@
from kyupy.circuit import Node, Line
def pin_index(cell_type, pin):
if cell_type.startswith('HADD') and pin == 'B0': return 1
if cell_type.startswith('HADD') and pin == 'SO': return 1
if cell_type.startswith('MUX21') and pin == 'S': return 2
if cell_type.startswith('SDFF') and pin == 'QN': return 1
if cell_type.startswith('DFF') and pin == 'QN': return 1
if cell_type.startswith('DFF') and pin == 'CLK': return 1
if cell_type.startswith('DFF') and pin == 'RSTB': return 2
if cell_type.startswith('DFF') and pin == 'SETB': return 3
if pin in ['A2', 'IN2', 'SE', 'B', 'CO']: return 1
if pin in ['A3', 'IN3', 'SI', 'CI']: return 2
if pin == 'A4' or pin == 'IN4' or pin == 'CLK': return 3 # CLK for scan cells SDFF
if pin == 'A5' or pin == 'IN5' or pin == 'RSTB': return 4
if pin == 'A6' or pin == 'IN6' or pin == 'SETB': return 5
return 0
def pin_is_output(kind, pin):
if 'MUX' in kind and pin == 'S':
return False
return pin in ['Q', 'QN', 'Z', 'ZN', 'Y', 'CO', 'S', 'SO', 'C1']
def add_and_connect(circuit, name, kind, in1=None, in2=None, out=None):
n = Node(circuit, name, kind)
if in1 is not None:
n.ins[0] = in1
in1.reader = n
in1.reader_pin = 0
if in2 is not None:
n.ins[1] = in2
in2.reader = n
in2.reader_pin = 1
if out is not None:
n.outs[0] = out
out.driver = n
out.driver_pin = 0
return n
def split_complex_gates(circuit):
node_list = circuit.nodes
for n in node_list:
name = n.name
ins = n.ins
outs = n.outs
if n.kind.startswith('AO21X'):
n.remove()
n_and = add_and_connect(circuit, name+'~and', 'AND2', ins[0], ins[1], None)
n_or = add_and_connect(circuit, name+'~or', 'OR2', None, ins[2], outs[0])
Line(circuit, n_and, n_or)
elif n.kind.startswith('AOI21X'):
n.remove()
n_and = add_and_connect(circuit, name+'~and', 'AND2', ins[0], ins[1], None)
n_nor = add_and_connect(circuit, name+'~nor', 'NOR2', None, ins[2], outs[0])
Line(circuit, n_and, n_nor)
elif n.kind.startswith('OA21X'):
n.remove()
n_or = add_and_connect(circuit, name+'~or', 'OR2', ins[0], ins[1], None)
n_and = add_and_connect(circuit, name+'~and', 'AND2', None, ins[2], outs[0])
Line(circuit, n_or, n_and)
elif n.kind.startswith('OAI21X'):
n.remove()
n_or = add_and_connect(circuit, name+'~or', 'OR2', ins[0], ins[1], None)
n_nand = add_and_connect(circuit, name+'~nand', 'NAND2', None, ins[2], outs[0])
Line(circuit, n_or, n_nand)
elif n.kind.startswith('OA22X'):
n.remove()
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n_and = add_and_connect(circuit, name+'~and', 'AND2', None, None, outs[0])
Line(circuit, n_or0, n_and)
Line(circuit, n_or1, n_and)
elif n.kind.startswith('OAI22X'):
n.remove()
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n_nand = add_and_connect(circuit, name+'~nand', 'NAND2', None, None, outs[0])
Line(circuit, n_or0, n_nand)
Line(circuit, n_or1, n_nand)
elif n.kind.startswith('AO22X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_or = add_and_connect(circuit, name+'~or', 'OR2', None, None, outs[0])
Line(circuit, n_and0, n_or)
Line(circuit, n_and1, n_or)
elif n.kind.startswith('AOI22X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_nor = add_and_connect(circuit, name+'~nor', 'NOR2', None, None, outs[0])
Line(circuit, n_and0, n_nor)
Line(circuit, n_and1, n_nor)
elif n.kind.startswith('AO221X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', None, None, None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', None, ins[4], outs[0])
Line(circuit, n_and0, n_or0)
Line(circuit, n_and1, n_or0)
Line(circuit, n_or0, n_or1)
elif n.kind.startswith('AOI221X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_or = add_and_connect(circuit, name+'~or', 'OR2', None, None, None)
n_nor = add_and_connect(circuit, name+'~nor', 'NOR2', None, ins[4], outs[0])
Line(circuit, n_and0, n_or)
Line(circuit, n_and1, n_or)
Line(circuit, n_or, n_nor)
elif n.kind.startswith('OA221X'):
n.remove()
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', None, None, None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', None, ins[4], outs[0])
Line(circuit, n_or0, n_and0)
Line(circuit, n_or1, n_and0)
Line(circuit, n_and0, n_and1)
elif n.kind.startswith('OAI221X'):
n.remove()
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', None, None, None)
n_nand1 = add_and_connect(circuit, name+'~nand1', 'NAND2', None, ins[4], outs[0])
Line(circuit, n_or0, n_and0)
Line(circuit, n_or1, n_and0)
Line(circuit, n_and0, n_nand1)
elif n.kind.startswith('AO222X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_and2 = add_and_connect(circuit, name+'~and2', 'AND2', ins[4], ins[5], None)
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', None, None, None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', None, None, outs[0])
Line(circuit, n_and0, n_or0)
Line(circuit, n_and1, n_or0)
Line(circuit, n_and2, n_or1)
Line(circuit, n_or0, n_or1)
elif n.kind.startswith('AOI222X'):
n.remove()
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n_and2 = add_and_connect(circuit, name+'~and2', 'AND2', ins[4], ins[5], None)
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', None, None, None)
n_nor1 = add_and_connect(circuit, name+'~nor1', 'NOR2', None, None, outs[0])
Line(circuit, n_and0, n_or0)
Line(circuit, n_and1, n_or0)
Line(circuit, n_and2, n_nor1)
Line(circuit, n_or0, n_nor1)
elif n.kind.startswith('OA222X'):
n.remove()
n_or0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n_or1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n_or2 = add_and_connect(circuit, name+'~or2', 'OR2', ins[4], ins[5], None)
n_and0 = add_and_connect(circuit, name+'~and0', 'AND2', None, None, None)
n_and1 = add_and_connect(circuit, name+'~and1', 'AND2', None, None, outs[0])
Line(circuit, n_or0, n_and0)
Line(circuit, n_or1, n_and0)
Line(circuit, n_or2, n_and1)
Line(circuit, n_and0, n_and1)
elif n.kind.startswith('OAI222X'):
n.remove()
n0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n2 = add_and_connect(circuit, name+'~or2', 'OR2', ins[4], ins[5], None)
n3 = add_and_connect(circuit, name+'~and0', 'AND2', None, None, None)
n4 = add_and_connect(circuit, name+'~nand1', 'NAND2', None, None, outs[0])
Line(circuit, n0, n3)
Line(circuit, n1, n3)
Line(circuit, n2, n4)
Line(circuit, n3, n4)
elif n.kind.startswith('AND3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~and1', 'AND2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('OR3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~or1', 'OR2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('XOR3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~xor0', 'XOR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~xor1', 'XOR2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('NAND3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~and', 'AND2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~nand', 'NAND2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('NOR3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~or', 'OR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~nor', 'NOR2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('XNOR3X'):
n.remove()
n0 = add_and_connect(circuit, name+'~xor', 'XOR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~xnor', 'XNOR2', None, ins[2], outs[0])
Line(circuit, n0, n1)
elif n.kind.startswith('AND4X'):
n.remove()
n0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n2 = add_and_connect(circuit, name+'~and2', 'AND2', None, None, outs[0])
Line(circuit, n0, n2)
Line(circuit, n1, n2)
elif n.kind.startswith('OR4X'):
n.remove()
n0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n2 = add_and_connect(circuit, name+'~or2', 'OR2', None, None, outs[0])
Line(circuit, n0, n2)
Line(circuit, n1, n2)
elif n.kind.startswith('NAND4X'):
n.remove()
n0 = add_and_connect(circuit, name+'~and0', 'AND2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~and1', 'AND2', ins[2], ins[3], None)
n2 = add_and_connect(circuit, name+'~nand2', 'NAND2', None, None, outs[0])
Line(circuit, n0, n2)
Line(circuit, n1, n2)
elif n.kind.startswith('NOR4X'):
n.remove()
n0 = add_and_connect(circuit, name+'~or0', 'OR2', ins[0], ins[1], None)
n1 = add_and_connect(circuit, name+'~or1', 'OR2', ins[2], ins[3], None)
n2 = add_and_connect(circuit, name+'~nor2', 'NOR2', None, None, outs[0])
Line(circuit, n0, n2)
Line(circuit, n1, n2)
elif n.kind.startswith('FADDX'):
n.remove()
# forks for fan-outs
f_a = add_and_connect(circuit, name + '~fork0', '__fork__', ins[0])
f_b = add_and_connect(circuit, name + '~fork1', '__fork__', ins[1])
f_ci = add_and_connect(circuit, name + '~fork2', '__fork__', ins[2])
f_ab = Node(circuit, name + '~fork3')
# sum-block
n_xor0 = Node(circuit, name + '~xor0', 'XOR2')
Line(circuit, f_a, n_xor0)
Line(circuit, f_b, n_xor0)
Line(circuit, n_xor0, f_ab)
if len(outs) > 0 and outs[0] is not None:
n_xor1 = add_and_connect(circuit, name + '~xor1', 'XOR2', None, None, outs[0])
Line(circuit, f_ab, n_xor1)
Line(circuit, f_ci, n_xor1)
# carry-block
if len(outs) > 1 and outs[1] is not None:
n_and0 = Node(circuit, name + '~and0', 'AND2')
Line(circuit, f_ab, n_and0)
Line(circuit, f_ci, n_and0)
n_and1 = Node(circuit, name + '~and1', 'AND2')
Line(circuit, f_a, n_and1)
Line(circuit, f_b, n_and1)
n_or = add_and_connect(circuit, name + '~or0', 'OR2', None, None, outs[1])
Line(circuit, n_and0, n_or)
Line(circuit, n_and1, n_or)
elif n.kind.startswith('HADDX'):
n.remove()
# forks for fan-outs
f_a = add_and_connect(circuit, name + '~fork0', '__fork__', ins[0])
f_b = add_and_connect(circuit, name + '~fork1', '__fork__', ins[1])
n_xor0 = add_and_connect(circuit, name + '~xor0', 'XOR2', None, None, outs[1])
Line(circuit, f_a, n_xor0)
Line(circuit, f_b, n_xor0)
n_and0 = add_and_connect(circuit, name + '~and0', 'AND2', None, None, outs[0])
Line(circuit, f_a, n_and0)
Line(circuit, f_b, n_and0)
elif n.kind.startswith('MUX21X'):
n.remove()
f_s = add_and_connect(circuit, name + '~fork0', '__fork__', ins[2])
n_not = Node(circuit, name + '~not', 'INV')
Line(circuit, f_s, n_not)
n_and0 = add_and_connect(circuit, name + '~and0', 'AND2', ins[0])
n_and1 = add_and_connect(circuit, name + '~and1', 'AND2', ins[1])
n_or0 = add_and_connect(circuit, name + '~or0', 'OR2', None, None, outs[0])
Line(circuit, n_not, n_and0)
Line(circuit, f_s, n_and1)
Line(circuit, n_and0, n_or0)
Line(circuit, n_and1, n_or0)
elif n.kind.startswith('DFFSSR'):
n.kind = 'DFFX1'
n_and0 = add_and_connect(circuit, name + '~and0', 'AND2', ins[0], ins[2], None)
Line(circuit, n_and0, (n, 0))
+213
View File
@@ -0,0 +1,213 @@
import numpy as np
from lark import Lark, Transformer
from collections import namedtuple
from . import log
import gzip
Interconnect = namedtuple('Interconnect', ['orig', 'dest', 'r', 'f'])
IOPath = namedtuple('IOPath', ['ipin', 'opin', 'r', 'f'])
class DelayFile:
def __init__(self, name, cells):
self.name = name
if None in cells:
self.interconnects = cells[None]
else:
self.interconnects = None
self.cells = dict((n, l) for n, l in cells.items() if n)
def __repr__(self):
return '\n'.join(f'{n}: {l}' for n, l in self.cells.items()) + '\n' + \
'\n'.join(str(i) for i in self.interconnects)
def annotation(self, circuit, pin_index_f, dataset=1, interconnect=True, ffdelays=True):
"""
Constructs an 3-dimensional array with timing data for each line in `circuit`.
Dimension 1 of the returned array is the line index.
Dimension 2 is the type of timing data: 0:`delay`, 1:`pulse rejection limit`.
Dimension 3 is the polarity at the output of the reading node: 0:`rising`, 1:`falling`.
The polarity for pulse rejection is determined by the latter transition of the pulse.
E.g., timing[42,1,0] is the rejection limit of a negative pulse at the output of the reader of line 42.
An IOPATH delay for a node is annotated to the line connected to the input pin specified in the IOPATH.
Currently, only ABSOLUTE IOPATH and INTERCONNECT delays are supported.
Pulse rejection limits are derived from absolute delays, explicit declarations (PATHPULSE etc.) are ignored.
:param ffdelays:
:param interconnect:
:param pin_index_f:
:param circuit:
:type dataset: int or tuple
"""
def select_del(_delvals, idx):
if type(dataset) is tuple:
s = 0
for d in dataset:
s += _delvals[idx][d]
return s / len(dataset)
else:
return _delvals[idx][dataset]
def find_cell(name):
if name not in circuit.cells:
name = name.replace('\\', '')
if name not in circuit.cells:
name = name.replace('[', '_').replace(']', '_')
if name not in circuit.cells:
return None
return circuit.cells[name]
timing = np.zeros((len(circuit.lines), 2, 2))
for cn, iopaths in self.cells.items():
for ipn, opn, *delvals in iopaths:
delvals = [d if len(d) > 0 else [0, 0, 0] for d in delvals]
if max(max(delvals)) == 0:
continue
cell = find_cell(cn)
if cell is None:
log.warn(f'Cell from SDF not found in circuit: {cn}')
continue
ipin = pin_index_f(cell.kind, ipn)
opin = pin_index_f(cell.kind, opn)
kind = cell.kind.lower()
ipn2 = ipn.replace('(posedge A1)', 'A1').replace('(negedge A1)', 'A1')\
.replace('(posedge A2)', 'A2').replace('(negedge A2)', 'A2')
def add_delays(_line):
if _line is not None:
timing[_line.index, :, 0] += select_del(delvals, 0)
timing[_line.index, :, 1] += select_del(delvals, 1)
take_avg = False
if kind.startswith('sdff'):
if not ipn.startswith('(posedge CLK'):
continue
if ffdelays and (len(cell.outs) > opin):
add_delays(cell.outs[opin])
else:
if kind.startswith(('xor', 'xnor')):
ipin = pin_index_f(cell.kind, ipn2)
# print(ipn, ipin, times[cell.i_lines[ipin].index, 0, 0])
take_avg = timing[cell.ins[ipin].index].sum() > 0
add_delays(cell.ins[ipin])
if take_avg:
timing[cell.ins[ipin].index] /= 2
if not interconnect or self.interconnects is None:
return timing
for n1, n2, *delvals in self.interconnects:
delvals = [d if len(d) > 0 else [0, 0, 0] for d in delvals]
if max(max(delvals)) == 0:
continue
if '/' in n1:
i = n1.rfind('/')
cn1 = n1[0:i]
pn1 = n1[i+1:]
else:
cn1, pn1 = (n1, 'Z')
if '/' in n2:
i = n2.rfind('/')
cn2 = n2[0:i]
pn2 = n2[i+1:]
else:
cn2, pn2 = (n2, 'IN')
c1 = find_cell(cn1)
if c1 is None:
log.warn(f'Cell from SDF not found in circuit: {cn1}')
continue
c2 = find_cell(cn2)
if c2 is None:
log.warn(f'Cell from SDF not found in circuit: {cn2}')
continue
p1, p2 = pin_index_f(c1.kind, pn1), pin_index_f(c2.kind, pn2)
line = None
f1, f2 = c1.outs[p1].reader, c2.ins[p2].driver
if f1 != f2: # possible branchfork
assert len(f2.ins) == 1
line = f2.ins[0]
assert f1.outs[f2.ins[0].driver_pin] == line
elif len(f2.outs) == 1: # no fanout?
line = f2.ins[0]
if line is not None:
timing[line.index, :, 0] += select_del(delvals, 0)
timing[line.index, :, 1] += select_del(delvals, 1)
else:
log.warn(f'No branchfork for annotating interconnect delay {c1.name}/{p1}->{c2.name}/{p2}')
return timing
def sanitize(args):
if len(args) == 3: args.append(args[2])
return [str(args[0]), str(args[1])] + args[2:]
class SdfTransformer(Transformer):
@staticmethod
def triple(args): return [float(a.value[:-1]) if len(a.value) > 1 else 0.0 for a in args]
@staticmethod
def interconnect(args): return Interconnect(*sanitize(args))
@staticmethod
def iopath(args): return IOPath(*sanitize(args))
@staticmethod
def cell(args):
name = next((a for a in args if isinstance(a, str)), None)
entries = [e for a in args if hasattr(a, 'children') for e in a.children]
return name, entries
@staticmethod
def start(args):
name = next((a for a in args if isinstance(a, str)), None)
cells = dict(t for t in args if isinstance(t, tuple))
return DelayFile(name, cells)
def parse(sdf):
grammar = r"""
start: "(DELAYFILE" ( "(SDFVERSION" _NOB ")"
| "(DESIGN" "\"" NAME "\"" ")"
| "(DATE" _NOB ")"
| "(VENDOR" _NOB ")"
| "(PROGRAM" _NOB ")"
| "(VERSION" _NOB ")"
| "(DIVIDER" _NOB ")"
| "(VOLTAGE" _NOB ")"
| "(PROCESS" _NOB? ")"
| "(TEMPERATURE" _NOB ")"
| "(TIMESCALE" _NOB ")"
| cell )* ")"
cell: "(CELL" ( "(CELLTYPE" _NOB ")"
| "(INSTANCE" ID? ")"
| "(TIMINGCHECK" _ignore* ")"
| delay )* ")"
delay: "(DELAY" "(ABSOLUTE" (interconnect | iopath)* ")" ")"
interconnect: "(INTERCONNECT" ID ID triple* ")"
iopath: "(IOPATH" ID_OR_EDGE ID_OR_EDGE triple* ")"
NAME: /[^"]+/
ID_OR_EDGE: ( /[^() ]+/ | "(" /[^)]+/ ")" )
ID: ( /[^"() ]+/ | "\"" /[^"]+/ "\"" )
triple: "(" ( /[-.0-9]*:/ /[-.0-9]*:/ /[-.0-9]*\)/ | ")" )
_ignore: "(" _NOB? _ignore* ")" _NOB?
_NOB: /[^()]+/
COMMENT: "//" /[^\n]*/
%ignore ( /\r?\n/ | COMMENT )+
%ignore /[\t\f ]+/
"""
if '\n' not in str(sdf): # One line?: Assuming it is a file name.
if str(sdf).endswith('.gz'):
with gzip.open(sdf, 'rt') as f:
text = f.read()
else:
with open(sdf, 'r') as f:
text = f.read()
else:
text = str(sdf)
return Lark(grammar, parser="lalr", transformer=SdfTransformer()).parse(text)
+252
View File
@@ -0,0 +1,252 @@
from lark import Lark, Transformer
from collections import namedtuple
import re
import gzip
from .packed_vectors import PackedVectors
from .logic_sim import LogicSim
Call = namedtuple('Call', ['name', 'parameters'])
ScanPattern = namedtuple('ScanPattern', ['load', 'launch', 'capture', 'unload'])
class StilFile:
def __init__(self, version, signal_groups, scan_chains, calls):
self.version = version
self.signal_groups = signal_groups
self.scan_chains = scan_chains
self.si_ports = dict((v[0], k) for k, v in scan_chains.items())
self.so_ports = dict((v[-1], k) for k, v in scan_chains.items())
self.calls = calls
self.patterns = []
launch = {}
capture = {}
load = {}
for call in self.calls:
if call.name == 'load_unload':
unload = {}
for so_port in self.so_ports:
if so_port in call.parameters:
unload[so_port] = call.parameters[so_port].replace('\n', '')
if len(launch) > 0:
self.patterns.append(ScanPattern(load, launch, capture, unload))
capture = {}
launch = {}
load = {}
for si_port in self.si_ports:
if si_port in call.parameters:
load[si_port] = call.parameters[si_port].replace('\n', '')
if call.name.endswith('_launch') or call.name.endswith('_capture'):
if len(launch) == 0:
launch = dict((k, v.replace('\n', '')) for k, v in call.parameters.items())
else:
capture = dict((k, v.replace('\n', '')) for k, v in call.parameters.items())
def _maps(self, c):
interface = list(c.interface) + [n for n in c.nodes if 'DFF' in n.kind]
intf_pos = dict([(n.name, i) for i, n in enumerate(interface)])
pi_map = [intf_pos[n] for n in self.signal_groups['_pi']]
po_map = [intf_pos[n] for n in self.signal_groups['_po']]
scan_maps = {}
scan_inversions = {}
for chain_name, chain in self.scan_chains.items():
scan_map = []
scan_in_inversion = []
scan_out_inversion = []
inversion = False
for n in chain[1:-1]:
if n == '!':
inversion = not inversion
else:
scan_in_inversion.append(inversion)
scan_in_inversion = list(reversed(scan_in_inversion))
inversion = False
for n in reversed(chain[1:-1]):
if n == '!':
inversion = not inversion
else:
scan_map.append(intf_pos[n])
scan_out_inversion.append(inversion)
scan_maps[chain[0]] = scan_map
scan_maps[chain[-1]] = scan_map
scan_inversions[chain[0]] = scan_in_inversion
scan_inversions[chain[-1]] = scan_out_inversion
return interface, pi_map, po_map, scan_maps, scan_inversions
def tests(self, c):
interface, pi_map, po_map, scan_maps, scan_inversions = self._maps(c)
tests = PackedVectors(len(self.patterns), len(interface), 2)
for i, p in enumerate(self.patterns):
for si_port in self.si_ports.keys():
tests.set_values(i, p.load[si_port], scan_maps[si_port], scan_inversions[si_port])
tests.set_values(i, p.launch['_pi'], pi_map)
return tests
def tests8v(self, c):
interface, pi_map, po_map, scan_maps, scan_inversions = self._maps(c)
init = PackedVectors(len(self.patterns), len(interface), 2)
for i, p in enumerate(self.patterns):
# init.set_values(i, '0' * len(interface))
for si_port in self.si_ports.keys():
init.set_values(i, p.load[si_port], scan_maps[si_port], scan_inversions[si_port])
init.set_values(i, p.launch['_pi'], pi_map)
sim4v = LogicSim(c, len(init), 2)
sim4v.assign(init)
sim4v.propagate()
launch = init.copy()
sim4v.capture(launch)
for i, p in enumerate(self.patterns):
# if there was no launch clock, then init = launch
if ('P' not in p.launch['_pi']) or ('P' not in p.capture['_pi']):
for si_port in self.si_ports.keys():
launch.set_values(i, p.load[si_port], scan_maps[si_port], scan_inversions[si_port])
if '_pi' in p.capture and 'P' in p.capture['_pi']:
launch.set_values(i, p.capture['_pi'], pi_map)
return PackedVectors.from_pair(init, launch)
def responses(self, c):
interface, pi_map, po_map, scan_maps, scan_inversions = self._maps(c)
resp = PackedVectors(len(self.patterns), len(interface), 2)
for i, p in enumerate(self.patterns):
if len(p.capture) > 0:
resp.set_values(i, p.capture['_po'], po_map)
else:
resp.set_values(i, p.launch['_po'], po_map)
for so_port in self.so_ports.keys():
resp.set_values(i, p.unload[so_port], scan_maps[so_port], scan_inversions[so_port])
return resp
class StilTransformer(Transformer):
def __init__(self):
super().__init__()
self._signal_groups = None
self._calls = None
self._scan_chains = None
@staticmethod
def quoted(args): return args[0][1:-1]
@staticmethod
def call(args): return Call(args[0], dict(args[1:]))
@staticmethod
def call_parameter(args): return args[0], args[1].value
@staticmethod
def signal_group(args): return args[0], args[1:]
@staticmethod
def scan_chain(args):
scan_in = None
scan_cells = None
scan_out = None
for t in args[1:]:
if t.data == 'scan_in':
scan_in = t.children[0]
elif t.data == 'scan_out':
scan_out = t.children[0]
if t.data == 'scan_cells':
scan_cells = [n.replace('.SI', '') for n in t.children]
scan_cells = [re.sub(r'.*\.', '', s) if '.' in s else s for s in scan_cells]
return args[0], ([scan_in] + scan_cells + [scan_out])
def signal_groups(self, args): self._signal_groups = dict(args)
def pattern(self, args): self._calls = [c for c in args if isinstance(c, Call)]
def scan_structures(self, args): self._scan_chains = dict(args)
def start(self, args):
return StilFile(float(args[0]), self._signal_groups, self._scan_chains, self._calls)
def parse(stil):
grammar = r"""
start: "STIL" FLOAT _ignore _block*
_block: signal_groups | scan_structures | pattern
| "Header" _ignore
| "Signals" _ignore
| "Timing" _ignore
| "PatternBurst" quoted _ignore
| "PatternExec" _ignore
| "Procedures" _ignore
| "MacroDefs" _ignore
signal_groups: "SignalGroups" "{" signal_group* "}"
signal_group: quoted "=" "'" quoted ( "+" quoted)* "'" _ignore? ";"?
scan_structures: "ScanStructures" "{" scan_chain* "}"
scan_chain: "ScanChain" quoted "{" ( scan_length
| scan_in | scan_out | scan_inversion | scan_cells | scan_master_clock )* "}"
scan_length: "ScanLength" /[0-9]+/ ";"
scan_in: "ScanIn" quoted ";"
scan_out: "ScanOut" quoted ";"
scan_inversion: "ScanInversion" /[0-9]+/ ";"
scan_cells: "ScanCells" (quoted | /!/)* ";"
scan_master_clock: "ScanMasterClock" quoted ";"
pattern: "Pattern" quoted "{" ( label | w | c | macro | ann | call )* "}"
label: quoted ":"
w: "W" quoted ";"
c: "C" _ignore
macro: "Macro" quoted ";"
ann: "Ann" _ignore
call: "Call" quoted "{" call_parameter* "}"
call_parameter: quoted "=" /[^;]+/ ";"
quoted: /"[^"]*"/
FLOAT: /[-0-9.]+/
_ignore: "{" _NOB? _ignore_inner* "}"
_ignore_inner: "{" _NOB? _ignore_inner* "}" _NOB?
_NOB: /[^{}]+/
%ignore ( /\r?\n/ | "//" /[^\n]*/ | /[\t\f ]/ )+
"""
if '\n' not in str(stil): # One line?: Assuming it is a file name.
if str(stil).endswith('.gz'):
with gzip.open(stil, 'rt') as f:
text = f.read()
else:
with open(stil, 'r') as f:
text = f.read()
else:
text = str(stil)
return Lark(grammar, parser="lalr", transformer=StilTransformer()).parse(text)
def extract_scan_pattens(stil_calls):
pats = []
pi = None
scan_in = None
for call in stil_calls:
if call.name == 'load_unload':
scan_out = call.parameters.get('Scan_Out')
if scan_out is not None:
scan_out = scan_out.replace('\n', '')
if pi: pats.append(ScanPattern(scan_in, pi, None, scan_out))
scan_in = call.parameters.get('Scan_In')
if scan_in is not None:
scan_in = scan_in.replace('\n', '')
if call.name == 'allclock_capture':
pi = call.parameters['_pi'].replace('\n', '')
return pats
def match_patterns(stil_file, pats, interface):
intf_pos = dict([(n.name, i) for i, n in enumerate(interface)])
pi_map = [intf_pos[n] for n in stil_file.signal_groups['_pi']]
scan_map = [intf_pos[re.sub(r'b..\.', '', n)] for n in reversed(stil_file.scan_chains['1'])]
# print(scan_map)
tests = PackedVectors(len(pats), len(interface), 2)
for i, p in enumerate(pats):
tests.set_values(i, p.scan_in, scan_map)
tests.set_values(i, p.pi, pi_map)
resp = PackedVectors(len(pats), len(interface), 2)
for i, p in enumerate(pats):
resp.set_values(i, p.pi, pi_map)
resp.set_values(i, p.scan_out, scan_map)
return tests, resp
+173
View File
@@ -0,0 +1,173 @@
from collections import namedtuple
import gzip
from lark import Lark, Transformer
from .circuit import Circuit, Node, Line
from .saed import pin_index, pin_is_output
Instantiation = namedtuple('Instantiation', ['type', 'name', 'pins'])
class SignalDeclaration:
def __init__(self, kind, tokens):
self.left = None
self.right = None
self.kind = kind
if len(tokens.children) == 1:
self.basename = tokens.children[0]
else:
self.basename = tokens.children[2]
self.left = int(tokens.children[0].value)
self.right = int(tokens.children[1].value)
@property
def names(self):
if self.left is None:
return [self.basename]
if self.left <= self.right:
return [f'{self.basename}[{i}]' for i in range(self.left, self.right + 1)]
else:
return [f'{self.basename}[{i}]' for i in range(self.left, self.right - 1, -1)]
def __repr__(self):
return f"{self.kind}:{self.basename}[{self.left}:{self.right}]"
class VerilogTransformer(Transformer):
def __init__(self, branchforks=False):
super().__init__()
self._signal_declarations = {}
self.branchforks = branchforks
@staticmethod
def name(args):
s = args[0].value
if s[0] == '\\':
s = s[1:-1]
return s
@staticmethod
def instantiation(args):
return Instantiation(args[0], args[1],
dict([(pin.children[0], pin.children[1]) for pin in args[2:]]))
def input(self, args):
for sd in [SignalDeclaration('input', signal) for signal in args]:
self._signal_declarations[sd.basename] = sd
def inout(self, args):
for sd in [SignalDeclaration('input', signal) for signal in args]: # just treat as input
self._signal_declarations[sd.basename] = sd
def output(self, args):
for sd in [SignalDeclaration('output', signal) for signal in args]:
self._signal_declarations[sd.basename] = sd
def wire(self, args):
for sd in [SignalDeclaration('wire', signal) for signal in args]:
self._signal_declarations[sd.basename] = sd
def module(self, args):
c = Circuit(args[0])
positions = {}
pos = 0
for intf_sig in args[1].children:
for name in self._signal_declarations[intf_sig].names:
positions[name] = pos
pos += 1
assignments = []
for stmt in args[2:]: # pass 1: instantiate cells and driven signals
if type(stmt) is Instantiation:
n = Node(c, stmt.name, kind=stmt.type)
for p, s in stmt.pins.items():
if pin_is_output(n.kind, p):
Line(c, (n, pin_index(stmt.type, p)), Node(c, s))
elif stmt is not None and stmt.data == 'assign':
assignments.append((stmt.children[0], stmt.children[1]))
for sd in self._signal_declarations.values():
if sd.kind == 'output' or sd.kind == 'input':
for name in sd.names:
n = Node(c, name, kind=sd.kind)
if name in positions:
c.interface[positions[name]] = n
if sd.kind == 'input':
Line(c, n, Node(c, name))
for s1, s2 in assignments: # pass 1.5: process signal assignments
if s1 in c.forks:
assert s2 not in c.forks, 'assignment between two driven signals'
Line(c, c.forks[s1], Node(c, s2))
elif s2 in c.forks:
assert s1 not in c.forks, 'assignment between two driven signals'
Line(c, c.forks[s2], Node(c, s1))
for stmt in args[2:]: # pass 2: connect signals to readers
if type(stmt) is Instantiation:
for p, s in stmt.pins.items():
n = c.cells[stmt.name]
if pin_is_output(n.kind, p): continue
if s.startswith("1'b"):
const = f'__const{s[3]}__'
if const not in c.cells:
Line(c, Node(c, const, const), Node(c, s))
fork = c.forks[s]
if self.branchforks:
branchfork = Node(c, fork.name + "~" + n.name + "/" + p)
Line(c, fork, branchfork)
fork = branchfork
Line(c, fork, (n, pin_index(stmt.type, p)))
for sd in self._signal_declarations.values():
if sd.kind == 'output':
for name in sd.names:
Line(c, c.forks[name], c.cells[name])
return c
@staticmethod
def start(args):
if len(args) == 1:
return args[0]
else:
return args
grammar = """
start: (module)*
module: "module" name parameters ";" (_statement)* "endmodule"
parameters: "(" [ name ( "," name )* ] ")"
_statement: input | output | inout | tri | wire | assign | instantiation
input: "input" signal ( "," signal )* ";"
output: "output" signal ( "," signal )* ";"
inout: "inout" signal ( "," signal )* ";"
tri: "tri" name ";"
wire: "wire" signal ( "," signal )* ";"
assign: "assign" name "=" name ";"
instantiation: name name "(" [ pin ( "," pin )* ] ")" ";"
pin: "." name "(" name ")"
signal: ( name | "[" /[0-9]+/ ":" /[0-9]+/ "]" name )
name: ( /[a-z_][a-z0-9_\\[\\]]*/i | /\\\\[^\\t \\r\\n]+[\\t \\r\\n](\\[[0-9]+\\])?/i | /1'b0/i | /1'b1/i )
COMMENT: "//" /[^\\n]*/
%ignore ( /\\r?\\n/ | COMMENT )+
%ignore /[\\t \\f]+/
"""
def loads(s, *, branchforks=False):
return Lark(grammar, parser="lalr", transformer=VerilogTransformer(branchforks)).parse(s)
def load(fp, *, branchforks=False):
return loads(fp.read(), branchforks=branchforks)
def parse(verilog, branchforks=False):
if '\n' not in str(verilog): # One line?: Assuming it is a file name.
if str(verilog).endswith('.gz'):
with gzip.open(verilog, 'rt') as f:
text = f.read()
else:
with open(verilog, 'r') as f:
text = f.read()
else:
text = str(verilog)
return loads(text, branchforks=branchforks)
+521
View File
@@ -0,0 +1,521 @@
import math
from bisect import bisect, insort_left
import numpy as np
from . import numba
TMAX = np.float32(2 ** 127) # almost np.PINF for 32-bit floating point values
TMAX_OVL = np.float32(1.1 * 2 ** 127) # almost np.PINF with overflow mark
TMIN = np.float32(-2 ** 127) # almost np.NINF for 32-bit floating point values
class Heap:
def __init__(self):
self.chunks = dict() # map start location to chunk size
self.released = list() # chunks that were released
self.current_size = 0
self.max_size = 0
def alloc(self, size):
for idx, loc in enumerate(self.released):
if self.chunks[loc] == size:
del self.released[idx]
return loc
elif self.chunks[loc] > size: # split chunk
chunksize = self.chunks[loc]
self.chunks[loc] = size
self.chunks[loc + size] = chunksize - size
self.released[idx] = loc + size # move released pointer: loc -> loc+size
return loc
# no previously released chunk; make new one
loc = self.current_size
self.chunks[loc] = size
self.current_size += size
self.max_size = max(self.max_size, self.current_size)
return loc
def free(self, loc):
size = self.chunks[loc]
if loc + size == self.current_size: # end of managed area, remove chunk
del self.chunks[loc]
self.current_size -= size
# check and remove prev chunk if free
if len(self.released) > 0:
prev = self.released[-1]
if prev + self.chunks[prev] == self.current_size:
chunksize = self.chunks[prev]
del self.chunks[prev]
del self.released[-1]
self.current_size -= chunksize
return
released_idx = bisect(self.released, loc)
if released_idx < len(self.released) and loc + size == self.released[released_idx]: # next chunk is free, merge
chunksize = size + self.chunks[loc + size]
del self.chunks[loc + size]
self.chunks[loc] = chunksize
size = self.chunks[loc]
self.released[released_idx] = loc
else:
insort_left(self.released, loc) # put in a new release
if released_idx > 0: # check if previous chunk is free
prev = self.released[released_idx - 1]
if prev + self.chunks[prev] == loc: # previous chunk is adjacent to freed one, merge
chunksize = size + self.chunks[prev]
del self.chunks[loc]
self.chunks[prev] = chunksize
del self.released[released_idx]
def __repr__(self):
r = []
for loc in sorted(self.chunks.keys()):
size = self.chunks[loc]
released_idx = bisect(self.released, loc)
is_released = released_idx > 0 and len(self.released) > 0 and self.released[released_idx - 1] == loc
r.append(f'{loc:5d}: {"free" if is_released else "used"} {size}')
return "\n".join(r)
class WaveSim:
def __init__(self, circuit, timing, sims=8, wavecaps=16, strip_forks=False, keep_waveforms=True):
self.circuit = circuit
self.sims = sims
self.overflows = 0
self.interface = list(circuit.interface) + [n for n in circuit.nodes if 'dff' in n.kind.lower()]
self.lst_eat_valid = False
self.cdata = np.zeros((len(self.interface), sims, 7), dtype='float32')
if type(wavecaps) is int:
wavecaps = [wavecaps] * len(circuit.lines)
intf_wavecap = 4 # sufficient for storing only 1 transition.
# indices for state allocation table (sat)
self.zero_idx = len(circuit.lines)
self.tmp_idx = self.zero_idx + 1
self.ppi_offset = self.tmp_idx + 1
self.ppo_offset = self.ppi_offset + len(self.interface)
self.sat_length = self.ppo_offset + len(self.interface)
# translate circuit structure into self.ops
ops = []
interface_dict = dict([(n, i) for i, n in enumerate(self.interface)])
for n in circuit.topological_order():
if n in interface_dict:
inp_idx = self.ppi_offset + interface_dict[n]
if len(n.outs) > 0 and n.outs[0] is not None: # first output of a PI/PPI
ops.append((0b1010, n.outs[0].index, inp_idx, self.zero_idx))
if 'dff' in n.kind.lower(): # second output of DFF is inverted
if len(n.outs) > 1 and n.outs[1] is not None:
ops.append((0b0101, n.outs[1].index, inp_idx, self.zero_idx))
else: # if not DFF, no output is inverted.
for o_line in n.outs[1:]:
if o_line is not None:
ops.append((0b1010, o_line.index, inp_idx, self.zero_idx))
else: # regular node, not PI/PPI or PO/PPO
o0_idx = n.outs[0].index if len(n.outs) > 0 and n.outs[0] is not None else self.tmp_idx
i0_idx = n.ins[0].index if len(n.ins) > 0 and n.ins[0] is not None else self.zero_idx
i1_idx = n.ins[1].index if len(n.ins) > 1 and n.ins[1] is not None else self.zero_idx
kind = n.kind.lower()
if kind == '__fork__':
if not strip_forks:
for o_line in n.outs:
ops.append((0b1010, o_line.index, i0_idx, i1_idx))
elif kind.startswith('nand'):
ops.append((0b0111, o0_idx, i0_idx, i1_idx))
elif kind.startswith('nor'):
ops.append((0b0001, o0_idx, i0_idx, i1_idx))
elif kind.startswith('and'):
ops.append((0b1000, o0_idx, i0_idx, i1_idx))
elif kind.startswith('or'):
ops.append((0b1110, o0_idx, i0_idx, i1_idx))
elif kind.startswith('xor'):
ops.append((0b0110, o0_idx, i0_idx, i1_idx))
elif kind.startswith('xnor'):
ops.append((0b1001, o0_idx, i0_idx, i1_idx))
elif kind.startswith('not') or kind.startswith('inv'):
ops.append((0b0101, o0_idx, i0_idx, i1_idx))
elif kind.startswith('buf') or kind.startswith('nbuf'):
ops.append((0b1010, o0_idx, i0_idx, i1_idx))
elif kind.startswith('__const1__') or kind.startswith('tieh'):
ops.append((0b0101, o0_idx, i0_idx, i1_idx))
elif kind.startswith('__const0__') or kind.startswith('tiel'):
ops.append((0b1010, o0_idx, i0_idx, i1_idx))
else:
print('unknown gate type', kind)
self.ops = np.asarray(ops, dtype='int32')
# create a map from fanout lines to stem lines for fork stripping
stems = np.zeros(self.sat_length, dtype='int32') - 1 # default to -1: 'no fanout line'
if strip_forks:
for f in circuit.forks.values():
prev_line = f.ins[0]
while prev_line.driver.kind == '__fork__':
prev_line = prev_line.driver.ins[0]
stem_idx = prev_line.index
for ol in f.outs:
stems[ol.index] = stem_idx
# calculate level (distance from PI/PPI) and reference count for each line
levels = np.zeros(self.sat_length, dtype='int32')
ref_count = np.zeros(self.sat_length, dtype='int32')
level_starts = [0]
current_level = 1
for i, op in enumerate(self.ops):
# if we fork-strip, always take the stems for determining fan-in level
i0_idx = stems[op[2]] if stems[op[2]] >= 0 else op[2]
i1_idx = stems[op[3]] if stems[op[3]] >= 0 else op[3]
if levels[i0_idx] >= current_level or levels[i1_idx] >= current_level:
current_level += 1
level_starts.append(i)
levels[op[1]] = current_level # set level of the output line
ref_count[i0_idx] += 1
ref_count[i1_idx] += 1
self.level_starts = np.asarray(level_starts, dtype='int32')
self.level_stops = np.asarray(level_starts[1:] + [len(self.ops)], dtype='int32')
# state allocation table. maps line and interface indices to self.state memory locations
self.sat = np.zeros((self.sat_length, 3), dtype='int')
self.sat[:, 0] = -1
h = Heap()
# allocate and keep memory for special fields
self.sat[self.zero_idx] = h.alloc(intf_wavecap), intf_wavecap, 0
self.sat[self.tmp_idx] = h.alloc(intf_wavecap), intf_wavecap, 0
ref_count[self.zero_idx] += 1
ref_count[self.tmp_idx] += 1
# allocate and keep memory for PI/PPI, keep memory for PO/PPO (allocated later)
for i, n in enumerate(self.interface):
if len(n.outs) > 0:
self.sat[self.ppi_offset + i] = h.alloc(intf_wavecap), intf_wavecap, 0
ref_count[self.ppi_offset + i] += 1
if len(n.ins) > 0:
i0_idx = stems[n.ins[0].index] if stems[n.ins[0].index] >= 0 else n.ins[0].index
ref_count[i0_idx] += 1
# allocate memory for the rest of the circuit
for op_start, op_stop in zip(self.level_starts, self.level_stops):
free_list = []
for op in self.ops[op_start:op_stop]:
# if we fork-strip, always take the stems
i0_idx = stems[op[2]] if stems[op[2]] >= 0 else op[2]
i1_idx = stems[op[3]] if stems[op[3]] >= 0 else op[3]
ref_count[i0_idx] -= 1
ref_count[i1_idx] -= 1
if ref_count[i0_idx] <= 0: free_list.append(self.sat[i0_idx, 0])
if ref_count[i1_idx] <= 0: free_list.append(self.sat[i1_idx, 0])
o_idx = op[1]
cap = wavecaps[o_idx]
self.sat[o_idx] = h.alloc(cap), cap, 0
if not keep_waveforms:
for loc in free_list:
h.free(loc)
# copy memory location and capacity from stems to fanout lines
for lidx, stem in enumerate(stems):
if stem >= 0: # if at a fanout line
self.sat[lidx] = self.sat[stem]
# copy memory location to PO/PPO area
for i, n in enumerate(self.interface):
if len(n.ins) > 0:
self.sat[self.ppo_offset + i] = self.sat[n.ins[0].index]
# pad timing
self.timing = np.zeros((self.sat_length, 2, 2))
self.timing[:len(timing)] = timing
# allocate self.state
self.state = np.zeros((h.max_size, sims), dtype='float32') + TMAX
m1 = np.array([2 ** x for x in range(7, -1, -1)], dtype='uint8')
m0 = ~m1
self.mask = np.rollaxis(np.vstack((m0, m1)), 1)
def get_line_delay(self, line, polarity):
return self.timing[line, 0, polarity]
def set_line_delay(self, line, polarity, delay):
self.timing[line, 0, polarity] = delay
def assign(self, vectors, time=0.0, offset=0):
nvectors = min(vectors.nvectors - offset, self.sims)
for i, node in enumerate(self.interface):
ppi_loc = self.sat[self.ppi_offset + i, 0]
if ppi_loc < 0: continue
for p in range(nvectors):
vector = p + offset
a = vectors.bits[i, :, vector // 8]
m = self.mask[vector % 8]
toggle = 0
if a[0] & m[1]:
self.state[ppi_loc, p] = TMIN
toggle += 1
if (len(a) > 2) and (a[2] & m[1]) and ((a[0] & m[1]) == (a[1] & m[1])):
self.state[ppi_loc + toggle, p] = time
toggle += 1
self.state[ppi_loc + toggle, p] = TMAX
def propagate(self, sims=None, sd=0.0, seed=1):
if sims is None:
sims = self.sims
else:
sims = min(sims, self.sims)
for op_start, op_stop in zip(self.level_starts, self.level_stops):
self.overflows += level_eval(self.ops, op_start, op_stop, self.state, self.sat, 0, sims,
self.timing, sd, seed)
self.lst_eat_valid = False
def wave(self, line, vector):
if line < 0:
return [TMAX]
mem, wcap, _ = self.sat[line]
if mem < 0:
return [TMAX]
return self.state[mem:mem + wcap, vector]
def wave_ppi(self, i, vector):
return self.wave(self.ppi_offset + i, vector)
def wave_ppo(self, o, vector):
return self.wave(self.ppo_offset + o, vector)
def capture(self, time=TMAX, sd=0, seed=1, cdata=None, offset=0):
for i, node in enumerate(self.interface):
if len(node.ins) == 0: continue
for p in range(self.sims):
self.cdata[i, p] = self.capture_wave(self.ppo_offset + i, p, time, sd, seed)
if cdata is not None:
assert offset < cdata.shape[1]
cap_dim = min(cdata.shape[1] - offset, self.sims)
cdata[:, offset:cap_dim + offset] = self.cdata[:, 0:cap_dim]
self.lst_eat_valid = True
return self.cdata
def reassign(self, time=0.0):
for i, node in enumerate(self.interface):
ppi_loc = self.sat[self.ppi_offset + i, 0]
ppo_loc = self.sat[self.ppo_offset + i, 0]
if ppi_loc < 0 or ppo_loc < 0: continue
for sidx in range(self.sims):
ival = self.val(self.ppi_offset + i, sidx, TMAX) > 0.5
oval = self.cdata[i, sidx, 1] > 0.5
toggle = 0
if ival:
self.state[ppi_loc, sidx] = TMIN
toggle += 1
if ival != oval:
self.state[ppi_loc + toggle, sidx] = time
toggle += 1
self.state[ppi_loc + toggle, sidx] = TMAX
def eat(self, line, vector):
eat = TMAX
for t in self.wave(line, vector):
if t >= TMAX: break
if t <= TMIN: continue
eat = min(eat, t)
return eat
def lst(self, line, vector):
lst = TMIN
for t in self.wave(line, vector):
if t >= TMAX: break
if t <= TMIN: continue
lst = max(lst, t)
return lst
def lst_ppo(self, o, vector):
if not self.lst_eat_valid:
self.capture()
return self.cdata[o, vector, 5]
def toggles(self, line, vector):
tog = 0
for t in self.wave(line, vector):
if t >= TMAX: break
if t <= TMIN: continue
tog += 1
return tog
def _vals(self, idx, vector, times, sd=0.0):
s_sqrt2 = sd * math.sqrt(2)
m = 0.5
accs = [0.0] * len(times)
values = [0] * len(times)
for t in self.wave(idx, vector):
if t >= TMAX: break
for idx, time in enumerate(times):
if t < time:
values[idx] = values[idx] ^ 1
m = -m
if t <= TMIN: continue
if s_sqrt2 > 0:
for idx, time in enumerate(times):
accs[idx] += m * (1 + math.erf((t - time) / s_sqrt2))
if (m < 0) and (s_sqrt2 > 0):
for idx, time in enumerate(times):
accs[idx] += 1
if s_sqrt2 == 0:
return values
else:
return accs
def vals(self, line, vector, times, sd=0):
return self._vals(line, vector, times, sd)
def val(self, line, vector, time=TMAX, sd=0):
return self.capture_wave(line, vector, time, sd)[0]
def vals_ppo(self, o, vector, times, sd=0):
return self._vals(self.ppo_offset + o, vector, times, sd)
def val_ppo(self, o, vector, time=TMAX, sd=0):
if not self.lst_eat_valid:
self.capture(time, sd)
return self.cdata[o, vector, 0]
def capture_wave(self, line, vector, time=TMAX, sd=0.0, seed=1):
s_sqrt2 = sd * math.sqrt(2)
m = 0.5
acc = 0.0
eat = TMAX
lst = TMIN
tog = 0
ovl = 0
val = int(0)
final = int(0)
for t in self.wave(line, vector):
if t >= TMAX:
if t == TMAX_OVL:
ovl = 1
break
m = -m
final ^= 1
if t < time:
val ^= 1
if t <= TMIN: continue
if s_sqrt2 > 0:
acc += m * (1 + math.erf((t - time) / s_sqrt2))
eat = min(eat, t)
lst = max(lst, t)
tog += 1
if s_sqrt2 > 0:
if m < 0:
acc += 1
if acc >= 0.99:
val = 1
elif acc > 0.01:
seed = (seed << 4) + (vector << 20) + (line-self.ppo_offset << 1)
seed = int(0xDEECE66D) * seed + 0xB
seed = int(0xDEECE66D) * seed + 0xB
rnd = float((seed >> 8) & 0xffffff) / float(1 << 24)
val = rnd < acc
else:
val = 0
else:
acc = val
return acc, val, final, (val != final), eat, lst, ovl
@numba.njit
def level_eval(ops, op_start, op_stop, state, sat, st_start, st_stop, line_times, sd, seed):
overflows = 0
for op_idx in range(op_start, op_stop):
op = ops[op_idx]
for st_idx in range(st_start, st_stop):
overflows += wave_eval(op, state, sat, st_idx, line_times, sd, seed)
return overflows
@numba.njit
def rand_gauss(seed, sd):
clamp = 0.5
if sd <= 0.0:
return 1.0
while True:
x = -6.0
for i in range(12):
seed = int(0xDEECE66D) * seed + 0xB
x += float((seed >> 8) & 0xffffff) / float(1 << 24)
x *= sd
if abs(x) <= clamp:
break
return x + 1.0
@numba.njit
def wave_eval(op, state, sat, st_idx, line_times, sd=0.0, seed=0):
lut, z_idx, a_idx, b_idx = op
overflows = int(0)
_seed = (seed << 4) + (z_idx << 20) + (st_idx << 1)
a_mem = sat[a_idx, 0]
b_mem = sat[b_idx, 0]
z_mem, z_cap, _ = sat[z_idx]
a_cur = int(0)
b_cur = int(0)
z_cur = lut & 1
if z_cur == 1:
state[z_mem, st_idx] = TMIN
a = state[a_mem, st_idx] + line_times[a_idx, 0, z_cur] * rand_gauss(_seed ^ a_mem ^ z_cur, sd)
b = state[b_mem, st_idx] + line_times[b_idx, 0, z_cur] * rand_gauss(_seed ^ b_mem ^ z_cur, sd)
previous_t = TMIN
current_t = min(a, b)
inputs = int(0)
while current_t < TMAX:
z_val = z_cur & 1
if b < a:
b_cur += 1
b = state[b_mem + b_cur, st_idx]
b += line_times[b_idx, 0, z_val ^ 1] * rand_gauss(_seed ^ b_mem ^ z_val ^ 1, sd)
thresh = line_times[b_idx, 1, z_val] * rand_gauss(_seed ^ b_mem ^ z_val, sd)
inputs ^= 2
next_t = b
else:
a_cur += 1
a = state[a_mem + a_cur, st_idx]
a += line_times[a_idx, 0, z_val ^ 1] * rand_gauss(_seed ^ a_mem ^ z_val ^ 1, sd)
thresh = line_times[a_idx, 1, z_val] * rand_gauss(_seed ^ a_mem ^ z_val, sd)
inputs ^= 1
next_t = a
if (z_cur & 1) != ((lut >> inputs) & 1):
# we generate a toggle in z_mem, if:
# ( it is the first toggle in z_mem OR
# following toggle is earlier OR
# pulse is wide enough ) AND enough space in z_mem.
if z_cur == 0 or next_t < current_t or (current_t - previous_t) > thresh:
if z_cur < (z_cap - 1):
state[z_mem + z_cur, st_idx] = current_t
previous_t = current_t
z_cur += 1
else:
overflows += 1
previous_t = state[z_mem + z_cur - 1, st_idx]
z_cur -= 1
else:
z_cur -= 1
if z_cur > 0:
previous_t = state[z_mem + z_cur - 1, st_idx]
else:
previous_t = TMIN
current_t = min(a, b)
if overflows > 0:
state[z_mem + z_cur, st_idx] = TMAX_OVL
else:
state[z_mem + z_cur, st_idx] = a if a > b else b # propagate overflow flags by storing biggest TMAX from input
return overflows
+317
View File
@@ -0,0 +1,317 @@
import numpy as np
import math
from .wave_sim import WaveSim
from . import cuda
TMAX = np.float32(2 ** 127) # almost np.PINF for 32-bit floating point values
TMAX_OVL = np.float32(1.1 * 2 ** 127) # almost np.PINF with overflow mark
TMIN = np.float32(-2 ** 127) # almost np.NINF for 32-bit floating point values
class WaveSimCuda(WaveSim):
def __init__(self, circuit, timing, sims=8, wavecaps=16, strip_forks=False, keep_waveforms=True):
super().__init__(circuit, timing, sims, wavecaps, strip_forks, keep_waveforms)
self.tdata = np.zeros((len(self.interface), 3, (sims - 1) // 8 + 1), dtype='uint8')
self.d_state = cuda.to_device(self.state)
self.d_sat = cuda.to_device(self.sat)
self.d_ops = cuda.to_device(self.ops)
self.d_timing = cuda.to_device(self.timing)
self.d_tdata = cuda.to_device(self.tdata)
self.d_cdata = cuda.to_device(self.cdata)
self._block_dim = (32, 16)
def get_line_delay(self, line, polarity):
return self.d_timing[line, 0, polarity]
def set_line_delay(self, line, polarity, delay):
self.d_timing[line, 0, polarity] = delay
def assign(self, vectors, time=0.0, offset=0):
assert (offset % 8) == 0
byte_offset = offset // 8
assert byte_offset < vectors.bits.shape[-1]
pdim = min(vectors.bits.shape[-1] - byte_offset, self.tdata.shape[-1])
self.tdata[..., 0:pdim] = vectors.bits[..., byte_offset:pdim + byte_offset]
if vectors.vdim == 1:
self.tdata[:, 1, 0:pdim] = ~self.tdata[:, 1, 0:pdim]
self.tdata[:, 2, 0:pdim] = 0
cuda.to_device(self.tdata, to=self.d_tdata)
grid_dim = self._grid_dim(self.sims, len(self.interface))
assign_kernel[grid_dim, self._block_dim](self.d_state, self.d_sat, self.ppi_offset,
len(self.interface), self.d_tdata, time)
def _grid_dim(self, x, y):
gx = math.ceil(x / self._block_dim[0])
gy = math.ceil(y / self._block_dim[1])
return gx, gy
def propagate(self, sims=None, sd=0.0, seed=1):
if sims is None:
sims = self.sims
else:
sims = min(sims, self.sims)
for op_start, op_stop in zip(self.level_starts, self.level_stops):
grid_dim = self._grid_dim(sims, op_stop - op_start)
wave_kernel[grid_dim, self._block_dim](self.d_ops, op_start, op_stop, self.d_state, self.sat, int(0),
sims, self.d_timing, sd, seed)
cuda.synchronize()
self.lst_eat_valid = False
def wave(self, line, vector):
if line < 0:
return None
mem, wcap, _ = self.sat[line]
if mem < 0:
return None
return self.d_state[mem:mem + wcap, vector]
def capture(self, time=TMAX, sd=0, seed=1, cdata=None, offset=0):
grid_dim = self._grid_dim(self.sims, len(self.interface))
capture_kernel[grid_dim, self._block_dim](self.d_state, self.d_sat, self.ppo_offset,
self.d_cdata, time, sd * math.sqrt(2), seed)
self.cdata[...] = self.d_cdata
if cdata is not None:
assert offset < cdata.shape[1]
cap_dim = min(cdata.shape[1] - offset, self.sims)
cdata[:, offset:cap_dim + offset] = self.cdata[:, 0:cap_dim]
self.lst_eat_valid = True
return self.cdata
def reassign(self, time=0.0):
grid_dim = self._grid_dim(self.sims, len(self.interface))
reassign_kernel[grid_dim, self._block_dim](self.d_state, self.d_sat, self.ppi_offset, self.ppo_offset,
self.d_cdata, time)
cuda.synchronize()
def wavecaps(self):
gx = math.ceil(len(self.circuit.lines) / 512)
wavecaps_kernel[gx, 512](self.d_state, self.d_sat, self.sims)
self.sat[...] = self.d_sat
return self.sat[..., 2]
@cuda.jit()
def wavecaps_kernel(state, sat, sims):
idx = cuda.grid(1)
if idx >= len(sat): return
lidx, lcap, _ = sat[idx]
if lidx < 0: return
wcap = 0
for sidx in range(sims):
for tidx in range(lcap):
t = state[lidx + tidx, sidx]
if tidx > wcap:
wcap = tidx
if t >= TMAX: break
sat[idx, 2] = wcap + 1
@cuda.jit()
def reassign_kernel(state, sat, ppi_offset, ppo_offset, cdata, ppi_time):
vector, y = cuda.grid(2)
if vector >= state.shape[-1]: return
if ppo_offset + y >= len(sat): return
ppo, ppo_cap, _ = sat[ppo_offset + y]
ppi, ppi_cap, _ = sat[ppi_offset + y]
if ppo < 0: return
if ppi < 0: return
ppo_val = int(cdata[y, vector, 1])
ppi_val = int(0)
for tidx in range(ppi_cap):
t = state[ppi + tidx, vector]
if t >= TMAX: break
ppi_val ^= 1
# make new waveform at PPI
toggle = 0
if ppi_val:
state[ppi + toggle, vector] = TMIN
toggle += 1
if ppi_val != ppo_val:
state[ppi + toggle, vector] = ppi_time
toggle += 1
state[ppi + toggle, vector] = TMAX
@cuda.jit()
def capture_kernel(state, sat, ppo_offset, cdata, time, s_sqrt2, seed):
x, y = cuda.grid(2)
if ppo_offset + y >= len(sat): return
line, tdim, _ = sat[ppo_offset + y]
if line < 0: return
if x >= state.shape[-1]: return
vector = x
m = 0.5
acc = 0.0
eat = TMAX
lst = TMIN
tog = 0
ovl = 0
val = int(0)
final = int(0)
for tidx in range(tdim):
t = state[line + tidx, vector]
if t >= TMAX:
if t == TMAX_OVL:
ovl = 1
break
m = -m
final ^= 1
if t < time:
val ^= 1
if t <= TMIN: continue
if s_sqrt2 > 0:
acc += m * (1 + math.erf((t - time) / s_sqrt2))
eat = min(eat, t)
lst = max(lst, t)
tog += 1
if s_sqrt2 > 0:
if m < 0:
acc += 1
if acc >= 0.99:
val = 1
elif acc > 0.01:
seed = (seed << 4) + (vector << 20) + (y << 1)
seed = int(0xDEECE66D) * seed + 0xB
seed = int(0xDEECE66D) * seed + 0xB
rnd = float((seed >> 8) & 0xffffff) / float(1 << 24)
val = rnd < acc
else:
val = 0
else:
acc = val
cdata[y, vector, 0] = acc
cdata[y, vector, 1] = val
cdata[y, vector, 2] = final
cdata[y, vector, 3] = (val != final)
cdata[y, vector, 4] = eat
cdata[y, vector, 5] = lst
cdata[y, vector, 6] = ovl
@cuda.jit()
def assign_kernel(state, sat, ppi_offset, intf_len, tdata, time):
x, y = cuda.grid(2)
if y >= intf_len: return
line = sat[ppi_offset + y, 0]
if line < 0: return
sdim = state.shape[-1]
if x >= sdim: return
vector = x
a0 = tdata[y, 0, vector // 8]
a1 = tdata[y, 1, vector // 8]
a2 = tdata[y, 2, vector // 8]
m = np.uint8(1 << (7 - (vector % 8)))
toggle = 0
if a0 & m:
state[line + toggle, x] = TMIN
toggle += 1
if (a2 & m) and ((a0 & m) == (a1 & m)):
state[line + toggle, x] = time
toggle += 1
state[line + toggle, x] = TMAX
@cuda.jit(device=True)
def rand_gauss(seed, sd):
clamp = 0.5
if sd <= 0.0:
return 1.0
while True:
x = -6.0
for i in range(12):
seed = int(0xDEECE66D) * seed + 0xB
x += float((seed >> 8) & 0xffffff) / float(1 << 24)
x *= sd
if abs(x) <= clamp:
break
return x + 1.0
@cuda.jit()
def wave_kernel(ops, op_start, op_stop, state, sat, st_start, st_stop, line_times, sd, seed):
x, y = cuda.grid(2)
st_idx = st_start + x
op_idx = op_start + y
if st_idx >= st_stop: return
if op_idx >= op_stop: return
lut = ops[op_idx, 0]
z_idx = ops[op_idx, 1]
a_idx = ops[op_idx, 2]
b_idx = ops[op_idx, 3]
overflows = int(0)
_seed = (seed << 4) + (z_idx << 20) + (st_idx << 1)
a_mem = sat[a_idx, 0]
b_mem = sat[b_idx, 0]
z_mem, z_cap, _ = sat[z_idx]
a_cur = int(0)
b_cur = int(0)
z_cur = lut & 1
if z_cur == 1:
state[z_mem, st_idx] = TMIN
a = state[a_mem, st_idx] + line_times[a_idx, 0, z_cur] * rand_gauss(_seed ^ a_mem ^ z_cur, sd)
b = state[b_mem, st_idx] + line_times[b_idx, 0, z_cur] * rand_gauss(_seed ^ b_mem ^ z_cur, sd)
previous_t = TMIN
current_t = min(a, b)
inputs = int(0)
while current_t < TMAX:
z_val = z_cur & 1
if b < a:
b_cur += 1
b = state[b_mem + b_cur, st_idx]
b += line_times[b_idx, 0, z_val ^ 1] * rand_gauss(_seed ^ b_mem ^ z_val ^ 1, sd)
thresh = line_times[b_idx, 1, z_val] * rand_gauss(_seed ^ b_mem ^ z_val, sd)
inputs ^= 2
next_t = b
else:
a_cur += 1
a = state[a_mem + a_cur, st_idx]
a += line_times[a_idx, 0, z_val ^ 1] * rand_gauss(_seed ^ a_mem ^ z_val ^ 1, sd)
thresh = line_times[a_idx, 1, z_val] * rand_gauss(_seed ^ a_mem ^ z_val, sd)
inputs ^= 1
next_t = a
if (z_cur & 1) != ((lut >> inputs) & 1):
# we generate a toggle in z_mem, if:
# ( it is the first toggle in z_mem OR
# following toggle is earlier OR
# pulse is wide enough ) AND enough space in z_mem.
if z_cur == 0 or next_t < current_t or (current_t - previous_t) > thresh:
if z_cur < (z_cap - 1):
state[z_mem + z_cur, st_idx] = current_t
previous_t = current_t
z_cur += 1
else:
overflows += 1
previous_t = state[z_mem + z_cur - 1, st_idx]
z_cur -= 1
else:
z_cur -= 1
if z_cur > 0:
previous_t = state[z_mem + z_cur - 1, st_idx]
else:
previous_t = TMIN
current_t = min(a, b)
if overflows > 0:
state[z_mem + z_cur, st_idx] = TMAX_OVL
else:
state[z_mem + z_cur, st_idx] = a if a > b else b # propagate overflow flags by storing biggest TMAX from input