artiq/artiq/coredevice/rtio.py

208 lines
5.4 KiB
Python
Raw Normal View History

2014-09-12 15:28:02 +08:00
from artiq.language.core import *
2014-10-19 23:51:49 +08:00
from artiq.coredevice.runtime_exceptions import RTIOSequenceError
2014-09-12 15:28:02 +08:00
2014-10-17 00:12:53 +08:00
class LLRTIOOut(AutoContext):
"""Low-level RTIO output driver.
Allows setting RTIO outputs at arbitrary times, without time unit
conversion and without zero-length transition suppression.
This is meant to be used mostly in drivers; consider using
``RTIOOut`` instead.
"""
parameters = "channel"
def build(self):
self.previous_timestamp = int64(0) # in RTIO cycles
self._set_oe()
@kernel
def _set_oe(self):
2014-11-21 04:49:09 +08:00
syscall("rtio_oe", self.channel, True)
2014-10-17 00:12:53 +08:00
@kernel
def set_value(self, t, value):
"""Sets the value of the RTIO channel.
:param t: timestamp in RTIO cycles (64-bit integer).
:param value: value to set at the output.
"""
if t <= self.previous_timestamp:
raise RTIOSequenceError
syscall("rtio_set", t, self.channel, value)
self.previous_timestamp = t
@kernel
def on(self, t):
"""Turns the RTIO channel on.
:param t: timestamp in RTIO cycles (64-bit integer).
"""
self.set_value(t, 1)
@kernel
def off(self, t):
"""Turns the RTIO channel off.
:param t: timestamp in RTIO cycles (64-bit integer).
"""
self.set_value(t, 0)
class _RTIOBase(AutoContext):
2014-09-12 15:28:02 +08:00
parameters = "channel"
def build(self):
self.previous_timestamp = int64(0) # in RTIO cycles
2014-09-12 15:28:02 +08:00
self.previous_value = 0
@kernel
def _set_oe(self, oe):
syscall("rtio_oe", self.channel, oe)
2014-09-12 15:28:02 +08:00
@kernel
def _set_value(self, value):
if time_to_cycles(now()) < self.previous_timestamp:
raise RTIOSequenceError
2014-09-12 15:28:02 +08:00
if self.previous_value != value:
if self.previous_timestamp == time_to_cycles(now()):
syscall("rtio_replace", time_to_cycles(now()),
self.channel, value)
2014-09-12 15:28:02 +08:00
else:
syscall("rtio_set", time_to_cycles(now()),
self.channel, value)
self.previous_timestamp = time_to_cycles(now())
2014-09-12 15:28:02 +08:00
self.previous_value = value
class RTIOOut(_RTIOBase):
"""RTIO output driver.
2014-09-30 18:10:40 +08:00
Configures the corresponding RTIO channel as output on the core device and
provides functions to set its level.
This driver supports zero-length transition suppression. For example, if
two pulses are emitted back-to-back with no delay between them, they will
be merged into a single pulse with a duration equal to the sum of the
durations of the original pulses.
:param core: core device
:param channel: channel number
"""
def build(self):
_RTIOBase.build(self)
2014-11-21 04:49:09 +08:00
self._set_oe(True)
2014-09-12 15:28:02 +08:00
@kernel
def sync(self):
"""Busy-waits until all programmed level switches have been effected.
This function is useful to synchronize CPU-controlled devices (such as
the AD9858 DDS bus) with related RTIO controls (such as RF switches at
the output of the DDS).
"""
2014-09-12 15:28:02 +08:00
syscall("rtio_sync", self.channel)
@kernel
def on(self):
"""Sets the output to a logic high state.
"""
2014-09-12 15:28:02 +08:00
self._set_value(1)
@kernel
def off(self):
"""Sets the output to a logic low state.
"""
2014-09-12 15:28:02 +08:00
self._set_value(0)
@kernel
def pulse(self, duration):
"""Pulses the output high for the specified duration.
"""
2014-09-12 15:28:02 +08:00
self.on()
delay(duration)
self.off()
2014-09-13 19:37:57 +08:00
class RTIOIn(_RTIOBase):
"""RTIO input driver.
2014-09-30 18:10:40 +08:00
Configures the corresponding RTIO channel as input on the core device and
provides functions to analyze the incoming signal, with real-time gating
to prevent overflows.
:param core: core device
:param channel: channel number
"""
def build(self):
_RTIOBase.build(self)
2014-11-21 04:49:09 +08:00
self._set_oe(False)
2014-09-13 19:37:57 +08:00
@kernel
def gate_rising(self, duration):
"""Register rising edge events for the specified duration.
"""
self._set_value(1)
delay(duration)
self._set_value(0)
@kernel
def gate_falling(self, duration):
"""Register falling edge events for the specified duration.
"""
self._set_value(2)
delay(duration)
self._set_value(0)
@kernel
def gate_both(self, duration):
"""Register both rising and falling edge events for the specified
duration.
"""
self._set_value(3)
delay(duration)
self._set_value(0)
2014-09-13 19:37:57 +08:00
2014-10-21 23:14:01 +08:00
@kernel
def pileup_count(self):
"""Returns the number of pileup events (a system clock cycle with too
many input transitions) since the last call to this function for this
channel (or since the last RTIO reset).
"""
return syscall("rtio_pileup_count", self.channel)
2014-09-13 19:37:57 +08:00
@kernel
def count(self):
"""Poll the RTIO input during all the previously programmed gate
openings, and returns the number of registered events.
"""
count = 0
while syscall("rtio_get", self.channel) >= 0:
count += 1
return count
2014-10-14 15:54:10 +08:00
@kernel
def timestamp(self):
"""Poll the RTIO input and returns an event timestamp, according to
the gating.
If the gate is permanently closed, returns a negative value.
"""
return cycles_to_time(syscall("rtio_get", self.channel))