test: hardware testbench

artiq/test/coredevice.py

@@ -0,0 +1,56 @@
+from artiq import *
+from artiq.test.hardware_testbench import *
+from artiq.coredevice.runtime_exceptions import RTIOUnderflow
+
+
+class RTTTest(ExperimentCase):
+    class RTT(Experiment, AutoDB):
+        class DBKeys:
+            core = Device()
+            ttl_inout = Device()
+            rtt = Result()
+
+        @kernel
+        def run(self):
+            self.ttl_inout.output()
+            delay(1*us)
+            t0 = now()
+            with parallel:
+                self.ttl_inout.gate_falling(2*us)
+                self.ttl_inout.pulse(1*us)
+            self.rtt = self.ttl_inout.timestamp() - t0
+
+    def test_rtt(self):
+        rtt = self.execute(self.RTT)["rtt"]
+        self.assertGreater(rtt, 0*ns)
+        self.assertLess(rtt, 40*ns)
+
+
+class PulseRateTest(ExperimentCase):
+    class PulseRate(Experiment, AutoDB):
+        class DBKeys:
+            core = Device()
+            loop_out = Device()
+            pulse_rate = Result()
+
+        @kernel
+        def run(self):
+            dt = time_to_cycles(1000*ns)
+            while True:
+                try:
+                    for i in range(1000):
+                        self.loop_out.pulse(cycles_to_time(dt))
+                        delay(cycles_to_time(dt))
+                except RTIOUnderflow:
+                    dt += 1
+                    self.core.break_realtime()
+                else:
+                    self.pulse_rate = cycles_to_time(2*dt)
+                    break
+
+    def test_rate(self):
+        rate = self.execute(self.PulseRate)["pulse_rate"]
+        self.assertGreater(rate, 100*ns)
+        self.assertLess(rate, 2000*ns)
+
+

artiq/test/hardware_testbench.py

@@ -0,0 +1,42 @@
+import os
+import sys
+import unittest
+import logging
+
+from artiq import *
+from artiq.protocols.file_db import FlatFileDB
+from artiq.master.worker_db import DBHub, ResultDB
+from artiq.frontend.artiq_run import (
+    DummyScheduler, DummyWatchdog, SimpleParamLogger)
+
+
+artiq_root = os.getenv("ARTIQ_ROOT")
+logger = logging.getLogger(__name__)
+
+
+@unittest.skipUnless(artiq_root, "no ARTIQ_ROOT")
+class ExperimentCase(unittest.TestCase):
+    def setUp(self):
+        self.ddb = FlatFileDB(os.path.join(artiq_root, "ddb.pyon"))
+        self.pdb = FlatFileDB(os.path.join(artiq_root, "pdb.pyon"))
+        self.rdb = ResultDB(lambda description: None, lambda mod: None)
+        self.dbh = DBHub(self.ddb, self.pdb, self.rdb)
+
+    def execute(self, cls, **kwargs):
+        expid = {
+            "file": sys.modules[cls.__module__].__file__,
+            "experiment": cls.__name__,
+            "arguments": kwargs
+        }
+        sched = DummyScheduler(expid)
+        try:
+            try:
+                exp = cls(self.dbh, scheduler=sched, **kwargs)
+            except KeyError as e:
+                raise unittest.SkipTest(*e.args)
+            self.rdb.build()
+            exp.run()
+            exp.analyze()
+            return self.rdb.data.read
+        finally:
+            self.dbh.close_devices()