From 112d1c88bb21c0e38634ff64e46bc75c5731b41c Mon Sep 17 00:00:00 2001
From: DSPOM <dspom@protonmail.com>
Date: Fri, 1 Jul 2022 10:49:34 +0200
Subject: [PATCH] feat: implement python CMC test runner

---
 test_cases/testing.py | 451 +++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 444 insertions(+), 7 deletions(-)

diff --git a/test_cases/testing.py b/test_cases/testing.py
index a131d477d..beb391571 100644
--- a/test_cases/testing.py
+++ b/test_cases/testing.py
@@ -1,15 +1,27 @@
 #this file defines some common routines used by the OSDI test cases
-import subprocess
 import os
 import shutil
 import glob
 from pathlib import Path
+from typing import Optional
+import regex as re
+from subprocess import run, PIPE
+import pandas as pd
+import numpy as np
+from math import atan2
+import sys
 
 # specify location of Ngspice executable to be tested
 directory_testing = os.path.dirname(__file__)
-ngspice_path = os.path.join(directory_testing, "../debug/src/ngspice")
+ngspice_path = os.path.join(directory_testing, "../release/src/ngspice")
 ngspice_path = os.path.abspath(ngspice_path)
 
+rtol = 0.032
+atol_dc = 1e-14
+atol_ac = 4e-19
+
+twoPi = 8.0*atan2(1.0,1.0)
+
 def create_shared_objects(directory):
     c_files = []
     for c_file in glob.glob(directory + "/*.c"):
@@ -17,7 +29,7 @@ def create_shared_objects(directory):
         c_files.append(basename)
 
     for c_file in c_files:
-        subprocess.run(
+        run(
             [
                 "gcc",
                 "-c",
@@ -30,14 +42,31 @@ def create_shared_objects(directory):
             ],
             cwd=directory,
         )
-        subprocess.run(
+        run(
             ["gcc", "-shared", "-o", c_file + ".osdi", c_file + ".o", "-ggdb"],
             cwd=directory,
         )
-        subprocess.run(
+        run(
             ["mv", c_file + ".osdi", "test_osdi/" + c_file + ".osdi"], cwd=directory
         )
-        subprocess.run(["rm", c_file + ".o"], cwd=directory)
+        run(["rm", c_file + ".o"], cwd=directory)
+
+    # for va_file in glob.glob(directory + "/*.va"):
+    #     result = run(
+    #         [
+    #             "openvaf","-b", va_file
+    #         ],
+    #         # capture_output=True,
+    #         cwd=directory,
+    #     )
+
+    #     run(
+    #         ["cp", result.stdout[:-1], "test_osdi/" + Path(va_file).stem + ".osdi"], cwd=directory
+    #     )
+
+
+
+
 
 def prepare_dirs(directory):
     # directories for test cases
@@ -76,7 +105,7 @@ def prepare_netlists(directory):
         
 def run_simulations(dirs):
     for dir_i in dirs:
-        subprocess.run(
+        run(
             [
                 ngspice_path,
                 "netlist.sp",
@@ -92,3 +121,411 @@ def prepare_test(directory):
     run_simulations([dir_osdi, dir_built_in])
 
     return dir_osdi, dir_built_in
+
+
+
+def parse_list(line):
+    return (val for val in re.split(r"\s+", line) if val != '')
+
+def parse_temps(line):
+    return [temp for temp in parse_list(line)]
+
+
+class TestInfo:
+    biases: Optional[dict[str, str]] = None
+    bias_list: Optional[tuple[str, list[str]]]  = None
+    bias_sweep = None
+    temps: Optional[list[str]] = None
+    freqs: Optional[str] = None
+    dc_outputs: Optional[list[tuple[str, str]]] = None
+    ac_outputs:  Optional[dict[str,list[tuple[str, str, bool, str, str]]]] = None
+    instanceParameters: str= ""
+    modelParameters: str = ""
+    line: str = ""
+
+    def __init__(self, name, lines, parent):
+        self.name = name
+        self.lines= lines
+        self.parse()
+        if self.temps is None:
+            self.temps = parent.temps
+        self.pins = parent.pins
+        self.floating = parent.floating
+
+
+
+    def parse_temps(self):
+        temps = parse_temps(self.line)
+        if self.temps is None:
+            self.temps = temps
+        else:
+            self.temps += temps
+
+    def parse_model_params(self):
+        for param in parse_list(self.line):
+            path = Path(param)
+            if path.exists():
+                self.modelParameters = path.read_text()
+            else:
+                self.modelParameters += f"+ {param}\n"
+
+    def parse_instance_params(self):
+        for param in parse_list(self.line):
+            self.instanceParameters += f" {param}"
+
+
+    def parse_bias_list(self):
+        if self.bias_list:
+            raise ValueError(f"ERROR second bias_list spec {self.line}")
+        res = re.match(r"V\s*\(\s*(\w+)\s*\)\s*=", self.line)
+        pin = res[1]
+        vals = self.line[res.end():].strip()
+        vals = [val for val in re.split(r"\s*,\s*", vals)]
+        self.bias_list = (pin, vals)
+
+
+    def parse_biases(self):
+        if self.biases:
+            raise ValueError(f"ERROR second biases spec {self.line}")
+        self.biases = {}
+        for bias in parse_list(self.line):
+            res = re.match(r"V\s*\(\s*(\w+)\s*\)\s*=", bias)
+            pin = res[1]
+            val = bias[res.end():].strip()
+            self.biases[pin] = val
+
+    def parse_outputs(self):
+        for output in parse_list(self.line):
+            res = re.match(r"([IV])\s*\(\s*(\w+)\s*\)", output)
+            if res:
+                pin = res[2]
+                if res[1] == "I":
+                    output = f"i(v{pin})", f"I({pin})"
+                else:
+                    output = f"v({pin})", f"V({pin})"
+                if self.dc_outputs:
+                    self.dc_outputs.append(output)
+                else:
+                    self.dc_outputs = [output]
+                continue
+
+
+            res = re.match(r"([CG])\s*\(\s*(\w+)\s*,\s*(\w+)\s*\)", output)
+            if res:
+                kind = res[1]
+                pin1 = res[2]
+                pin2 = res[3]
+
+                if kind == "G":
+                    output = f"real(i(v{pin1}))", f"g({pin1},{pin2})", False, pin1, pin2
+                elif kind == "C":
+                    output = f"imag(i(v{pin1}))", f"c({pin1},{pin2})", True, pin1, pin2
+
+                if self.ac_outputs:
+                    if pin2 in self.ac_outputs:
+                        self.ac_outputs[pin2].append(output)
+                    else:
+                        self.ac_outputs[pin2] = [output]
+                else:
+                    self.ac_outputs = {pin2: [output]}
+                continue
+
+    def parse_frequency(self):
+        res = re.match(r"(lin|oct|dec)\s+(\S+)\s+(\S+)\s+(\S+)\s*", self.line)
+        kind = res[1]
+        num_steps = int(res[2])
+        start = res[3]
+        end = res[4]
+        if start != end:
+
+            if kind == "lin":
+                num_points = num_steps + 1
+            else:
+                num_points = num_steps
+        else:
+            assert num_steps == 1
+            num_points = 1
+        self.freqs = f"{kind} {num_points} {start} {end}"
+
+
+    def parse_bias_sweep(self):
+        res = re.match(r"V\s*\(\s*(\w+)\s*\)\s*=", self.line)
+        pin = res[1]
+        args = self.line[res.end():]
+        args = [float(arg) for arg in re.split(r"\s*,\s*", args)]
+        if len(args) != 3:
+            raise ValueError(f"bias sweep must have 3 arguments found {args} in {self.line}")
+        self.bias_sweep = (pin, args)
+
+
+    def try_parse(self, prefix: str, f):
+        if self.line.startswith(prefix):
+            self.line = self.line[len(prefix):].strip()
+            f()
+
+    def parse_line(self):
+        if self.try_parse("temperature", self.parse_temps):
+            return
+        if self.try_parse("modelParameters", self.parse_model_params):
+            return
+        if self.try_parse("instanceParameters", self.parse_instance_params):
+            return
+        if self.try_parse("biasList", self.parse_bias_list):
+            return
+        if self.try_parse("listBias", self.parse_bias_list):
+            return
+        if self.try_parse("biases", self.parse_biases):
+            return
+        if self.try_parse("output", self.parse_outputs):
+            return
+        if self.try_parse("outputs", self.parse_outputs):
+            return
+        if self.try_parse("biasSweep", self.parse_bias_sweep):
+            return
+        if self.try_parse("freq", self.parse_frequency):
+            return
+        if self.try_parse("frequency", self.parse_frequency):
+            return
+
+    def parse(self):
+        for line in self.lines:
+            self.line = line
+            self.parse_line()
+    
+    def gen_netlist(self, osdi_file, va_module, type_arg):
+        if self.bias_list:
+            bias_start = f"foreach bias {' '.join(self.bias_list[1])}\nalter v{self.bias_list[0]}=$bias"
+            bias_end = "end"
+        else:
+            bias_start = bias_end = ""
+        
+        if self.dc_outputs:
+            if not self.bias_sweep:
+                raise ValueError("dc bias sweep msising")
+            outputs = " ".join(output for output, _ in self.dc_outputs)
+            sweep = f"dc v{self.bias_sweep[0]} {self.bias_sweep[1][0]} {self.bias_sweep[1][1]} {self.bias_sweep[1][2]}\n wrdata {self.dc_results_path()} {outputs}"
+        elif self.ac_outputs:
+            freqs = self.freqs
+            if not self.freqs:
+                freqs = f"lin 1 {1/twoPi} {1/twoPi}"
+            if self.bias_sweep:
+                if self.bias_list:
+                    bias_start += "\n"
+                    bias_end += "\n"
+                vals = np.arange(self.bias_sweep[1][0], self.bias_sweep[1][1] + self.bias_sweep[1][2]*0.1, self.bias_sweep[1][2])
+                vals = [str(val) for val in vals]
+                bias_start += f"foreach bias {' '.join(vals)}\nalter v{self.bias_sweep[0]}=$bias"
+                bias_end += "end"
+
+            sweep = ""
+            for pin, outputs in self.ac_outputs.items():
+                sweep += f"alter v{pin} ac = 1\nac {freqs}\n"
+                outputs = " ".join(output[0] for output in outputs)
+                sweep += f"wrdata {self.ac_results_path(pin)} {outputs}\n"
+                sweep += f"alter v{pin} ac = 0\n"
+        else:
+            return ""
+
+        biases = self.biases
+        if not biases:
+            biases = dict()
+
+        source = "\n".join(f"v{pin} {pin} {0} dc={biases.get(pin, 0)}" for pin in self.pins if not pin in self.floating)
+        source += "".join(f"\nr{i} {pin} {0} r=1G" for i,pin in enumerate(self.floating))
+
+        return f"""CMC testsuite {self.name}
+.options abstol=1e-15
+
+{source}
+
+.model test_model {va_module}
+{self.modelParameters} {type_arg}
+
+A1 {' '.join(self.pins)} test_model {self.instanceParameters}
+
+.control
+pre_osdi {osdi_file}
+
+set filetype=ascii
+set wr_vecnames
+set wr_singlescale
+set appendwrite
+
+foreach tamb {' '.join(self.temps)}
+  set temp=$tamb
+  {bias_start}
+    {sweep}
+   {bias_end}
+end
+quit 0
+.endc
+.end
+"""
+
+    def dc_results_path(self) -> Path:
+        return Path("results")/f"{self.name}.ngspice"
+
+    def ac_results_path(self, pin: str) -> Path:
+        return Path("results")/f"{self.name}_{pin}.ngspice"
+
+    def run(self, osdi_file, va_module, type_arg):
+        if not (self.dc_outputs or self.ac_outputs):
+            return
+
+        print(f"running {self.name}...")
+
+        netlist_path = Path("netlists")/f"{self.name}.sp"
+        netlist = self.gen_netlist(osdi_file, va_module, type_arg)
+        Path(netlist_path).write_text(netlist)
+
+        res = run([ngspice_path, netlist_path, "-b"], capture_output=True)
+        res.check_returncode()
+        # res.check_returncode()
+
+        reference_path = Path("reference")/f"{self.name}.standard"
+        references = pd.read_csv(reference_path, sep="\\s+")
+
+        if self.dc_outputs:
+            results_path = self.dc_results_path()
+
+            if not results_path.exists():
+                print(f"ERROR check failed for {self.name}\nsimulation file is missing - likely convergence issues!")
+                return
+
+            results = pd.read_csv(results_path, sep="\\s+")
+            results = results.apply(pd.to_numeric, errors='coerce')
+            firstcol = results.iloc[:,1].to_numpy()
+            results = results[np.bitwise_not(np.isnan(firstcol))]
+
+            for result_col, ref_col in self.dc_outputs:
+                reference = references[ref_col].to_numpy()
+                result = results[result_col].to_numpy()
+                if "I(" in ref_col:
+                    result = -result
+
+                adiff = np.abs(result-reference)
+                rdiff = adiff/reference
+                err = np.bitwise_not(np.bitwise_or(rdiff < rtol, adiff < atol_dc))
+                if not np.any(err):
+                    continue
+                maxatol = np.max(adiff[err])
+                maxrtol = np.max(rdiff[err])
+                print(f"ERROR check failed for {ref_col}\nrtol={maxrtol} atol={maxatol}\nresult:\n{result[err]}\nreference:\n{reference[err]}\nrtol:\n{rdiff[err]}")
+
+        elif self.ac_outputs:
+            for pin, outputs in self.ac_outputs.items():
+                results_path = self.ac_results_path(pin)
+                if not results_path.exists():
+                    print(f"ERROR check failed for {self.name} (ac {pin})\nsimulation file is missing - likely convergence issues!")
+                    continue
+
+                results = pd.read_csv(results_path, sep="\\s+")
+                results = results.apply(pd.to_numeric, errors='coerce')
+                firstcol = results.iloc[:,1].to_numpy()
+                results = results[np.bitwise_not(np.isnan(firstcol))]
+
+                for result_col, ref_col, is_cap, pin1, pin2  in outputs:
+                    result = results[result_col].to_numpy()
+                    reference = references[ref_col].to_numpy()
+                    if is_cap:
+                        if"Freq" in references:
+                            result = result /(twoPi*results["frequency"])
+                        if pin1 == pin2:
+                            result = -result
+                    else:
+                        result = -result
+
+                    adiff = np.abs(result-reference)
+                    rdiff = adiff/reference
+                    err = np.bitwise_not(np.bitwise_or(rdiff < rtol, adiff < atol_ac))
+                    if not np.any(err):
+                        continue
+                    maxatol = np.max(adiff[err])
+                    maxrtol = np.max(rdiff[err])
+                    print(f"ERROR check failed for {ref_col}\nrtol={maxrtol} atol={maxatol}\nresult:\n{result[err]}\nreference:\n{reference[err]}\nrtol:\n{rdiff[err]}")
+
+
+
+
+
+        
+
+    
+def removeComments(string):
+    string = re.sub(re.compile(r"/\*.*?\*/",re.DOTALL ) ,"" ,string) # remove all occurrences streamed comments (/*COMMENT */) from string
+    string = re.sub(re.compile(r"//.*?\n" ) ,"" ,string) # remove all occurrence single-line comments (//COMMENT\n ) from string
+    return string
+
+class QaSpec:
+    temps: list[str]
+    pins: list[str]
+    floating: list[str]
+    tests: list[TestInfo]
+    dir: Path
+
+    def __init__(self, dir: Path):
+        self.dir = dir
+        self.temps = []
+        self.pins = []
+        self.tests = []
+        self.floating = []
+        self.parse()
+
+    def parse(self):
+        old_dir = os.getcwd()
+        os.chdir(self.dir)
+        qa_spec = Path("qaSpec").read_text()
+        qa_spec = removeComments(qa_spec)
+        lines = [line.strip() for line in qa_spec.split('\n')]
+
+        i = 0
+        while i < len(lines):
+            line = lines[i]
+            i+= 1
+            if line.startswith("temperature"):
+                line = line[len("temperature"):]
+                self.temps = parse_temps(line)
+            elif line.startswith("pins"):
+                line = line[len("pins"):]
+                self.pins = [pin for pin in re.findall(r"\w+", line) if pin != "pins"]
+
+            elif line.startswith("float") or line.startswith("floating"):
+                self.floating = [pin for pin in re.findall(r"\w+", line) if pin != "floating" and pin != "float"]
+            elif line.startswith("test"):
+                test_name = line[4:].strip()
+                start = i
+                while i < len(lines) and lines[i] != "":
+                    i += 1
+                end = i
+
+                test = TestInfo(test_name, lines[start:end], self)
+                self.tests.append(test)
+
+        os.chdir(old_dir)
+
+
+    def run(self, va_file, va_module, type_arg, filter=None):
+        result = run(
+            ["openvaf","-b", va_file],
+            stdout=PIPE,
+        )
+        result.check_returncode()
+        osdi_file = result.stdout[:-1].decode("utf-8") 
+        
+        old_dir = os.getcwd()
+        os.chdir(self.dir)
+
+        dirpath = Path('netlists')
+        if dirpath.exists() and dirpath.is_dir():
+            shutil.rmtree(dirpath)
+        os.mkdir("netlists")
+
+        dirpath = Path('results')
+        if dirpath.exists() and dirpath.is_dir():
+            shutil.rmtree(dirpath)
+        os.mkdir("results")
+        for test in self.tests:
+            if filter and not test.name in filter:
+                continue
+            test.run(osdi_file, va_module, type_arg)
+        os.chdir(old_dir)