Add access pattern log test for reference generator

Change-Id: I374c45fff60d38b8b5b76ee813d706f19160f2ab

libnux/dls_vx.h

@@ -49,3 +49,6 @@ static constexpr omnibus_address_t dls_syndrv_base = 0x1c000000ul;
 
 /* Address for spike injection */
 static constexpr omnibus_address_t dls_spike_base = 0x1c000040ul;
+
+/* MADC and reference current generator configuration base address */
+static constexpr omnibus_address_t madc_base_address = (1ul << 19 | 1ul << 18);

test/with_hostcode/log_access_pattern_host.py

+import json
+import os
+from os.path import join
+import unittest
+
+from dlens_vx.halco import iter_all
+from dlens_vx.halco import PPUOnDLS
+from dlens_vx.sta import PlaybackProgramExecutor, generate, DigitalInit
+from dlens_vx import logger
+from dlens_vx.tools.run_ppu_program import load_and_start_program, \
+    wait_until_ppu_finished, stop_program
+
+_THIS_DIR = os.path.dirname(os.path.realpath(__file__))
+TEST_BINARY_PATH = os.environ.get("TEST_BINARY_PATH",
+                                  join(_THIS_DIR,
+                                       os.pardir,
+                                       os.pardir,
+                                       os.pardir,
+                                       "build",
+                                       "libnux",
+                                       "test",
+                                       "with_hostcode")
+                                  )
+
+FLANGE_LOG_PATH = os.environ.get("FLANGE_LOG_PATH",
+                                 join(_THIS_DIR,
+                                      os.pardir,
+                                      os.pardir,
+                                      os.pardir,
+                                      "hxfpga",
+                                      "units",
+                                      "synplify_wrapper",
+                                      "sim",
+                                      "xrun.log")
+                                 )
+
+
+class LibnuxAccessPatternTestVx(unittest.TestCase):
+    EXECUTOR = PlaybackProgramExecutor()
+
+    class LogParser:
+        LOG_PREFIX = "HX_SIMULATION_LOG: "
+
+        def __init__(self, pattern: str, offset_bytes: int):
+            self.pattern = pattern
+            self.offset_bytes = offset_bytes
+            self.logfile = open(FLANGE_LOG_PATH, "rt")
+
+        def __iter__(self):
+            self.logfile.seek(self.offset_bytes)
+            return self
+
+        def __next__(self):
+            for line in self.logfile:
+                if line.startswith(self.LOG_PREFIX):
+                    event = json.loads(line[len(self.LOG_PREFIX):])
+                    if self.pattern in event:
+                        return event[self.pattern]
+
+            raise StopIteration
+
+        def __del__(self):
+            self.logfile.close()
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        # Connect to some executor (sim or hardware)
+        cls.EXECUTOR.connect()
+
+        # Initialize the chip
+        init_builder, _ = generate(DigitalInit())
+        cls.EXECUTOR.run(init_builder.done())
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        # Disconnect the executor
+        cls.EXECUTOR.disconnect()
+
+    def run_ppu_program(self, ppu: PPUOnDLS, program_path: str, timeout: int):
+        load_and_start_program(self.EXECUTOR, program_path, ppu)
+        wait_until_ppu_finished(self.EXECUTOR, timeout=timeout, ppu=ppu)
+        ret_code = stop_program(self.EXECUTOR, ppu=ppu,
+                                print_mailbox=False)
+
+        self.assertEqual(0, ret_code,
+                         f"PPU exit code was {ret_code}, expected 0.")
+
+    def test_refgen_access(self):
+        log = logger.get("LibnuxAccessPatternTestVx.test_refgen_access")
+        program = join(TEST_BINARY_PATH, "refgen_access_pattern-ppu.bin")
+        initial_logsize = os.path.getsize(FLANGE_LOG_PATH)
+        log.debug("Initial size of %s: %dbytes." % (FLANGE_LOG_PATH,
+                                                    initial_logsize))
+
+        for ppu in iter_all(PPUOnDLS):
+            log.info("Running test on %s." % ppu)
+            self.run_ppu_program(ppu, program, int(5e5))
+
+        # Evaluate flange log
+        log_parser = self.LogParser("refgen_config_write", initial_logsize)
+        log_events = iter(log_parser)
+
+        initial_run = True
+        for _ in iter_all(PPUOnDLS):
+            # Reset by the first PPU has no effect, all values are already zero
+            if not initial_run:
+                self.assertDictEqual(next(log_events)["rg_i_amp_c"],
+                                     {'0': 0, '1': 0, '2': 15, '3': 20})
+                self.assertDictEqual(next(log_events)["rg_i_amp_c"],
+                                     {'0': 0, '1': 0, '2': 0, '3': 0})
+                self.assertDictEqual(next(log_events)["rg_i_offset_c"],
+                                     {'0': 0, '1': 0, '2': 35, '3': 40})
+                self.assertDictEqual(next(log_events)["rg_i_offset_c"],
+                                     {'0': 0, '1': 0, '2': 0, '3': 0})
+                self.assertDictEqual(next(log_events)["rg_i_slope_c"],
+                                     {'0': 0, '1': 0, '2': 55, '3': 60})
+                self.assertDictEqual(next(log_events)["rg_i_slope_c"],
+                                     {'0': 0, '1': 0, '2': 0, '3': 0})
+
+            initial_run = False
+            self.assertDictEqual(next(log_events)["rg_i_amp_c"],
+                                 {'0': 5, '1': 10, '2': 0, '3': 0})
+            self.assertDictEqual(next(log_events)["rg_i_amp_c"],
+                                 {'0': 5, '1': 10, '2': 15, '3': 20})
+            self.assertDictEqual(next(log_events)["rg_i_offset_c"],
+                                 {'0': 25, '1': 30, '2': 0, '3': 0})
+            self.assertDictEqual(next(log_events)["rg_i_offset_c"],
+                                 {'0': 25, '1': 30, '2': 35, '3': 40})
+            self.assertDictEqual(next(log_events)["rg_i_slope_c"],
+                                 {'0': 45, '1': 50, '2': 0, '3': 0})
+            self.assertDictEqual(next(log_events)["rg_i_slope_c"],
+                                 {'0': 45, '1': 50, '2': 55, '3': 60})
+
+        with self.assertRaises(StopIteration):
+            next(log_events)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/with_hostcode/refgen_access_pattern-ppu.cpp

+#include <array>
+
+#include "libnux/omnibus.h"
+#include "libnux/dls_vx.h"
+
+using namespace libnux;
+
+
+// Reference generator configuration values used within this test
+static constexpr std::array<unsigned int, 4> amp_values{5, 10, 15, 20};
+static constexpr std::array<unsigned int, 4> slope_values{25, 30, 35, 40};
+static constexpr std::array<unsigned int, 4> offset_values{45, 50, 55, 60};
+
+
+int start()
+{
+	// Zero-initialize all reference generator config registers
+	for (unsigned int i = 21; i < 27; ++i) {
+		omnibus_write(madc_base_address + i, 0);
+	}
+
+	// refgen_amp configuration
+	omnibus_write(madc_base_address + 21, (amp_values.at(1) << 6 | amp_values.at(0)));
+	omnibus_write(madc_base_address + 22, (amp_values.at(3) << 6 | amp_values.at(2)));
+
+	// refgen_slope configuration
+	omnibus_write(madc_base_address + 23, (slope_values.at(1) << 6 | slope_values.at(0)));
+	omnibus_write(madc_base_address + 24, (slope_values.at(3) << 6 | slope_values.at(2)));
+
+	// refgen_offset configuration
+	omnibus_write(madc_base_address + 25, (offset_values.at(1) << 6 | offset_values.at(0)));
+	omnibus_write(madc_base_address + 26, (offset_values.at(3) << 6 | offset_values.at(2)));
+
+	return 0;
+}

test/with_hostcode/wscript

+import os
+from os.path import join
+from waflib.extras.test_base import summary
+from waflib.extras.symwaf2ic import get_toplevel_path
+
+
+def depends(ctx):
+    ctx("haldls")
+    ctx("libnux")
+
+
+def options(opt):
+    opt.load("pytest")
+
+
+def configure(conf):
+    conf.load("compiler_cxx")
+    conf.load("python")
+    conf.check_python_version()
+    conf.check_python_headers()
+    conf.load("pytest")
+
+
+def build(bld):
+    bld.env.DLSvx_HARDWARE_AVAILABLE = "cube" == os.environ.get("SLURM_JOB_PARTITION")
+    bld.env.DLSvx_SIM_AVAILABLE = "FLANGE_SIMULATION_RCF_PORT" in os.environ
+
+    build_host_python(bld)
+    build_ppu_cpp(bld)
+
+    bld.add_post_fun(summary)
+
+
+def build_host_python(bld):
+    bld(name="libnux-simtest-log_access_pattern",
+        tests="log_access_pattern_host.py",
+        features="use pytest pylint pycodestyle",
+        use="dlens_vx",
+        install_path="${PREFIX}/bin/tests/sim",
+        pylint_config=join(get_toplevel_path(), "code-format", "pylintrc"),
+        pycodestyle_config=join(get_toplevel_path(), "code-format", "pycodestyle"),
+        skip_run=not bld.env.DLSvx_SIM_AVAILABLE,
+        test_environ=dict(
+            TEST_BINARY_PATH=os.path.join(get_toplevel_path(),
+                                          "build",
+                                          "libnux",
+                                          "test",
+                                          "with_hostcode"),
+            FLANGE_LOG_PATH=os.path.join(get_toplevel_path(),
+                                         "hxfpga",
+                                         "units",
+                                         "synplify_wrapper",
+                                         "sim",
+                                         "xrun.log")),
+        test_timeout=3600
+        )
+
+
+def build_ppu_cpp(bld):
+    bld.program(name="libnux-test-refgen_access_pattern-ppu",
+                features="cxx",
+                target="refgen_access_pattern-ppu.bin",
+                source="refgen_access_pattern-ppu.cpp",
+                use=["nux_runtime_vx"],
+                env=bld.all_envs["nux_vx"])

wscript

@@ -9,6 +9,7 @@ from waflib.extras.symwaf2ic import get_toplevel_path
 
 def depends(dep):
     dep("haldls")
+    dep("libnux", "test/with_hostcode")
 
 
 def options(opt):