Source code for atomiq.components.sinara.kasli

from __future__ import annotations

from atomiq.components.primitives import Component, Switchable, Measurable

from artiq.experiment import kernel, at_mu, now_mu, delay, us
from artiq.language.types import TFloat, TStr, TBool
import numpy as np




class Kasli(Component):
    """Kasli Module"""
    def __init__(self, *args, **kwargs):
        Component.__init__(self, *args, **kwargs)





class DioOutput(Component, Switchable):
    """Fast Digital Output Channel Attached to Kasli

    This class describes the fast digital output channels directly attached to your Kasli.

    Args:
        kasli: Kasli module that this digital output belongs to
        ttl: ARTIQ device from `device_db.py` for the DIO channel
    """

    kernel_invariants = {"kasli", "ttl"}

    def __init__(self, kasli: Kasli, ttl, *args, **kwargs):
        Component.__init__(self, *args, **kwargs)
        Switchable.__init__(self, ['TTL'])
        self.kasli = kasli
        self.ttl = ttl



    @kernel
    def on(self):
        self.ttl.on()




    @kernel
    def off(self):
        self.ttl.off()






class DioInput(Component, Measurable):
    """Fast Digital Input Channel Attached to Kasli

    This class describes the fast digital input channels directly attached to your Kasli.

    Args:
        kasli: Kasli module that this digital input belongs to
        ttl: ARTIQ device from `device_db.py` for the DIO channel
    """

    kernel_invariants = {"kasli", "ttl"}


    def __init__(self, kasli: Kasli, ttl, *args, **kwargs):
        Component.__init__(self, *args, **kwargs)
        Measurable.__init__(self, ['TTL'])
        self.kasli = kasli
        self.ttl = ttl


    @kernel
    def measure(self):
        return self.ttl.sample_input()




    @kernel
    def wait_for_trigger(
            self,
            timeout: TFloat,
            edge: TStr = "rising",
            log: TBool = False,
            keep_open: TBool = False
        ) -> TBool:
        """Wait for a trigger to occur. This function sets the time cursor to the time where the trigger edge is
        detected. If no edge is detected in the given timeout, the time cursor is set to the end of the gating window
        (set by `timeout`).

        Args:
            timeout: Timeout in seconds
            edge: Edge to wait for (rising or falling)
            log: If True emits a warning if no trigger is detected
            keep_open: True keeps the gate open after the trigger window. This can be used to rapidly trigger on
                subsequent pulses on the microsecond timescale. Note that the trigger listener must be closed manually
                afterwards by either running this function with `keep_open=True` or by calling
                `self.ttl._set_sensitivity(0)` to avoid kernel slowdown.

        Returns:
            bool: True if a trigger is detected, False if not

        Info:
            Typically a delay of about 2 us is necessary after this command to avoid underflows.

        """
        # clear previous events
        self.ttl._set_sensitivity(0)
        while self.ttl.timestamp_mu(now_mu()) >= 0:
            pass
        delay(2*us) # necessary to avoid underflows
        if edge == "rising":
            self.ttl._set_sensitivity(1)
        else:
            self.ttl._set_sensitivity(2)
        delay(timeout)
        t_edge = self.ttl.timestamp_mu(now_mu())
        if not keep_open:
            self.ttl._set_sensitivity(0)
        if t_edge > 0:
            at_mu(t_edge)
            return True
        else:
            if log:
                self.experiment.log.warning("No edge detected @ ttl {0}", [self.identifier,])
            return False