An integrated test framework for digital components of Cadence

toxic.py

#coding=utf-8
from __future__ import print_function

import os
import subprocess
import inspect
import traceback

__author__ = "Nigshoxiz"

# ref : http://stackoverflow.com/questions/287871/print-in-terminal-with-colors-using-python
class Colors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

    @staticmethod
    def format_header(str):
        return "%s%s%s" % (Colors.HEADER, str, Colors.ENDC)
    @staticmethod
    def format_blue(str):
        return "%s%s%s" % (Colors.OKBLUE, str, Colors.ENDC)
    @staticmethod
    def format_underline(str):
        return "%s%s%s" % (Colors.UNDERLINE, str, Colors.ENDC)
    @staticmethod
    def format_green(str):
        return "%s%s%s" % (Colors.OKBLUE, str, Colors.ENDC)
    @staticmethod
    def format_warning(str):
        return "%s%s%s" % (Colors.WARNING, str, Colors.ENDC)
    @staticmethod
    def format_fail(str):
        return "%s%s%s" % (Colors.FAIL, str, Colors.ENDC)
    @staticmethod
    def format_bold(str):
        return "%s%s%s" % (Colors.BOLD, str, Colors.ENDC)

class SpectreWaveForm:
    # generate waveform
    def _math_abs(self, a, b):
        if a > b:
            return a - b
        else:
            return b - a

    def __init__(self, input_pins, filename = None):
        self.vdd_value = 2.5 # unit : V
        self.input_pins = input_pins

        self._rise_time = 0.05 # unit: ns
        self._fall_time = self._rise_time
        self._wave_width = 5
        self._space_time = 0.05

        self._capacitor_val = 0.3 # unit: pF
        self.file_str = []

        self.output_file = filename

        self._total_time = 0.0

        self.wave_bits = 0
    def set_vdd(self, val):
        self.vdd_value = val

    def get_vdd(self):
        return self.vdd_value

    def set_rise_time(self, val):
        self._rise_time = val
        self._fall_time = self._rise_time

    def set_wave_width(self, val):
        self._wave_width = val

    def set_space_time(self, val):
        self._space_time = val

    def set_capacitor(self, val):
        self._capacitor = val

    def _title_line(self):
        s = "simulator lang=spectre"
        self.file_str.append(s)

    def _vsource_line(self):
        s = "v0 (vdd! 0) vsource type=dc dc=%s" % self.vdd_value
        self.file_str.append(s)

    def _input_pin_line(self, index, sig_series):
        '''

        :param index: index of pin, e.g : A1 -> index = 1
        :param sig_series: array of signal [0,0,0,0,0,1], it will transform to wave data
        :return:
        '''
        #template : "v1 (A 0) vsource type=pwl wave=[0n 0 2n 0 2.2n 2.5 9.2n 2.5 9.4n 0]"

        wave = []
        _time = 0
        def _append(_tuple):
            wave.append(_tuple[0])
            wave.append(_tuple[1])

        # first add (0,0) <- (<time>, <voltage>)
        _append((0,0))

        for i in sig_series:
            # space time
            _time += self._space_time
            _space_voltage = 0.0
            _append((_time, _space_voltage))

            if i == 1:
                # rise
                _time += self._rise_time
                _rise_voltage = self.vdd_value
                _append((_time, _rise_voltage))
                #width
                _time += self._wave_width
                _append((_time, _rise_voltage))

                # fall
                _time += self._fall_time
                _fall_voltage = 0.0
                _append((_time,_fall_voltage))
            elif i == 0:
                _time += (self._rise_time + self._wave_width + self._fall_time)
                _voltage = 0.0
                _append((_time, _voltage))

        waveform_string = ""

        _index = 0
        for x in wave:
            if self._math_abs(x, int(x)) < 0.005:
                waveform_string += "%s" % int(x)
            else:
                waveform_string += "%.2f" % x

            if _index % 2 == 0:
                waveform_string += "n"

            if _index != len(wave) - 1:
                waveform_string += " "
            _index += 1

            s = "v%s (%s 0) vsource type=pwl wave=[%s]" % (
                index + 1,
                self.input_pins[index],
                waveform_string
            )

        self.file_str.append(s)
        self.wave_bits = len(sig_series)

    def calc_total_time(self):
        _time = (self._space_time + self._rise_time + self._fall_time + self._wave_width)
        return _time * self.wave_bits

    def _capacitor(self, out_pin, _index = None):
        if _index == None:
            s = "ca (%s 0) capacitor c=%sp" % (out_pin, self._capacitor_val)
        else:
            s = "ca%s (%s%s 0) capacitor c=%sp" % (_index, out_pin, _index, self._capacitor_val)
        self.file_str.append(s)

    def _save_data(self, out_pin):
        s = "save %s" % out_pin
        self.file_str.append(s)

    def export(self):
        _str = "\n".join(self.file_str)
        if self.output_file == None:
            print(_str)
        else:
            f = open(self.output_file, "w+")
            f.write(_str)
            f.close()

class DirectoryVariables:
    instance = None

    def __init__(self):
        self.simulation_dir = os.getcwd() # total_dir
        self.library_name = ""
        self.cellview_name = ""
        self.simulation_type = ""
        self.simulation_temperature = "25"

        self.model_file = ""
        self._home_dir = os.getcwd()
        pass

    @staticmethod
    def getInstance():
        if DirectoryVariables.instance == None:
            DirectoryVariables.instance = DirectoryVariables()
        return DirectoryVariables.instance

    def init(self, **kwargs):
        for item in kwargs:
            if getattr(self, item) != None:
                setattr(self, item, kwargs[item])

class OceanFileFormatter:
    def __init__(self):
        pass

    def _format_string(self, val):
        '''
        "<string>"
        '''
        return "\"%s\"" % val
        pass

    def _format_label(self, val):
        '''
        '<label>
        :return:
        '''
        return "'%s" % val

    def _format_key_value(self, key, value):
        '''
        ?<key> <value>
        '''
        return "?%s %s" % (key, value)

    def _format_array(self, *args):
        '''
        (<str1> <str2>)
        '''
        arr_str = " ".join(args)
        return "(%s)" % arr_str

    def _format_function(self, name, *args):
        '''
        <func_name>( <inner_data> )
        '''
        if len(args) > 0:
            return "%s( %s )" % (name, " ".join(args))
        else:
            return "%s()" % name

    def _format_equal(self, a, b):
        return "%s=%s" % (a, b)

class OceanFileGenerator(OceanFileFormatter):
    def __init__(self, filename = None):
        OceanFileFormatter.__init__(self)
        self.dir_vars = DirectoryVariables.getInstance()
        if filename == None:
            self.file_name = os.getcwd() +"/.ocean"
        else:
            self.file_name = filename

    def gen_netlist(self):
        F = self._format_function
        S = self._format_string
        L = self._format_label
        A = self._format_array
        KV = self._format_key_value

        V = self.dir_vars
        # project dir
        return [
            F("envSetVal", S("asimenv.startup"), S("projectDir"), L("string"), S(V.simulation_dir)),
            F("simulator", L("spectre")),
            F("design", S(V.library_name), S(V.cellview_name), S(V.simulation_type)),
            F("createNetlist", KV("recreateAll", "t"))
        ]

    def gen_pre_simulation(self, stop_time):
        F = self._format_function
        S = self._format_string
        L = self._format_label
        A = self._format_array
        KV = self._format_key_value

        V = self.dir_vars

        s_s_n_n = "spectre/schematic/netlist/netlist"
        s_s_n   = "spectre/schematic/netlist"
        s_s     = "spectre/schematic"
        # project dir
        return [
            F("ocnWaveformTool", L("wavescan")),
            F("simulator", L("spectre")),
            F("design", S(os.path.join(V.simulation_dir, V.cellview_name, s_s_n_n))),
            F("resultsDir", S(os.path.join(V.simulation_dir, V.cellview_name, s_s))),
            F("modelFile", L(A(S(V.model_file), S(""))) ),
            #F("stimulusFile", KV("xlate", "nil"), S(os.path.join(V.simulation_dir, V.cellview_name, s_s_n, "_graphical_stimuli.scs")))
            #F("definitionFile", S("filename"))
            F("analysis", L("tran"), KV("stop", S("%.1fn" % stop_time))),
            F("temp", V.simulation_temperature)
        ]

    def gen_definition_file(self, def_name):
        F = self._format_function
        S = self._format_string
        L = self._format_label
        A = self._format_array
        KV = self._format_key_value

        V = self.dir_vars
        return [
            F("definitionFile", S(os.path.join(V.simulation_dir, V.cellview_name, "%s.def" % def_name)))
        ]

    def gen_run_simulation(self):
        return ["run()"]

    def gen_post_simulation(self, logic_fn):
        E = self._format_equal
        F = self._format_function
        S = self._format_string
        L = self._format_label
        A = self._format_array
        KV = self._format_key_value

        V = self.dir_vars

        fname = "test_%s" % V.cellview_name
        output_file = os.path.join(V.simulation_dir, V.cellview_name, "%s.out" % fname)
        simu_arr = [
            E("of", F("outfile", S(output_file), S("w") ))
        ]

        if inspect.isfunction(logic_fn):
            _arr = logic_fn()
            for i in _arr:
                simu_arr.append(i)

        simu_arr.append(
            F("close","of")
        )

        return simu_arr

    def export(self, *args):
        str_arr = []
        for i in args:
            for j in i:
                str_arr.append(j)

        str = "\n".join(str_arr)
        f = open(self.file_name, "w+")
        f.write(str)
        f.close()

class Process:
    @staticmethod
    def ocean(exec_filename = None):
        V = DirectoryVariables.getInstance()

        wv_name = "test_%s" % V.cellview_name
        if exec_filename == None:
            exec_filename = os.path.join(V.simulation_dir, V.cellview_name, wv_name+".ocn")
        cmd = "ocean < %s" % os.path.realpath(exec_filename)
        p = subprocess.Popen(cmd, shell=True,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT,
                             stdin=subprocess.PIPE)
        BL = Colors.format_blue
        while True:
            line = p.stdout.readline()
            print(BL(line), end='')
            if not line: break

class PinNameExpander:
    @staticmethod
    def expand(pin_tuple, policy = "alphabet",backslash=True):
        POLICY_NUMBER = 1
        POLICY_ALPHABET = 2

        _policy = POLICY_ALPHABET

        _alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
        final_arr = []

        if policy == "alphabet":
            _policy = POLICY_ALPHABET
        elif policy == "numeric" or policy == "integer" or policy == "number":
            _policy = POLICY_NUMBER
        else:
            _policy = POLICY_ALPHABET

        def _isnumber(obj):
            return isinstance(obj, int)

        def _is_in_alphabet(strs):
            if len(strs) != 1:
                return None
            elif strs.isupper() == True:
                return _alphabet.find(strs)
            else:
                return None

        expand_arr = []
        # store
        last = None
        for i in pin_tuple:
            if _isnumber(i) == False:
                if last != None and _isnumber(last) == False:
                    expand_arr.append((last, -1))
            else:
                expand_arr.append((last, i))
            if _isnumber(i) == False:
                last = str(i)
            else:
                last = i

        if _isnumber(last) == False:
            expand_arr.append((last, -1))

        # expand
        for tup in expand_arr:
            name, times = tup
            if _policy == POLICY_ALPHABET:
                if times == -1:
                    final_arr.append(name)
                else:
                    for i in range(0, times):
                        _index = _is_in_alphabet(name)
                        if _index != None:
                            final_arr.append(_alphabet[(_index+i) % 26])
                        else:
                            # for other strings ,do like what
                            final_arr.append(name + str(i))
            elif _policy == POLICY_NUMBER:
                if times == -1:
                    final_arr.append(name)
                else:
                    for i in range(0, times):
                        if backslash == True:
                            final_arr.append(name +  "\\<" + str(i) + "\\>")
                        else:
                            final_arr.append(name +  "<" + str(i) + ">")
        return final_arr
           

class OutputAnalyser:
    def __init__(self, vdd_value, filename = None):
        V = DirectoryVariables.getInstance()
        self.wv_name = "test_%s" % V.cellview_name        
        if filename == None:
            self.filename = os.path.join(V.simulation_dir, V.cellview_name, self.wv_name+".out")
        else:
            self.filename = filename
        
        self.vdd_value = vdd_value
    
    def _threshold(self, data):
        vdd = self.vdd_value
        if data > vdd * 0.9 and data < vdd * 1.1:
            return (1, data)
        elif data > vdd * -0.1 and data < vdd * 0.1:
            return (0, data)
        else:
            return (-1, data)

    def read_data(self):
        try:
            arr = []
            f = open(self.filename, "r")
            for i in f.readlines():
                i = i.strip()
                i_split = i.split(" ")
    
                row_arr = []                
                for dat in i_split:
                    v, _ = self._threshold(float(dat))
                    row_arr.append(v)
                arr.append(row_arr)
            return arr
        except:
            a = []
            traceback.print_exc()
            f.close()
            return a
            

class BenchmarkSimulator(SpectreWaveForm, OceanFileFormatter):
    def __init__(self, input_tup, policy = "alphabet"):
        V = DirectoryVariables.getInstance()

        self.wv_name = "test_%s" % V.cellview_name
        self.wave_filename   = os.path.join(V.simulation_dir, V.cellview_name, self.wv_name+".def")
        self.test_ocean_file = os.path.join(V.simulation_dir, V.cellview_name, self.wv_name+".ocn")
        
        input_pins = PinNameExpander.expand(input_tup, policy = policy)
        SpectreWaveForm.__init__(self, input_pins, filename=self.wave_filename)
        OceanFileFormatter.__init__(self)
        self.g = OceanFileGenerator(filename=self.test_ocean_file)

    def generate_test_file(self, wave_input, output_tup, policy = "alphabet"):
        
        output_pins = PinNameExpander.expand(output_tup, policy = policy)

        def digital_output_func():
            outputs_name = PinNameExpander.expand(output_tup, policy = policy, backslash = False) 
            if outputs_name == None:
                raise ValueError("outputs pin name shouldn't be null!")

            E = self._format_equal
            F = self._format_function
            S = self._format_string
            L = self._format_label
            A = self._format_array
            KV = self._format_key_value

            _len = len(wave_input[0])
            
            unit_time = self.calc_total_time() / _len
            offset_time = unit_time / 2
            time_exp = "%.2fe-9 + i * %.2fe-9" % (offset_time, unit_time)

            # gen outputs printf
            format_string = ""
            params_string = ""
            outputs_arr = []

            if isinstance(outputs_name, list):
                outputs_arr = outputs_name
            else:
                outputs_arr.append(outputs_name)
    
            for item in outputs_arr:
                format_string += ("%.3f ")
                params_string += ","+F("value", F("v", S("/"+item)), time_exp)

            format_string = "%s\\n" % format_string

            return [
                F("selectResults", L("tran")),
                F("for", "i", "0", str(_len-1),
                    F("fprintf", "of", S(format_string) + params_string)
                )
            ]

        # gen waveform
        self.set_vdd(VDD)
        self.set_wave_width(5)
        self.set_capacitor(30)
        self.set_rise_time(0.05)
        self.set_space_time(0.05)

        self._title_line()
        self._vsource_line()

        _index = 0
        for wave_arr in wave_input:
            self._input_pin_line(_index , wave_arr)
            _index += 1
        
        for pin_name in output_pins:    
            self._save_data(pin_name)
        self.export()
        
        # ocean file generator
        self.g.export(
            self.g.gen_pre_simulation(self.calc_total_time()),
            self.g.gen_definition_file(self.wv_name),
            self.g.gen_run_simulation(),
            self.g.gen_post_simulation(digital_output_func)
        )

    def gen_data_ALL(self, num_of_pin):
        if num_of_pin > 12:
            raise Exception("Pin number is too Biiiiiiiiiig! ( %s > 12 )" % num_of_pin)
        total_test = 1 << num_of_pin
        num = 0
        
        input_array = []
        
        for x in range(0, num_of_pin):
            empty_arr = []
            input_array.append(empty_arr)

        while num < total_test:
            for digit in range(0, num_of_pin):
                input_array[digit].append((num & (1 << digit)) >> digit)
            num += 1
        return input_array

# operation functions
def Init():
    B = Colors.format_bold
    print("\n"+B("[INFO]")+" Initializing...")
    DirectoryVariables.getInstance().init(
        simulation_dir = os.path.realpath(SIMULATION_DIR),
        model_file     = MODEL_FILE,
        simulation_temperature = str(TEMPERATURE),
        library_name = LIBRARY_NAME,
        cellview_name = CELLVIEW_NAME,
        simulation_type = SIMULATION_TYPE
    )

def CreateNetlist():
    H = Colors.format_header
    B = Colors.format_bold
    GR = Colors.format_green
    U  = Colors.format_underline

    g = OceanFileGenerator()

    hint_str = """\n\
Command file (.ocean) generation finish!
However, to create the netlist, you have to execute the command file manually in `icfb`
What you're asked to do is simple:

1. Open `icfb` : prompt `%s` in terminal
2. On Command-Line Window in icfb, prompt `%s`
3. Observe the output

If everything is fine, just press %s to continue.""" % (
        U("icfb &"),
        U("load(\".ocean\")"),
        B("[ENTER]")
    )

    print("\n"+B("[INFO]")+" Start Netlist Generating Process...")
    g.export(g.gen_netlist())
    print(hint_str)

    raw_input("[continue] > ")
    print("\n%s Netlist Generating Process Finish!" % B("[INFO]"))

def WelcomeBanner():
    H = Colors.format_header
    BL = Colors.format_blue
    GR = Colors.format_green
    U  = Colors.format_underline
    B  = Colors.format_bold
    print("    " + H("Welcome to %s!" % GR("toxic.py") ))
    print("")
    print("    " + "Please edit this file using your most favourite editor,")
    print("    " + "    " + U("e.g. `gedit toxic.py`"))
    print("    " + "drag the scroll bar to the very end,")
    print("    " + "follow the instructions in the comments to do what you want!")
    print("")
    print("  " + BL("by Nigshoxiz"))
    print("  " + ("10/10/2016"))

def Benchmark(inputs=None, input_sequence=None, outputs=None, output_sequence=None,assert_method=None,gen_policy=None, _run=False):
    # just for debugging this script        
    if _run == False:
        return None

    H = Colors.format_header
    BL = Colors.format_blue
    GR = Colors.format_green
    U  = Colors.format_underline
    B  = Colors.format_bold
    F  = Colors.format_fail

    def _print_input(input_arr, index):
        s = ""
        l = len(input_arr)
        for x in range(0, l):
            s += str(input_arr[l-1-x][index])
        return s   

    input_arr = []
    output_arr = []
    simu = BenchmarkSimulator(inputs, policy=input_sequence)
    print("\n"+B("[INFO]")+" Start Simulation...")
    print(B("[INFO]")+" Generating test file...")
    
    pin_num = len(PinNameExpander.expand(inputs, policy=input_sequence))
    input_arr = simu.gen_data_ALL(pin_num)
    simu.generate_test_file(input_arr, outputs, policy=output_sequence)

    print("\n"+B("[INFO]")+" Run simulation...")
    Process.ocean()
    
    print("\n"+B("[INFO]")+" Simulation Finish!") 
    print(B("[INFO]")+" Now let's assert data!\n")
    analyser = OutputAnalyser(VDD)
    output_arr = analyser.read_data()
    
    print("-"*80)
    print("    %s %s, %s %s\n" % (B("Pins:"), pin_num, B("Tests:"), len(output_arr)))
    _index = 0
    while _index < len(output_arr):
        ar = []
        for j in range(len(input_arr)):
            ar.append(input_arr[j][_index]) 
        assert_result, exp_data = assert_method(ar, output_arr[_index])
        
        assert_str = GR("OK")
        if assert_result == False:
            assert_str = F("FAIL")
        
        print("  No. %s\tIN=%s\tOUT=%s\tEXP=%s\t%s" % ((_index+1), _print_input(input_arr,_index), output_arr[_index], exp_data, assert_str))
        _index += 1

# assert Functions
def NOT(inputs, output):
    _in = inputs[0]
    _expect = ~_in & 1
    _assert = (_expect == output[0])
    return (_assert, _expect)

def NAND(inputs, output):
    _and = 1
    for i in inputs:
        _and = _and & i
    _expect = ~_and & 1
    _assert = (_expect == output[0])
    return (_assert, _expect)

def NOR(inputs, output):
    _or = 0
    for i in inputs:
        _or = _or | i

    _expect = ~_or & 1
    _assert = (_expect == output[0])
    return (_assert, _expect)

def XOR(inputs, output):
    _expect = (inputs[0] + inputs[1]) & 1
    _assert = (_expect == output[0])
    return (_assert, _expect)

def MUX(inputs, output):
    input_len = (3,6,11,20,37) # sel bits = (1,2,3,4,5)
    _index = 1
    for i in input_len:
        if len(inputs) == i:
            break
        _index += 1
    
    if _index >= 6:
        raise ValueError("Not match!")
    
    _sel_index = 0
    _counter = len(inputs)-1#1 << _index
    while _counter >= (1<< _index): #len(inputs):
        _sel_index <<= 1
        _sel_index += inputs[_counter]
        _counter -= 1
    
    _expect = inputs[_sel_index]
    _assert = (_expect == output[0])
    return (_assert, _expect)

# decoder
# inputs : A0, A1, A2, A3, ... NEN
# outputs: OUT<0>, OUT<1>, ...
def DECODER(inputs, outputs):
    bits = len(inputs) - 1
    if (1 << bits) != len(outputs):
        raise ValueError("not match!")
    else:
        out_val = 0
        sel = 0
        _i = bits - 1
        _o = len(outputs) - 1
        while _i >= 0:
            sel <<= 1
            sel += inputs[_i]
            _i -= 1
        while _o >= 0:
            out_val <<= 1
            out_val += outputs[_o]
            _o -= 1
    NOT_EN = inputs[-1]

    if NOT_EN == 1:
        _expect = 0
        _assert = (out_val == _expect)
        return (_assert, _expect)
    else:
        _expect = 1 << sel
        _assert = (out_val == _expect)
        return (_assert, _expect)
            
def DECODER_2(inputs, outputs):
    i = inputs
    i.append(0)
    _, _expect  =DECODER(i, outputs)
    bits = len(outputs)
    _exp = ~ _expect & (~(1<<bits))

    return (False, _exp)

def DECODER_32(inputs, outputs):
    arr = inputs
    arr.append(0)
    return DECODER(arr, outputs)
#########################################################################
#########################################################################
#                                                                       #
#                     toxic.py User Edit Area                           #
#                                                                       #
#########################################################################
#########################################################################

MODEL_FILE = "/afs/ee.ust.hk/staff/ee/dept/public/elec301/model/t02d.scs"
SIMULATION_DIR = "./on9_simulation"

LIBRARY_NAME = "on9_shifter"
CELLVIEW_NAME = "decoder-5to32"
SIMULATION_TYPE = "schematic"

TEMPERATURE = 25
VDD = 2.5

# Variable initialization, please never delete this line!
Init()    
WelcomeBanner()
# Before simulation, we have to generate netlist first
CreateNetlist()
# Then do the benchmark
# Inputs : A, B, C, D, ...
# Output: OUT

Benchmark(
    inputs          = ["A0","A1","A2","A3","A4"],
    input_sequence  = "alphabet", # "alphabet" || "integer" 
    outputs         = ["OUT", 32],
    output_sequence = "number",
    assert_method   = DECODER_32,
    gen_policy      = "all", # "all" or "random" or your customize function
    
    # just comment this line to shutdown the function (not run)
    _run            = True
)
# Benchmark("A",4 , NAND, alphabet=True)