artiq/test/full_stack.py

190 lines
5.0 KiB
Python

import unittest
from artiq.language.core import *
from artiq.language.units import *
from artiq.devices import corecom_serial, core, runtime_exceptions, rtio_core
from artiq.sim import devices as sim_devices
def _run_on_device(k_class, **parameters):
with corecom_serial.CoreCom() as com:
coredev = core.Core(com)
k_inst = k_class(core=coredev, **parameters)
k_inst.run()
def _run_on_host(k_class, **parameters):
coredev = sim_devices.Core()
k_inst = k_class(core=coredev, **parameters)
k_inst.run()
class _Primes(AutoContext):
parameters = "output_list max"
@kernel
def run(self):
for x in range(1, self.max):
d = 2
prime = True
while d*d <= x:
if x % d == 0:
prime = False
break
d += 1
if prime:
self.output_list.append(x)
class _PulseLogger(AutoContext):
parameters = "name"
def print_on(self, t, f):
print("{} ON:{:4} @{}".format(self.name, f, t))
def print_off(self, t):
print("{} OFF @{}".format(self.name, t))
@kernel
def pulse(self, f, duration):
self.print_on(int(now()), f)
delay(duration)
self.print_off(int(now()))
class _Pulses(AutoContext):
def build(self):
for name in "a", "b", "c", "d":
pl = _PulseLogger(self, name=name)
setattr(self, name, pl)
@kernel
def run(self):
for i in range(3):
with parallel:
with sequential:
self.a.pulse(100+i, 20*us)
self.b.pulse(200+i, 20*us)
with sequential:
self.c.pulse(300+i, 10*us)
self.d.pulse(400+i, 20*us)
class _MyException(Exception):
pass
class _Exceptions(AutoContext):
parameters = "trace"
@kernel
def run(self):
for i in range(10):
self.trace.append(i)
if i == 4:
try:
self.trace.append(10)
try:
self.trace.append(11)
break
except:
pass
else:
self.trace.append(12)
try:
self.trace.append(13)
except:
pass
except _MyException:
self.trace.append(14)
for i in range(4):
try:
self.trace.append(100)
if i == 1:
raise _MyException
elif i == 2:
raise IndexError
except (TypeError, IndexError):
self.trace.append(101)
raise
except:
self.trace.append(102)
else:
self.trace.append(103)
finally:
self.trace.append(104)
class SimCompareCase(unittest.TestCase):
def test_primes(self):
l_device, l_host = [], []
_run_on_device(_Primes, max=100, output_list=l_device)
_run_on_host(_Primes, max=100, output_list=l_host)
self.assertEqual(l_device, l_host)
def test_pulses(self):
# TODO: compare results on host and device
# (this requires better unit management in the compiler)
_run_on_device(_Pulses)
def test_exceptions(self):
t_device, t_host = [], []
with self.assertRaises(IndexError):
_run_on_device(_Exceptions, trace=t_device)
with self.assertRaises(IndexError):
_run_on_host(_Exceptions, trace=t_host)
self.assertEqual(t_device, t_host)
class _RTIOLoopback(AutoContext):
parameters = "i o npulses"
def report(self, n):
self.result = n
@kernel
def run(self):
with parallel:
with sequential:
for i in range(self.npulses):
delay(25*ns)
self.o.pulse(25*ns)
self.i.gate_rising(10*us)
self.report(self.i.count())
class _RTIOUnderflow(AutoContext):
parameters = "o"
@kernel
def run(self):
while True:
delay(25*ns)
self.o.pulse(25*ns)
class RTIOCase(unittest.TestCase):
def test_loopback(self):
npulses = 4
with corecom_serial.CoreCom() as com:
coredev = core.Core(com)
uut = _RTIOLoopback(
core=coredev,
i=rtio_core.RTIOIn(core=coredev, channel=0),
o=rtio_core.RTIOOut(core=coredev, channel=1),
npulses=npulses
)
uut.run()
self.assertEqual(uut.result, npulses)
def test_underflow(self):
with corecom_serial.CoreCom() as com:
coredev = core.Core(com)
uut = _RTIOUnderflow(
core=coredev,
o=rtio_core.RTIOOut(core=coredev, channel=1)
)
with self.assertRaises(runtime_exceptions.RTIOUnderflow):
uut.run()