forked from M-Labs/artiq
284 lines
7.9 KiB
Python
284 lines
7.9 KiB
Python
import unittest
|
|
from operator import itemgetter
|
|
import os
|
|
from fractions import Fraction
|
|
|
|
from artiq import *
|
|
from artiq.language.units import *
|
|
from artiq.coredevice import comm_serial, core, runtime_exceptions, rtio
|
|
from artiq.sim import devices as sim_devices
|
|
|
|
|
|
no_hardware = bool(os.getenv("ARTIQ_NO_HARDWARE"))
|
|
|
|
|
|
def _run_on_device(k_class, **parameters):
|
|
with comm_serial.Comm() as comm:
|
|
coredev = core.Core(comm)
|
|
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 _Misc(AutoContext):
|
|
def build(self):
|
|
self.input = 84
|
|
self.inhomogeneous_units = []
|
|
|
|
@kernel
|
|
def run(self):
|
|
self.half_input = self.input//2
|
|
decimal_fraction = Fraction("1.2")
|
|
self.decimal_fraction_n = int(decimal_fraction.numerator)
|
|
self.decimal_fraction_d = int(decimal_fraction.denominator)
|
|
self.inhomogeneous_units.append(Quantity(1000, "Hz"))
|
|
self.inhomogeneous_units.append(Quantity(10, "s"))
|
|
|
|
@kernel
|
|
def dimension_error1(self):
|
|
print(1*Hz + 1*s)
|
|
|
|
@kernel
|
|
def dimension_error2(self):
|
|
print(1*Hz < 1*s)
|
|
|
|
@kernel
|
|
def dimension_error3(self):
|
|
check_unit(1*Hz, "s")
|
|
|
|
@kernel
|
|
def dimension_error4(self):
|
|
delay(10*Hz)
|
|
|
|
|
|
class _PulseLogger(AutoContext):
|
|
parameters = "output_list name"
|
|
|
|
def _append(self, t, l, f):
|
|
if not hasattr(self, "first_timestamp"):
|
|
self.first_timestamp = t
|
|
self.output_list.append((self.name, t-self.first_timestamp, l, f))
|
|
|
|
def on(self, t, f):
|
|
self._append(t, True, f)
|
|
|
|
def off(self, t):
|
|
self._append(t, False, 0)
|
|
|
|
@kernel
|
|
def pulse(self, f, duration):
|
|
self.on(int(now().amount*1000000000), f)
|
|
delay(duration)
|
|
self.off(int(now().amount*1000000000))
|
|
|
|
|
|
class _Pulses(AutoContext):
|
|
parameters = "output_list"
|
|
|
|
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)
|
|
|
|
|
|
@unittest.skipIf(no_hardware, "no hardware")
|
|
class ExecutionCase(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_misc(self):
|
|
with comm_serial.Comm() as comm:
|
|
coredev = core.Core(comm)
|
|
uut = _Misc(core=coredev)
|
|
uut.run()
|
|
self.assertEqual(uut.half_input, 42)
|
|
self.assertEqual(Fraction(uut.decimal_fraction_n,
|
|
uut.decimal_fraction_d),
|
|
Fraction("1.2"))
|
|
self.assertEqual(uut.inhomogeneous_units, [
|
|
Quantity(1000, "Hz"), Quantity(10, "s")])
|
|
with self.assertRaises(DimensionError):
|
|
uut.dimension_error1()
|
|
with self.assertRaises(DimensionError):
|
|
uut.dimension_error2()
|
|
with self.assertRaises(DimensionError):
|
|
uut.dimension_error3()
|
|
with self.assertRaises(DimensionError):
|
|
uut.dimension_error4()
|
|
|
|
def test_pulses(self):
|
|
l_device, l_host = [], []
|
|
_run_on_device(_Pulses, output_list=l_device)
|
|
_run_on_host(_Pulses, output_list=l_host)
|
|
l_host = sorted(l_host, key=itemgetter(1))
|
|
for channel in "a", "b", "c", "d":
|
|
c_device = [x for x in l_device if x[0] == channel]
|
|
c_host = [x for x in l_host if x[0] == channel]
|
|
self.assertEqual(c_device, c_host)
|
|
|
|
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 _RTIOSequenceError(AutoContext):
|
|
parameters = "o"
|
|
|
|
@kernel
|
|
def run(self):
|
|
t = now()
|
|
self.o.pulse(25*us)
|
|
at(t)
|
|
self.o.pulse(25*us)
|
|
|
|
|
|
@unittest.skipIf(no_hardware, "no hardware")
|
|
class RTIOCase(unittest.TestCase):
|
|
# Connect channels 0 and 1 together for this test
|
|
# (C11 and C13 on Papilio Pro)
|
|
def test_loopback(self):
|
|
npulses = 4
|
|
with comm_serial.Comm() as comm:
|
|
coredev = core.Core(comm)
|
|
uut = _RTIOLoopback(
|
|
core=coredev,
|
|
i=rtio.RTIOIn(core=coredev, channel=0),
|
|
o=rtio.RTIOOut(core=coredev, channel=1),
|
|
npulses=npulses
|
|
)
|
|
uut.run()
|
|
self.assertEqual(uut.result, npulses)
|
|
|
|
def test_underflow(self):
|
|
with comm_serial.Comm() as comm:
|
|
coredev = core.Core(comm)
|
|
uut = _RTIOUnderflow(
|
|
core=coredev,
|
|
o=rtio.RTIOOut(core=coredev, channel=1)
|
|
)
|
|
with self.assertRaises(runtime_exceptions.RTIOUnderflow):
|
|
uut.run()
|
|
|
|
def test_sequence_error(self):
|
|
with comm_serial.Comm() as comm:
|
|
coredev = core.Core(comm)
|
|
uut = _RTIOSequenceError(
|
|
core=coredev,
|
|
o=rtio.RTIOOut(core=coredev, channel=1)
|
|
)
|
|
with self.assertRaises(runtime_exceptions.RTIOSequenceError):
|
|
uut.run()
|