Module dqcsim.plugin

Contains the base classes for implementing DQCsim plugins.

You should be implementing one of Frontend, Operator, or Backend. The documentation of these classes indicates which functions you should implement to specify the plugin's functionality. Overriding __init__() is also allowed, but if you do this you must call super().__init__().

A completed plugin script looks something like this:

from dqcsim.plugin import Frontend
from dqcsim.common.arb import ArbData

class MyPlugin(Frontend):
    def get_name(self):
        return "My Plugin"

    def get_author(self):
        return "Me!"

    def get_version(self):
        return "3.14"

    def handle_run(self, *args, **kwargs):
        # Just return whatever argument we received as the return value!
        return ArbData(*args, **kwargs)

MyPlugin().run()

This allows the script to be called using DQCsim's command-line interface by simply specifying the Python script.

You can use the @plugin decorator to make plugin classes a little shorter. It automatically generates the metadata getters for you.

@plugin("My Plugin", "Me!", "3.14")
class MyPlugin(Frontend):
    def handle_run(self, *args, **kwargs):
        ...

"""Contains the base classes for implementing DQCsim plugins.

You should be implementing one of `Frontend`, `Operator`, or `Backend`. The
documentation of these classes indicates which functions you should implement
to specify the plugin's functionality. Overriding `__init__()` is also allowed,
but if you do this you must call `super().__init__()`.

A completed plugin script looks something like this:

    from dqcsim.plugin import Frontend
    from dqcsim.common.arb import ArbData

    class MyPlugin(Frontend):
        def get_name(self):
            return "My Plugin"

        def get_author(self):
            return "Me!"

        def get_version(self):
            return "3.14"

        def handle_run(self, *args, **kwargs):
            # Just return whatever argument we received as the return value!
            return ArbData(*args, **kwargs)

    MyPlugin().run()

This allows the script to be called using DQCsim's command-line interface by
simply specifying the Python script.

You can use the `@plugin` decorator to make plugin classes a little shorter. It
automatically generates the metadata getters for you.

    @plugin("My Plugin", "Me!", "3.14")
    class MyPlugin(Frontend):
        def handle_run(self, *args, **kwargs):
            ...
"""

__all__ = [ #@
    'Frontend',
    'Operator',
    'Backend',
    'GateStreamSource',
    'Plugin',
    'JoinHandle',
    'plugin'
]
__pdoc__ = { #@
    'JoinHandle.__init__': False
}

import dqcsim._dqcsim as raw
from dqcsim.common import *
import sys
import inspect
import traceback
import math, cmath

class JoinHandle(object):
    """Returned by `Plugin.start()` to allow waiting for completion."""
    def __init__(self, handle):
        super().__init__()
        if raw.dqcs_handle_type(int(handle)) != raw.DQCS_HTYPE_PLUGIN_JOIN:
            raise TypeError("Specified handle is not a JoinHandle")
        self._handle = handle

    def wait(self):
        """Waits for the associated plugin to finish executing."""
        if self._handle:
            raw.dqcs_plugin_wait(int(self._handle))
            self._handle.take()

class plugin(object):
    """Decorator for Plugin class implementations to take some of the
    boilerplate code away."""

    def __init__(self, name, author, version):
        super().__init__()
        self._name = name
        self._author = author
        self._version = version

    def __call__(self, cls):
        setattr(cls, "get_name", lambda _: self._name)
        setattr(cls, "get_author", lambda _: self._author)
        setattr(cls, "get_version", lambda _: self._version)
        return cls

class Plugin(object):
    """Represents a plugin implementation. Must be subclassed; use Frontend,
    Operator, or Backend instead."""

    #==========================================================================
    # Launching the plugin
    #==========================================================================
    def __init__(self, host_arb_ifaces=None, upstream_arb_ifaces=None):
        """Creates the plugin object.

        Overriding `__init__()` in your implementation is fine, but you must
        call `super().__init__()` if you do this.

        Among other things, this function auto-detects which arb interfaces are
        supported by your plugin by scanning for `handle_host_*()` and
        `handle_upstream_*()` implementations. However, this auto-detection
        will fail if there are underscores in any of the supported interface or
        operation IDs. In this case, you MUST override `__init__()` and call
        `super().__init__(host_arb_ifaces, upstream_arb_ifaces)`, where the two
        arguments are lists of strings representing the supported interface
        IDs.
        """
        super().__init__()

        # CPython's trace functionality is used by kcov to test Python
        # coverage. That's great and all, but when we call into the C API and
        # the C API calls back into us (from another thread, no less), we're in
        # a new Python context that doesn't have the trace function set. We
        # need to fix this manually when we get such a callback. This is done
        # by the functions generated by `self._cbent()` if this is not `None`.
        # If no trace function is specified, there should be no runtime
        # overhead.
        self._trace_fn = sys.gettrace()

        # This local is used to store whether we're inside a user-defined
        # callback and, if so, what the value of the `dqcs_plugin_type_t`
        # handle is. This allows us to A) check that the user isn't calling
        # `dqcs_plugin_*` functions taking such a state without having a valid
        # one, and B) abstract the requirement of passing the plugin handle
        # around away from the user. Note that there are no thread-safety
        # problems with this, since the C API plugin callbacks are all called
        # from the same thread.
        self._state_handle = None

        # Stores whether this plugin has been instantiated. This can only be
        # done once, otherwise we could get plugin callbacks from multiple
        # plugin instances/threads, which would mess up the "thread-safety"
        # of `_state_handle`.
        self._started = False

        # Configure the supported arb interfaces.
        self._arb_interfaces = {}
        if host_arb_ifaces is not None:
            self._arb_interfaces['host'] = set(host_arb_ifaces)
        if upstream_arb_ifaces is not None:
            self._arb_interfaces['upstream'] = set(upstream_arb_ifaces)
        for source in ['host', 'upstream']:
            if source in self._arb_interfaces:
                continue

            # Try to auto-detect arb interfaces for this source.
            ifaces = set()
            for mem, _ in inspect.getmembers(self, predicate=inspect.ismethod):
                if not mem.startswith('handle_{}_'.format(source)):
                    continue
                s = mem.split('_')
                if len(s) > 4:
                    raise RuntimeError(
                        "cannot auto-detect interface and operation ID for arb "
                        "handler {}(), please pass the '{}_arb_ifaces' argument "
                        "to __init__() to specify the supported interfaces manually"
                        "".format(mem, source))
                elif len(s) < 4:
                    continue
                ifaces.add(s[2])
            self._arb_interfaces[source] = ifaces

    def _check_run(self, simulator):
        """Checks that the plugin is ready to be started and figures out the
        simulator address."""
        if not hasattr(self, '_state_handle'):
            raise RuntimeError("It looks like you've overridden __init__ and forgot to call super().__init__(). Please fix!")
        if self._started:
            raise RuntimeError("Plugin has been started before. Make a new instance!")
        if simulator is None:
            if len(sys.argv) != 2:
                print("Usage: [python3] <script> <simulator-address>", file=sys.stderr)
                print("Note: you should be calling this Python script with DQCsim!", file=sys.stderr)
                sys.exit(1)
            simulator = sys.argv[1]
        return simulator

    def run(self, simulator=None):
        """Instantiates and runs the plugin.

        simulator represents the DQCsim address that the plugin must connect to
        when initializing. It is usually passed as the first argument to the
        plugin process; therefore, if it is not specified, it is taken directly
        from sys.argv."""
        simulator = self._check_run(simulator)
        with self._to_pdef() as pdef:
            self._started = True
            raw.dqcs_plugin_run(pdef, simulator)

    def start(self, simulator=None):
        """Instantiates and starts the plugin.

        This has the same behavior as run(), except the plugin is started in a
        different thread, so it returns immediately. The returned object is a
        JoinHandle, which contains a wait() method that can be used to wait
        until the plugin finishes executing. Alternatively, if this is not
        done, the plugin thread will (try to) survive past even the main
        thread.

        Note that the JoinHandle can NOT be transferred to a different
        thread!"""
        simulator = self._check_run(simulator)
        with self._to_pdef() as pdef:
            handle = Handle(raw.dqcs_plugin_start(pdef, simulator))
        self._started = True
        return JoinHandle(handle)

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def _pc(self, plugin_fn, *args):
        """Use this to call dqcs_plugin functions that take a plugin state."""
        if self._state_handle is None:
            raise RuntimeError("Cannot call plugin operator outside of a callback")
        return plugin_fn(self._state_handle, *args)

    def random_float(self):
        """Produces a random floating point value between 0 (inclusive) and 1
        (exclusive).

        This function is guaranteed to return the same result every time as
        long as the random seed allocated to us by DQCsim stays the same. This
        allows simulations to be reproduced using a reproduction file. Without
        such a reproduction file or user-set seed, this is of course properly
        (pseudo)randomized."""
        return self._pc(raw.dqcs_plugin_random_f64)

    def random_long(self):
        """Produces a random 64-bit unsigned integer.

        This function is guaranteed to return the same result every time as
        long as the random seed allocated to us by DQCsim stays the same. This
        allows simulations to be reproduced using a reproduction file. Without
        such a reproduction file or user-set seed, this is of course properly
        (pseudo)randomized."""
        return self._pc(raw.dqcs_plugin_random_u64)

    #==========================================================================
    # Logging functions
    #==========================================================================
    def _log(self, level, msg, *args, **kwargs):
        # NOTE: we don't need the state handle technically, but this ensures
        # that we're in the right thread.
        if self._state_handle is None:
            raise RuntimeError("Cannot call plugin operator outside of a callback")
        msg = str(msg)
        if args or kwargs:
            msg = msg.format(*args, **kwargs)
        frame = inspect.currentframe().f_back.f_back
        module = frame.f_globals.get('__name__', '?')
        fname = frame.f_globals.get('__file__', '?')
        lineno = frame.f_lineno
        raw.dqcs_log_raw(level, module, fname, lineno, msg)

    def log(self, level, msg, *args, **kwargs):
        """Logs a message with the specified loglevel to DQCsim.

        If any additional positional or keyword arguments are specified, the
        message is formatted using `str.format()`. Otherwise, `str()` is
        applied to the message."""
        # NOTE: this level of indirection is needed to make function name,
        # filename, and line number metadata correct.
        if not isinstance(level, Loglevel):
            raise TypeError('level must be a Loglevel')
        self._log(level, msg, *args, **kwargs)

    def trace(self, msg, *args, **kwargs):
        """Convenience function for logging trace messages. See `log()`."""
        self._log(Loglevel.TRACE, msg, *args, **kwargs)

    def debug(self, msg, *args, **kwargs):
        """Convenience function for logging debug messages. See `log()`."""
        self._log(Loglevel.DEBUG, msg, *args, **kwargs)

    def info(self, msg, *args, **kwargs):
        """Convenience function for logging info messages. See `log()`."""
        self._log(Loglevel.INFO, msg, *args, **kwargs)

    def note(self, msg, *args, **kwargs):
        """Convenience function for logging note messages. See `log()`."""
        self._log(Loglevel.NOTE, msg, *args, **kwargs)

    def warn(self, msg, *args, **kwargs):
        """Convenience function for logging warning messages. See `log()`."""
        self._log(Loglevel.WARN, msg, *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        """Convenience function for logging error messages. See `log()`."""
        self._log(Loglevel.ERROR, msg, *args, **kwargs)

    def fatal(self, msg, *args, **kwargs):
        """Convenience function for logging fatal messages. See `log()`."""
        self._log(Loglevel.FATAL, msg, *args, **kwargs)

    warning = warn
    critical = fatal

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _cbent(self, router):
        """All callbacks from the C API world are generated by this function.
        Normally it just returns a "router" function, which determines which
        user callback(s) should be called for the given C API callback. If a
        trace function is set however (this is done by kcov for Python code
        coverage), we need to set this trace function every time we're called
        from the C domain for it to work. In that case, a proxy function is
        returned that first sets the trace function and then calls the actual
        router."""
        router_fn = getattr(self, '_route_' + router)
        if self._trace_fn:
            def traced_router_fn(*args, **kwargs):
                sys.settrace(self._trace_fn) # no_kcoverage
                return router_fn(*args, **kwargs) # no_kcoverage
            return traced_router_fn
        return router_fn # no_kcoverage

    def _cb(self, state_handle, name, *args, **kwargs):
        """This function is used to call into the Python callbacks specified by
        the user. It saves the state handle in a local variable so the user
        code doesn't have to carry it around everywhere (and can't accidentally
        break it), then calls the named function is it exists. It also logs the
        user's stack traces with trace loglevel for debugging; if it wouldn't,
        this information would be lost (only the str() representation of the
        exception is passed on to the C API layer). If the user did not
        implement the callback, a NotImplementedError is returned, which may or
        may not be handled by the router function if there is an alternative
        call or a sane default operation."""
        if hasattr(self, name):
            if self._state_handle is not None:
                raise RuntimeError("Invalid state, recursive callback")
            self._state_handle = state_handle
            try:
                try:
                    return getattr(self, name)(*args, **kwargs)
                except Exception as e:
                    for line in traceback.format_exc().split('\n'):
                        self.trace(line)
                    raise
            finally:
                self._state_handle = None
        raise NotImplementedError("Python plugin doesn't implement {}(), which is a required function!".format(name))

    def _route_initialize(self, state_handle, init_cmds_handle):
        """Routes the initialization callback to the user's implementation, if
        there is any. If there isn't, try to route the initialization `ArbCmd`s
        as if they're normal host arbs."""
        cmds = ArbCmdQueue._from_raw(Handle(init_cmds_handle))
        try:
            self._cb(state_handle, 'handle_init', cmds)
        except NotImplementedError:
            for cmd in cmds:
                self._route_converted_arb(state_handle, 'host', cmd)

    def _route_drop(self, state_handle):
        """Routes the drop callback to the user's implementation, if there is
        any."""
        try:
            self._cb(state_handle, 'handle_drop')
        except NotImplementedError:
            pass

    def _route_converted_arb(self, state_handle, source, cmd, forward_fn=None):
        """Routes an `ArbCmd` originating from the given source, which must be
        'upstream' or 'host'. `cmd` should already have been converted to an
        `ArbCmd` object (instead of being passed as a handle)."""
        if cmd.iface not in self._arb_interfaces.get(source, {}):
            if forward_fn is not None:
                return self._cb(state_handle, forward_fn, cmd)
            return ArbData()
        try:
            result = self._cb(state_handle,
                'handle_{}_{}_{}'.format(source, cmd.iface, cmd.oper),
                *cmd._args, **cmd._json
            ) #@
            if result is None:
                result = ArbData()
            elif not isinstance(result, ArbData):
                raise TypeError("User implementation of host arb should return None or ArbData but returned {}".format(type(result)))
            return result
        except NotImplementedError:
            raise ValueError("Invalid operation ID {} for interface ID {}".format(cmd.oper, cmd.iface))

    def _route_arb(self, state_handle, source, cmd_handle, forward_fn=None):
        """Routes an `ArbCmd` originating from the given source, which must be
        'upstream' or 'host'. Takes an integer handle to an `ArbCmd` and
        returns an integer handle to an `ArbData` (that is, they're not wrapped
        in `Handle` objects)."""
        cmd = ArbCmd._from_raw(Handle(cmd_handle))
        result = self._route_converted_arb(state_handle, source, cmd, forward_fn)
        return result._to_raw().take()

    def _route_host_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the host."""
        return self._route_arb(state_handle, 'host', cmd_handle)

    def _new_pdef(self, typ):
        """Constructs a pdef `Handle` configured with appropriate metadata and
        the callbacks common to all plugins."""
        pdef = Handle(raw.dqcs_pdef_new(
            typ,
            self._cb(None, 'get_name'),
            self._cb(None, 'get_author'),
            self._cb(None, 'get_version')
        )) #@
        with pdef as pd:
            raw.dqcs_pdef_set_initialize_cb_pyfun(pd, self._cbent('initialize'))
            raw.dqcs_pdef_set_drop_cb_pyfun(pd, self._cbent('drop'))
            raw.dqcs_pdef_set_host_arb_cb_pyfun(pd, self._cbent('host_arb'))
        return pdef

class GateStreamSource(Plugin):
    """Adds gatestream source functions."""

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def allocate(self, num_qubits=None, *cmds):
        """Instructs the downstream plugin to allocate one or more qubits.

        If `num_qubits` is specified, this function returns a list of qubit
        references that you can use to refer to the qubits in later function
        calls. These are just integers. If `num_qubits` is not specified or
        `None`, a single qubit is allocated and returned without being wrapped
        in a list.

        Optionally, you can pass (a list of) ArbCmd objects to associate with
        the qubits."""
        with ArbCmdQueue._to_raw(*cmds) as cmds:
            qubits = QubitSet._from_raw(Handle(self._pc(
                raw.dqcs_plugin_allocate,
                1 if num_qubits is None else num_qubits,
                cmds
            ))) #@
        if num_qubits is None:
            return qubits[0]
        else:
            return qubits

    def free(self, *qubits):
        """Instructs the downstream plugin to free the given qubits."""
        with QubitSet._to_raw(*qubits) as qubits:
            self._pc(raw.dqcs_plugin_free, qubits)

    def unitary(self, targets, matrix, controls=[], arb=None):
        """Instructs the downstream plugin to execute a unitary quantum gate.

        `targets` must be a non-empty iterable of qubits or a single qubit,
        representing the qubit(s) targeted by the gate. `matrix` must be a
        unitary matrix appropriately sized for the number of target qubits,
        specified as a row-major one-dimensional list of Python complex
        numbers. `controls` optionally allows additional control qubits to be
        specified to make controlled gates; these qubits should NOT be
        reflected in the gate matrix. The matrix will automatically be extended
        by the downstream plugin, instead. The `targets` and `controls` sets
        must not intersect.
        """
        with QubitSet._to_raw(targets) as targets:
            with QubitSet._to_raw(controls) as controls:
                with Handle(raw.dqcs_mat_new(matrix)) as mat:
                    gate = Handle(raw.dqcs_gate_new_unitary(targets, controls, mat))
                    if arb is not None:
                        if not isinstance(arb, ArbData):
                            raise TypeError('arb must be None or an instance of ArbData')
                        arb._to_raw(gate)
                    with gate as gate_raw:
                        self._pc(raw.dqcs_plugin_gate, gate_raw)

    def i_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute an I gate.

        `target` is the targetted qubit."""
        self.unitary(target, [1.0, 0.0, 0.0, 1.0], arb=arb)

    def rx_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary X rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = math.cos(0.5 * theta)
        b = -1.0j * math.sin(0.5 * theta)
        self.unitary(target, [a, b, b, a], arb=arb)

    def ry_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary Y rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = math.cos(0.5 * theta)
        b = math.sin(0.5 * theta)
        self.unitary(target, [a, -b, b, a], arb=arb)

    def rz_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary Z rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = cmath.exp(-0.5j * theta)
        b = cmath.exp(0.5j * theta)
        self.unitary(target, [a, 0.0, 0.0, b], arb=arb)

    def r_gate(self, target, theta, phi, lambd, arb=None):
        """Instructs the downstream plugin to perform a number of rotations at
        once.

        `target` is the targetted qubit. `theta`, `phi`, and `lambd` are the
        angles in radians."""
        a = math.cos(0.5 * theta)
        b = math.sin(0.5 * theta)
        self.unitary(target, [
            a,
            -b * cmath.exp(1.0j * lambd),
            b * cmath.exp(1.0j * phi),
            a * cmath.exp(1.0j * (phi + lambd)),
        ], arb=arb)

    def swap_gate(self, a, b, arb=None):
        """Instructs the downstream plugin to execute a swap gate.

        `a` and `b` are the targetted qubits."""
        self.unitary([a, b], [
            1.0, 0.0, 0.0, 0.0,
            0.0, 0.0, 1.0, 0.0,
            0.0, 1.0, 0.0, 0.0,
            0.0, 0.0, 0.0, 1.0,
        ], arb=arb) #@

    def sqswap_gate(self, a, b, arb=None):
        """Instructs the downstream plugin to execute a square-root-of-swap
        gate.

        `a` and `b` are the targetted qubits."""
        self.unitary([a, b], [
            1.0, 0.0,      0.0,      0.0,
            0.0, 0.5+0.5j, 0.5-0.5j, 0.0,
            0.0, 0.5-0.5j, 0.5+0.5j, 0.0,
            0.0, 0.0,      0.0,      1.0,
        ], arb=arb) #@

    def x_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute an X gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, math.pi, arb=arb)

    def x90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree X gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, 0.5 * math.pi, arb=arb)

    def mx90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree X
        gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, -0.5 * math.pi, arb=arb)

    def y_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Y gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, math.pi, arb=arb)

    def y90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree Y gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, 0.5 * math.pi, arb=arb)

    def my90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree Y
        gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, -0.5 * math.pi, arb=arb)

    def z_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Z gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, math.pi, arb=arb)

    def z90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree Z gate, also
        known as an S gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, 0.5 * math.pi, arb=arb)

    def mz90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree Z
        gate, also known as an S-dagger gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, -0.5 * math.pi, arb=arb)

    s_gate = z90_gate
    sdag_gate = mz90_gate

    def t_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a T gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, 0.25 * math.pi, arb=arb)

    def tdag_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a T-dagger gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, -0.25 * math.pi, arb=arb)

    def h_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Hadamard gate.

        `target` is the targetted qubit."""
        x = 1 / math.sqrt(2.0)
        self.unitary([target], [x, x, x, -x], arb=arb)

    def cnot_gate(self, control, target, arb=None):
        """Instructs the downstream plugin to execute a CNOT gate."""
        self.unitary([target], [
            0.0, 1.0,
            1.0, 0.0,
        ], controls=[control], arb=arb) #@

    def toffoli_gate(self, c1, c2, target, arb=None):
        """Instructs the downstream plugin to execute a Toffoli gate."""
        self.unitary([target], [
            0.0, 1.0,
            1.0, 0.0,
        ], controls=[c1, c2], arb=arb) #@

    def fredkin_gate(self, control, a, b, arb=None):
        """Instructs the downstream plugin to execute a Fredkin gate."""
        self.unitary([a, b], [
            1.0, 0.0, 0.0, 0.0,
            0.0, 0.0, 1.0, 0.0,
            0.0, 1.0, 0.0, 0.0,
            0.0, 0.0, 0.0, 1.0,
        ], controls=[control], arb=arb) #@

    def measure(self, *qubits, basis='Z', arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        given basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only positional
        argument. The basis is always a keyword argument.

        The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
        numbers representing a 2x2 matrix with the following semantics:

         - the qubits are rotated by the inverse of the matrix
         - a Z-axis measurement is performed on each qubit
         - the qubits are rotated by the matrix
        """
        if isinstance(basis, str):
            basis = Handle(raw.dqcs_mat_basis({
                'X': raw.DQCS_BASIS_X,
                'Y': raw.DQCS_BASIS_Y,
                'Z': raw.DQCS_BASIS_Z,
            }[basis]))
        else:
            basis = Handle(raw.dqcs_mat_new(basis))
        if len(qubits) == 1 and not isinstance(qubits[0], int):
            qubits = list(qubits[0])
        with QubitSet._to_raw(qubits) as qubits:
            with basis as mat:
                gate = Handle(raw.dqcs_gate_new_measurement(qubits, mat))
                if arb is not None:
                    if not isinstance(arb, ArbData):
                        raise TypeError('arb must be None or an instance of ArbData')
                    arb._to_raw(gate)
                with gate as gate_raw:
                    self._pc(raw.dqcs_plugin_gate, gate_raw)

    def measure_x(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        X basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='X', arb=arb)

    def measure_y(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        Y basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='Y', arb=arb)

    def measure_z(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        Z basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='Z', arb=arb)

    def prepare(self, *qubits, basis='Z', arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the given basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only positional
        argument. The basis is always a keyword argument.

        The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
        numbers representing a 2x2 matrix with the following semantics:

         - the qubits are initialized to |0>
         - the qubits are rotated by the matrix
        """
        if isinstance(basis, str):
            basis = Handle(raw.dqcs_mat_basis({
                'X': raw.DQCS_BASIS_X,
                'Y': raw.DQCS_BASIS_Y,
                'Z': raw.DQCS_BASIS_Z,
            }[basis]))
        else:
            basis = Handle(raw.dqcs_mat_new(basis))
        if len(qubits) == 1 and not isinstance(qubits[0], int):
            qubits = list(qubits[0])
        with QubitSet._to_raw(qubits) as qubits:
            with basis as mat:
                gate = Handle(raw.dqcs_gate_new_prep(qubits, mat))
                if arb is not None:
                    if not isinstance(arb, ArbData):
                        raise TypeError('arb must be None or an instance of ArbData')
                    arb._to_raw(gate)
                with gate as gate_raw:
                    self._pc(raw.dqcs_plugin_gate, gate_raw)

    def prepare_x(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the X basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='X', arb=arb)

    def prepare_y(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the Y basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='Y', arb=arb)

    def prepare_z(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the Z basis, being |0>.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='Z', arb=arb)

    def custom_gate(self, name, targets=[], controls=[], measures=[], matrix=None, *args, **kwargs):
        """Instructs the downstream plugin to execute a custom gate.

        `name` must be a non-empty string identifying the gate to be performed.
        `targets`, `constrols`, and `measures` must be iterables of qubits or
        singular qubits, representing respectively the set of qubits to operate
        on, the set of control qubits for controlled gates, and the set of
        qubits measured by the gate.  The `targets` and `controls` sets must
        not intersect. If specified, `matrix` must be a unitary matrix
        appropriately sized for the number of target qubits, specified as a
        row-major one-dimensional list of Python complex numbers. If no matrix
        is applicable or necessary for the custom gate, `None` can be used
        instead. The remainder of the arguments are passed to the constructor
        for `ArbData`; the resulting `ArbData` object is passed along with the
        gate for custom data.
        """
        if matrix is not None:
            matrix = Handle(raw.dqcs_mat_new(matrix))
        with QubitSet._to_raw(targets) as targets:
            with QubitSet._to_raw(controls) as controls:
                with QubitSet._to_raw(measures) as measures:
                    if matrix is not None:
                        with matrix:
                            gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, int(matrix)))
                    else:
                        gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, 0))
                    ArbData(*args, **kwargs)._to_raw(gate)
                    with gate as gate:
                        self._pc(raw.dqcs_plugin_gate, gate)

    def get_measurement(self, qubit):
        """Returns the `Measurement` representing the latest measurement result
        for the given downstream qubit."""
        return Measurement._from_raw(Handle(self._pc(raw.dqcs_plugin_get_measurement, qubit)))

    def get_cycles_since_measure(self, qubit):
        """Returns the number of cycles that have been advanced since the
        latest measurement of the given downstream qubit."""
        return self._pc(raw.dqcs_plugin_get_cycles_since_measure, qubit)

    def get_cycles_between_measures(self, qubit):
        """Returns the number of cycles that were advanced between the
        latest measurement of the given downstream qubit and the one before."""
        return self._pc(raw.dqcs_plugin_get_cycles_between_measures, qubit)

    def advance(self, cycles):
        """Instructs the downstream plugin to advance the simulation time.

        `cycles` must be a nonnegative integer, representing the number of
        cycles to advance the simulation by. The simulation time after the
        advancement is returned."""
        return self._pc(raw.dqcs_plugin_advance, int(cycles))

    def get_cycle(self):
        """Returns the current simulation time for the downstream plugin."""
        return self._pc(raw.dqcs_plugin_get_cycle)

    def arb(self, *args, **kwargs):
        """Sends an `ArbCmd` to the downstream plugin.

        The arguments passed to this function are forwarded directly to the
        `ArbCmd` constructor. The return value is an `ArbData` object
        representing the value returned by the command."""
        with ArbCmd(*args, **kwargs)._to_raw() as cmd:
            return ArbData._from_raw(Handle(self._pc(raw.dqcs_plugin_arb, cmd)))

class Frontend(GateStreamSource):
    """Implements a frontend plugin.

    Frontends execute mixed quantum-classical algorithms, turning them into a
    gatestream for a downstream plugin to consume. They run as slaves to the
    host program, with which they can communicate by means of an ArbData queue
    in either direction.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

     - `handle_run(*args, **kwargs) -> ArbData or None`

        Called by the host program through its `start()` API call. The positional
        arguments are set to the list of binary strings from the ArbData
        argument, and **kwargs is set to the JSON object. The returned ArbData
        object can be retrieved by the host using the `wait()` API call. If you
        return None, an empty ArbData object will be automatically generated for
        the response.

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_host_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an ArbCmd is received from the host with the interface and
        operation identifiers embedded in the name. That is, you don't have to
        do interface/operation identifier matching yourself; you just specify
        the operations that you support. The positional arguments are set to the
        list of binary strings attached to the ArbCmd, and **kwargs is set to
        the JSON object. If you return None, an empty ArbData object will be
        automatically generated for the response.
    """

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def send(self, *args, **kwargs):
        """Sends an ArbData object to the host.

        The arguments to this function are passed to the constructor of
        `ArbData` to produce the object that is to be sent."""
        data = ArbData(*args, **kwargs)._to_raw()
        with data as d:
            self._pc(raw.dqcs_plugin_send, d)

    def recv(self):
        """Receives an ArbData object to the host.

        This blocks until data is received. The data is returned in the form of
        an `ArbData` object."""
        handle = Handle(self._pc(raw.dqcs_plugin_recv))
        return ArbData._from_raw(handle)

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_run(self, state_handle, arb_handle):
        """Routes the run callback to user code."""
        arg = ArbData._from_raw(Handle(arb_handle))
        result = self._cb(state_handle, 'handle_run', *arg._args, **arg._json)
        if result is None:
            result = ArbData()
        if not isinstance(result, ArbData):
            raise TypeError("User implementation of handle_run() should return None or ArbData but returned {}".format(type(result)))
        return result._to_raw().take()

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_FRONT)
        with pdef as pd:
            raw.dqcs_pdef_set_run_cb_pyfun(pd, self._cbent('run'))
        return pdef

class Operator(GateStreamSource):
    """Implements an operator plugin.

    Operators sit between frontends and backends, allowing them to observe or
    modify the quantum gate and measurement streams between them.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None`

        Called when the upstream plugin needs more qubits. The qubits list
        specifies the (integer) references that will be used by future calls to
        refer to the qubits (thus, the length of the list is the number of
        qubits that are to be allocated). The cmds parameter is passed a list of
        ArbCmds that the upstream plugin wants to associate with the qubits.

        In almost all cases, this handler must call `allocate()` to forward the
        allocation downstream, in some cases modifying the number of qubits or
        command list. If the handler is not specified, the allocation is
        forwarded automatically.

     - `handle_free(qubits: [Qubit]) -> None`

        Called when the upstream plugin doesn't need the specified qubits
        anymore.

        In almost all cases, this handler must call `free()` to forward the
        deallocation downstream. If the operator performs some kind of mapping
        function and/or modifies allocations, it may also need to modify the
        qubit list. If the handler is not specified, the deallocation is
        forwarded automatically.

     - `handle_unitary_gate(
            targets: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to execute a non-controlled
        unitary gate. The gate is normally forwarded downstream using
        `unitary()`. If this callback is not defined, DQCsim attempts to call
        `handle_controlled_gate()` with an empty list for the control qubits.
        If that callback is also not defined, the gate is forwarded downstream
        automatically.

     - `handle_controlled_gate(
            targets: [Qubit],
            controls: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to execute a controlled unitary
        gate, or a non-controlled gate if `handle_unitary_gate()` is not
        defined. The gate is normally forwarded downstream using `unitary()`.
        If this handler is not defined, the gate is forwarded downstream
        automatically.

     - `handle_measurement_gate(
            meas: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> [Measurement]`

        Called when the upstream plugin wants to execute a basic measurement on
        the given set of qubits. The gate is normally forwarded downstream using
        `measure()`. If this handler is not defined, this is done automatically.

        The basis is a 2x2 matrix. The semantics are as follows:

         - rotate each qubit by the inverse/hermitian/conjugate transpose of the
           given matrix
         - do a Z measurement for each qubit
         - rotate each qubit by the given matrix

        The result of this handler must be that measurements for exactly those
        qubits specified in `meas` are forwarded upstream after all downstream
        gates issued by it finish executing. Usually this happens naturally by
        forwarding the measurement gate unchanged. However, operators that
        perform some kind of qubit remapping will need to do some extra work
        to perform the inverse mapping when the measurements are returned
        upstream. The recommended way to do this is by defining
        `handle_measurement()`, but it is also possible to return measurement
        results directly in the same way backends must do it (or to do a
        combination of the two). The latter might negatively impact simulation
        speed, though.

     - `handle_prepare_gate(
            target: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to reset the state for the given
        qubits. The gate is normally forwarded downstream using `prepare()`. If
        this handler is not defined, this is done automatically.

        The basis is a 2x2 matrix. The semantics are as follows:

         - initialize each qubit to |0>
         - rotate each qubit by the given matrix

     -  `handle_<name>_gate(
            targets: [Qubit],
            controls: [Qubit],
            measures: [Qubit],
            matrix: [complex] or None,
            *args, **kwargs
        ) -> {Qubit: value} or None
        `

        Called when a custom (named) gate must be performed. The targets,
        controls, measures, and matrix share the functionality of
        `handle_controlled_gate()` and `handle_measurement_gate()`, as does the
        return value for the latter. Custom gates also have an attached
        `ArbData`, of which the binary string list is passed to `*args`, and
        the JSON object is passed to `**kwargs`.

        The gate is normally forwarded downstream using `custom_gate()`.
        If this handler is not defined, this is done automatically.

     - `handle_measurement(meas: Measurement) -> [Measurement]`

        Called when measurement data is received from the downstream plugin,
        allowing it to be modified before it is forwarded upstream. Modification
        includes not passing the measurement through (by returning an empty
        list), turning it into multiple measurements, changing the qubit
        reference to support qubit mapping, or just changing the measurement
        data itself to introduce errors or compensate for an earlier
        modification of the gatestream. If this handler is not defined,
        measurements are forwarded without modification.

     - `handle_advance(cycles: int) -> None`

        Called when the upstream plugin wants to advance simulation time.
        Normally this handler calls `advance(cycles)` in response, which is the
        default behavior if it is not defined. The primary use of this handler
        is to send additional gates downstream to model errors.

     - `handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an `ArbCmd` is received from the upstream plugin or from
        the host with the interface and operation identifiers embedded in the
        name. That is, you don't have to do interface/operation identifier
        matching yourself; you just specify the operations that you support.
        The positional arguments are set to the list of binary strings attached
        to the `ArbCmd`, and `**kwargs` is set to the JSON object. If you
        return `None`, an empty `ArbData` object will be automatically
        generated for the response.

        If no handlers are available for the requested command interface, the
        command is forwarded downstream. If there is at least one, it is NOT
        forwarded downstream, even if the requested operation does not have a
        handler (an error will be reported instead).
    """

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_allocate(self, state_handle, qubits_handle, cmds_handle):
        """Routes the allocate callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        cmds = ArbCmdQueue._from_raw(Handle(cmds_handle))
        self._cb(state_handle, 'handle_allocate', qubits, cmds)

    def _route_free(self, state_handle, qubits_handle):
        """Routes the free callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        self._cb(state_handle, 'handle_free', qubits)

    def _forward_unitary_gate(self, targets, controls, matrix, arb):
        self.unitary(targets, matrix, controls, arb)

    def _forward_measurement_gate(self, qubits, basis, arb):
        self.measure(qubits, basis=basis, arb=arb)

    def _forward_prepare_gate(self, qubits, basis, arb):
        self.prepare(qubits, basis=basis, arb=arb)

    def _route_gate(self, state_handle, gate_handle):
        """Routes the gate callback to user code."""

        typ = raw.dqcs_gate_type(gate_handle)
        name = None
        if raw.dqcs_gate_has_name(gate_handle):
            name = raw.dqcs_gate_name(gate_handle)

        # Forward gate types that don't have handlers as soon as possible.
        fast_forward = False
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            if raw.dqcs_gate_has_controls(gate_handle):
                fast_forward = not hasattr(self, 'handle_controlled_gate')
            else:
                fast_forward = ( #@
                    not hasattr(self, 'handle_unitary_gate')
                    and not hasattr(self, 'handle_controlled_gate'))
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            fast_forward = fast_forward and not hasattr(self, 'handle_measurement_gate')
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            fast_forward = fast_forward and not hasattr(self, 'handle_prepare_gate')
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            fast_forward = not hasattr(self, 'handle_{}_gate'.format(name))
        if fast_forward:
            raw.dqcs_plugin_gate(state_handle, gate_handle)
            return MeasurementSet._to_raw([]).take()

        # Convert from Rust domain to Python domain.
        targets = QubitSet._from_raw(Handle(raw.dqcs_gate_targets(gate_handle)))
        controls = QubitSet._from_raw(Handle(raw.dqcs_gate_controls(gate_handle)))
        measures = QubitSet._from_raw(Handle(raw.dqcs_gate_measures(gate_handle)))
        if raw.dqcs_gate_has_matrix(gate_handle):
            matrix = raw.dqcs_gate_matrix(gate_handle)
            matrix = raw.dqcs_mat_get(matrix)
        else:
            matrix = None

        # Route to the user's callback functions or execute the default
        # actions.
        data = ArbData._from_raw(Handle(gate_handle))
        measurements = []
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            try:
                if not controls and hasattr(self, 'handle_unitary_gate'):
                    self._cb(state_handle, 'handle_unitary_gate', targets, matrix, data)
                else:
                    self._cb(state_handle, 'handle_controlled_gate', targets, controls, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_unitary_gate', targets, controls, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            try:
                measurements = self._cb(state_handle, 'handle_measurement_gate', measures, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_measurement_gate', measures, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            try:
                self._cb(state_handle, 'handle_prepare_gate', targets, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_prepare_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            # Note that `handle_<name>_gate` must exist at this point,
            # otherwise it would have been forwarded earlier.
            cb_name = 'handle_{}_gate'.format(name)
            assert(hasattr(self, cb_name))
            measurements = self._cb(state_handle,
                cb_name, targets, controls, measures, matrix, *data._args, **data._json)
        else:
            raise NotImplementedError("unknown gate type")

        if measurements is None:
            measurements = []
        return MeasurementSet._to_raw(measurements).take()

    def _route_measurement(self, state_handle, measurement_handle):
        """Routes the measurement callback to user code."""
        measurement = Measurement._from_raw(Handle(measurement_handle))
        measurements = self._cb(state_handle, 'handle_measurement', measurement)
        if measurements is None:
            measurements = []
        elif isinstance(measurements, Measurement):
            measurements = [measurements]
        else:
            measurements = list(measurements)
        return MeasurementSet._to_raw(measurements).take()

    def _route_advance(self, state_handle, cycles):
        """Routes the advance callback to user code."""
        self._cb(state_handle, 'handle_advance', cycles)

    def _forward_upstream_arb(self, cmd):
        """Forwards an `ArbCmd` that originates from the upstream plugin
        further downstream."""
        return self.arb(cmd)

    def _route_upstream_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the upstream plugin."""
        return self._route_arb(state_handle, 'upstream', cmd_handle, '_forward_upstream_arb')

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_OPER)
        with pdef as pd:
            # Install Python callback handlers only when the user actually has
            # handlers defined for them. When they're not installed, events
            # will be forwarded downstream by Rust code, which is probably
            # significantly faster.
            handlers = set((
                member[0]
                for member in inspect.getmembers(
                    self, predicate=inspect.ismethod)
                if member[0].startswith('handle_')))
            if 'handle_allocate' in handlers:
                raw.dqcs_pdef_set_allocate_cb_pyfun(pd, self._cbent('allocate'))
            if 'handle_free' in handlers:
                raw.dqcs_pdef_set_free_cb_pyfun(pd, self._cbent('free'))
            if any(map(lambda name: name.endswith('_gate'), handlers)):
                raw.dqcs_pdef_set_gate_cb_pyfun(pd, self._cbent('gate'))
            if 'handle_measurement' in handlers:
                raw.dqcs_pdef_set_modify_measurement_cb_pyfun(pd, self._cbent('measurement'))
            if 'handle_advance' in handlers:
                raw.dqcs_pdef_set_advance_cb_pyfun(pd, self._cbent('advance'))
            if any(map(lambda name: name.startswith('handle_upstream_'), handlers)):
                raw.dqcs_pdef_set_upstream_arb_cb_pyfun(pd, self._cbent('upstream_arb'))
        return pdef

class Backend(Plugin):
    """Implements a backend plugin.

    Backends consume a quantum gate stream, simulate the gates and qubits, and
    return measurement data to the upstream plugin.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

     - `handle_unitary_gate(
            targets: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when a unitary gate must be handled: it must apply the given
        unitary matrix to the given list of qubits.

     - `handle_measurement_gate(
            qubits: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> [Measurement]`

        Called when a measurement must be performed. The returned list must
        contain measurement data for exactly those qubits specified by the
        qubits parameter.

        The basis is a 2x2 matrix. The semantics are as follows:

         - rotate each qubit by the inverse/hermitian/conjugate transpose of the
           given matrix
         - do a Z measurement for each qubit
         - rotate each qubit by the given matrix

     - `handle_prepare_gate(
            target: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to reset the state for the given
        qubits. The basis is a 2x2 matrix. The semantics are as follows:

         - initialize each qubit to |0>
         - rotate each qubit by the given matrix

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None`

        Called when the upstream plugin needs more qubits. The qubits list
        specifies the (integer) references that will be used by future calls to
        refer to the qubits (thus, the length of the list is the number of
        qubits that are to be allocated). The cmds parameter is passed a list of
        ArbCmds that the upstream plugin wants to associate with the qubits.

     - `handle_free(qubits: [Qubit]) -> None`

        Called when the upstream plugin doesn't need the specified qubits
        anymore.

     - `handle_controlled_gate(
            targets: [Qubit],
            controls: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when a controlled gate must be handled: it must apply the given
        unitary matrix to the target qubits "if the control qubits are set". In
        other words, it must first turn the given matrix into a controlled
        matrix for the specified number of control qubits, and then apply that
        gate to the concatenation of the target and control lists. If this
        function is not specified, this matrix upscaling is performed
        automatically, allowing `handle_unitary_gate()` to be called instead.
        You only have to implement this if your implementation can get a
        performance boost by doing this conversion manually.

     - `handle_<name>_gate(
            targets: [Qubit],
            controls: [Qubit],
            measures: [Qubit],
            matrix: [complex] or None,
            *args, **kwargs
        ) -> [Measurement]`

        Called when a custom (named) gate must be performed. The targets,
        controls, measures, and matrix share the functionality of
        `handle_controlled_gate()` and `handle_measurement_gate()`, as does the
        return value for the latter. Custom gates also have an attached ArbData,
        of which the binary string list is passed to `*args`, and the JSON
        object is passed to `**kwargs`.

     - `handle_advance(cycles: int) -> None`

        Called to advance simulation time.

     - `handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an ArbCmd is received from the upstream plugin or from the
        host with the interface and operation identifiers embedded in the name.
        That is, you don't have to do interface/operation identifier matching
        yourself; you just specify the operations that you support. The
        positional arguments are set to the list of binary strings attached to
        the ArbCmd, and **kwargs is set to the JSON object. If you return None,
        an empty ArbData object will be automatically generated for the
        response.
    """

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_allocate(self, state_handle, qubits_handle, cmds_handle):
        """Routes the allocate callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        cmds = ArbCmdQueue._from_raw(Handle(cmds_handle))
        try:
            self._cb(state_handle, 'handle_allocate', qubits, cmds)
        except NotImplementedError:
            pass

    def _route_free(self, state_handle, qubits_handle):
        """Routes the free callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        try:
            self._cb(state_handle, 'handle_free', qubits)
        except NotImplementedError:
            pass

    def _route_gate(self, state_handle, gate_handle):
        """Routes the gate callback to user code."""
        typ = raw.dqcs_gate_type(gate_handle)
        name = None
        if raw.dqcs_gate_has_name(gate_handle):
            name = raw.dqcs_gate_name(gate_handle)
        targets = QubitSet._from_raw(Handle(raw.dqcs_gate_targets(gate_handle)))
        controls = QubitSet._from_raw(Handle(raw.dqcs_gate_controls(gate_handle)))
        measures = QubitSet._from_raw(Handle(raw.dqcs_gate_measures(gate_handle)))
        if raw.dqcs_gate_has_matrix(gate_handle):
            matrix = raw.dqcs_gate_matrix(gate_handle)
            matrix = raw.dqcs_mat_get(matrix)
        else:
            matrix = None

        data = ArbData._from_raw(Handle(gate_handle))
        measurements = []
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            if controls:
                try:
                    self._cb(state_handle, 'handle_controlled_gate', targets, controls, matrix, data)
                    return MeasurementSet._to_raw([]).take()
                except NotImplementedError:
                    pass

                # Convert the gate matrix to a controlled gate matrix.
                cur_nq = len(targets)
                cur_size = 2**cur_nq
                assert(len(matrix) == cur_size * cur_size)
                ext_nq = len(controls) + len(targets)
                ext_size = 2**ext_nq
                offset = ext_size - cur_size

                # Make zero matrix of the right size.
                ext_matrix = [0.0+0.0j] * (ext_size * ext_size)

                # Override the lower-right block of the upscaled matrix
                # with the original matrix.
                for i in range(cur_size):
                    ext_matrix[(offset + i)*ext_size + offset : (offset + i)*ext_size + ext_size] = matrix[i*cur_size : i*cur_size + cur_size]

                # Turn the top-left block into an identity matrix.
                for i in range(offset):
                    ext_matrix[i*ext_size + i] = 1.0+0.0j

                # Replace the matrix and update the targets.
                matrix = ext_matrix
                targets = controls + targets

            self._cb(state_handle, 'handle_unitary_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            measurements = self._cb(state_handle, 'handle_measurement_gate', measures, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            self._cb(state_handle, 'handle_prepare_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            try:
                measurements = self._cb(state_handle,
                    'handle_{}_gate'.format(name),
                    targets, controls, measures, matrix, *data._args, **data._json)
            except NotImplementedError:
                raise NotImplementedError("{} gate is not implemented by this plugin".format(name))
        else:
            raise NotImplementedError("unknown gate type")


        if measurements is None:
            measurements = []
        return MeasurementSet._to_raw(measurements).take()

    def _route_advance(self, state_handle, cycles):
        """Routes the advance callback to user code."""
        try:
            self._cb(state_handle, 'handle_advance', cycles)
        except NotImplementedError:
            pass

    def _route_upstream_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the upstream plugin."""
        return self._route_arb(state_handle, 'upstream', cmd_handle)

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_BACK)
        with pdef as pd:
            raw.dqcs_pdef_set_allocate_cb_pyfun(pd, self._cbent('allocate'))
            raw.dqcs_pdef_set_free_cb_pyfun(pd, self._cbent('free'))
            raw.dqcs_pdef_set_gate_cb_pyfun(pd, self._cbent('gate'))
            raw.dqcs_pdef_set_advance_cb_pyfun(pd, self._cbent('advance'))
            raw.dqcs_pdef_set_upstream_arb_cb_pyfun(pd, self._cbent('upstream_arb'))
        return pdef

Classes

class Backend (host_arb_ifaces=None, upstream_arb_ifaces=None)

Implements a backend plugin.

Backends consume a quantum gate stream, simulate the gates and qubits, and return measurement data to the upstream plugin.

The following functions MUST be implemented by the user:

• get_name() -> str

  Must return the name of the plugin implementation.

• get_author() -> str

  Must return the name of the plugin author.

• get_version() -> str

  Must return the plugin's version string.

• handle_unitary_gate( targets: [Qubit], matrix: [complex], arb: ArbData ) -> None

  Called when a unitary gate must be handled: it must apply the given unitary matrix to the given list of qubits.

• handle_measurement_gate( qubits: [Qubit], basis: [complex], arb: ArbData ) -> [Measurement]

  Called when a measurement must be performed. The returned list must contain measurement data for exactly those qubits specified by the qubits parameter.

  The basis is a 2x2 matrix. The semantics are as follows:

  • rotate each qubit by the inverse/hermitian/conjugate transpose of the given matrix
  • do a Z measurement for each qubit
  • rotate each qubit by the given matrix

• handle_prepare_gate( target: [Qubit], basis: [complex], arb: ArbData ) -> None

  Called when the upstream plugin wants to reset the state for the given qubits. The basis is a 2x2 matrix. The semantics are as follows:

  • initialize each qubit to |0>
  • rotate each qubit by the given matrix

The following functions MAY be implemented by the user:

• handle_init(cmds: [ArbCmd]) -> None

  Called by the simulator to initialize this plugin. The cmds parameter is passed a list of ArbCmds that the simulator wishes to associate with the plugin. If this function is not implemented, or if it is implemented but does not take an argument, the initialization ArbCmds are treated as regular host arbs (that is, they're passed to handle_host_<iface>_<oper>() if those functions do exist).

• handle_drop() -> None

  Called by the simulator when the simulation terminates.

• handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None

  Called when the upstream plugin needs more qubits. The qubits list specifies the (integer) references that will be used by future calls to refer to the qubits (thus, the length of the list is the number of qubits that are to be allocated). The cmds parameter is passed a list of ArbCmds that the upstream plugin wants to associate with the qubits.

• handle_free(qubits: [Qubit]) -> None

  Called when the upstream plugin doesn't need the specified qubits anymore.

• handle_controlled_gate( targets: [Qubit], controls: [Qubit], matrix: [complex], arb: ArbData ) -> None

  Called when a controlled gate must be handled: it must apply the given unitary matrix to the target qubits "if the control qubits are set". In other words, it must first turn the given matrix into a controlled matrix for the specified number of control qubits, and then apply that gate to the concatenation of the target and control lists. If this function is not specified, this matrix upscaling is performed automatically, allowing handle_unitary_gate() to be called instead. You only have to implement this if your implementation can get a performance boost by doing this conversion manually.

• handle_<name>_gate( targets: [Qubit], controls: [Qubit], measures: [Qubit], matrix: [complex] or None, *args, **kwargs ) -> [Measurement]

  Called when a custom (named) gate must be performed. The targets, controls, measures, and matrix share the functionality of handle_controlled_gate() and handle_measurement_gate(), as does the return value for the latter. Custom gates also have an attached ArbData, of which the binary string list is passed to *args, and the JSON object is passed to **kwargs.

• handle_advance(cycles: int) -> None

  Called to advance simulation time.

• handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None

  Called when an ArbCmd is received from the upstream plugin or from the host with the interface and operation identifiers embedded in the name. That is, you don't have to do interface/operation identifier matching yourself; you just specify the operations that you support. The positional arguments are set to the list of binary strings attached to the ArbCmd, and **kwargs is set to the JSON object. If you return None, an empty ArbData object will be automatically generated for the response.

Creates the plugin object.

Overriding __init__() in your implementation is fine, but you must call super().__init__() if you do this.

Among other things, this function auto-detects which arb interfaces are supported by your plugin by scanning for handle_host_*() and handle_upstream_*() implementations. However, this auto-detection will fail if there are underscores in any of the supported interface or operation IDs. In this case, you MUST override __init__() and call super().__init__(host_arb_ifaces, upstream_arb_ifaces), where the two arguments are lists of strings representing the supported interface IDs.

Expand source code

class Backend(Plugin):
    """Implements a backend plugin.

    Backends consume a quantum gate stream, simulate the gates and qubits, and
    return measurement data to the upstream plugin.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

     - `handle_unitary_gate(
            targets: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when a unitary gate must be handled: it must apply the given
        unitary matrix to the given list of qubits.

     - `handle_measurement_gate(
            qubits: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> [Measurement]`

        Called when a measurement must be performed. The returned list must
        contain measurement data for exactly those qubits specified by the
        qubits parameter.

        The basis is a 2x2 matrix. The semantics are as follows:

         - rotate each qubit by the inverse/hermitian/conjugate transpose of the
           given matrix
         - do a Z measurement for each qubit
         - rotate each qubit by the given matrix

     - `handle_prepare_gate(
            target: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to reset the state for the given
        qubits. The basis is a 2x2 matrix. The semantics are as follows:

         - initialize each qubit to |0>
         - rotate each qubit by the given matrix

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None`

        Called when the upstream plugin needs more qubits. The qubits list
        specifies the (integer) references that will be used by future calls to
        refer to the qubits (thus, the length of the list is the number of
        qubits that are to be allocated). The cmds parameter is passed a list of
        ArbCmds that the upstream plugin wants to associate with the qubits.

     - `handle_free(qubits: [Qubit]) -> None`

        Called when the upstream plugin doesn't need the specified qubits
        anymore.

     - `handle_controlled_gate(
            targets: [Qubit],
            controls: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when a controlled gate must be handled: it must apply the given
        unitary matrix to the target qubits "if the control qubits are set". In
        other words, it must first turn the given matrix into a controlled
        matrix for the specified number of control qubits, and then apply that
        gate to the concatenation of the target and control lists. If this
        function is not specified, this matrix upscaling is performed
        automatically, allowing `handle_unitary_gate()` to be called instead.
        You only have to implement this if your implementation can get a
        performance boost by doing this conversion manually.

     - `handle_<name>_gate(
            targets: [Qubit],
            controls: [Qubit],
            measures: [Qubit],
            matrix: [complex] or None,
            *args, **kwargs
        ) -> [Measurement]`

        Called when a custom (named) gate must be performed. The targets,
        controls, measures, and matrix share the functionality of
        `handle_controlled_gate()` and `handle_measurement_gate()`, as does the
        return value for the latter. Custom gates also have an attached ArbData,
        of which the binary string list is passed to `*args`, and the JSON
        object is passed to `**kwargs`.

     - `handle_advance(cycles: int) -> None`

        Called to advance simulation time.

     - `handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an ArbCmd is received from the upstream plugin or from the
        host with the interface and operation identifiers embedded in the name.
        That is, you don't have to do interface/operation identifier matching
        yourself; you just specify the operations that you support. The
        positional arguments are set to the list of binary strings attached to
        the ArbCmd, and **kwargs is set to the JSON object. If you return None,
        an empty ArbData object will be automatically generated for the
        response.
    """

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_allocate(self, state_handle, qubits_handle, cmds_handle):
        """Routes the allocate callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        cmds = ArbCmdQueue._from_raw(Handle(cmds_handle))
        try:
            self._cb(state_handle, 'handle_allocate', qubits, cmds)
        except NotImplementedError:
            pass

    def _route_free(self, state_handle, qubits_handle):
        """Routes the free callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        try:
            self._cb(state_handle, 'handle_free', qubits)
        except NotImplementedError:
            pass

    def _route_gate(self, state_handle, gate_handle):
        """Routes the gate callback to user code."""
        typ = raw.dqcs_gate_type(gate_handle)
        name = None
        if raw.dqcs_gate_has_name(gate_handle):
            name = raw.dqcs_gate_name(gate_handle)
        targets = QubitSet._from_raw(Handle(raw.dqcs_gate_targets(gate_handle)))
        controls = QubitSet._from_raw(Handle(raw.dqcs_gate_controls(gate_handle)))
        measures = QubitSet._from_raw(Handle(raw.dqcs_gate_measures(gate_handle)))
        if raw.dqcs_gate_has_matrix(gate_handle):
            matrix = raw.dqcs_gate_matrix(gate_handle)
            matrix = raw.dqcs_mat_get(matrix)
        else:
            matrix = None

        data = ArbData._from_raw(Handle(gate_handle))
        measurements = []
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            if controls:
                try:
                    self._cb(state_handle, 'handle_controlled_gate', targets, controls, matrix, data)
                    return MeasurementSet._to_raw([]).take()
                except NotImplementedError:
                    pass

                # Convert the gate matrix to a controlled gate matrix.
                cur_nq = len(targets)
                cur_size = 2**cur_nq
                assert(len(matrix) == cur_size * cur_size)
                ext_nq = len(controls) + len(targets)
                ext_size = 2**ext_nq
                offset = ext_size - cur_size

                # Make zero matrix of the right size.
                ext_matrix = [0.0+0.0j] * (ext_size * ext_size)

                # Override the lower-right block of the upscaled matrix
                # with the original matrix.
                for i in range(cur_size):
                    ext_matrix[(offset + i)*ext_size + offset : (offset + i)*ext_size + ext_size] = matrix[i*cur_size : i*cur_size + cur_size]

                # Turn the top-left block into an identity matrix.
                for i in range(offset):
                    ext_matrix[i*ext_size + i] = 1.0+0.0j

                # Replace the matrix and update the targets.
                matrix = ext_matrix
                targets = controls + targets

            self._cb(state_handle, 'handle_unitary_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            measurements = self._cb(state_handle, 'handle_measurement_gate', measures, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            self._cb(state_handle, 'handle_prepare_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            try:
                measurements = self._cb(state_handle,
                    'handle_{}_gate'.format(name),
                    targets, controls, measures, matrix, *data._args, **data._json)
            except NotImplementedError:
                raise NotImplementedError("{} gate is not implemented by this plugin".format(name))
        else:
            raise NotImplementedError("unknown gate type")


        if measurements is None:
            measurements = []
        return MeasurementSet._to_raw(measurements).take()

    def _route_advance(self, state_handle, cycles):
        """Routes the advance callback to user code."""
        try:
            self._cb(state_handle, 'handle_advance', cycles)
        except NotImplementedError:
            pass

    def _route_upstream_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the upstream plugin."""
        return self._route_arb(state_handle, 'upstream', cmd_handle)

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_BACK)
        with pdef as pd:
            raw.dqcs_pdef_set_allocate_cb_pyfun(pd, self._cbent('allocate'))
            raw.dqcs_pdef_set_free_cb_pyfun(pd, self._cbent('free'))
            raw.dqcs_pdef_set_gate_cb_pyfun(pd, self._cbent('gate'))
            raw.dqcs_pdef_set_advance_cb_pyfun(pd, self._cbent('advance'))
            raw.dqcs_pdef_set_upstream_arb_cb_pyfun(pd, self._cbent('upstream_arb'))
        return pdef

Ancestors

• Plugin

Inherited members

• Plugin:
  • critical
  • debug
  • error
  • fatal
  • info
  • log
  • note
  • random_float
  • random_long
  • run
  • start
  • trace
  • warn
  • warning

class Frontend (host_arb_ifaces=None, upstream_arb_ifaces=None)

Implements a frontend plugin.

Frontends execute mixed quantum-classical algorithms, turning them into a gatestream for a downstream plugin to consume. They run as slaves to the host program, with which they can communicate by means of an ArbData queue in either direction.

The following functions MUST be implemented by the user:

• get_name() -> str

  Must return the name of the plugin implementation.

• get_author() -> str

  Must return the name of the plugin author.

• get_version() -> str

  Must return the plugin's version string.

• handle_run(*args, **kwargs) -> ArbData or None

  Called by the host program through its start() API call. The positional arguments are set to the list of binary strings from the ArbData argument, and **kwargs is set to the JSON object. The returned ArbData object can be retrieved by the host using the wait() API call. If you return None, an empty ArbData object will be automatically generated for the response.

The following functions MAY be implemented by the user:

• handle_init(cmds: [ArbCmd]) -> None

  Called by the simulator to initialize this plugin. The cmds parameter is passed a list of ArbCmds that the simulator wishes to associate with the plugin. If this function is not implemented, or if it is implemented but does not take an argument, the initialization ArbCmds are treated as regular host arbs (that is, they're passed to handle_host_<iface>_<oper>() if those functions do exist).

• handle_drop() -> None

  Called by the simulator when the simulation terminates.

• handle_host_<iface>_<oper>(*args, **kwargs) -> ArbData or None

  Called when an ArbCmd is received from the host with the interface and operation identifiers embedded in the name. That is, you don't have to do interface/operation identifier matching yourself; you just specify the operations that you support. The positional arguments are set to the list of binary strings attached to the ArbCmd, and **kwargs is set to the JSON object. If you return None, an empty ArbData object will be automatically generated for the response.

Creates the plugin object.

Overriding __init__() in your implementation is fine, but you must call super().__init__() if you do this.

Among other things, this function auto-detects which arb interfaces are supported by your plugin by scanning for handle_host_*() and handle_upstream_*() implementations. However, this auto-detection will fail if there are underscores in any of the supported interface or operation IDs. In this case, you MUST override __init__() and call super().__init__(host_arb_ifaces, upstream_arb_ifaces), where the two arguments are lists of strings representing the supported interface IDs.

Expand source code

class Frontend(GateStreamSource):
    """Implements a frontend plugin.

    Frontends execute mixed quantum-classical algorithms, turning them into a
    gatestream for a downstream plugin to consume. They run as slaves to the
    host program, with which they can communicate by means of an ArbData queue
    in either direction.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

     - `handle_run(*args, **kwargs) -> ArbData or None`

        Called by the host program through its `start()` API call. The positional
        arguments are set to the list of binary strings from the ArbData
        argument, and **kwargs is set to the JSON object. The returned ArbData
        object can be retrieved by the host using the `wait()` API call. If you
        return None, an empty ArbData object will be automatically generated for
        the response.

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_host_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an ArbCmd is received from the host with the interface and
        operation identifiers embedded in the name. That is, you don't have to
        do interface/operation identifier matching yourself; you just specify
        the operations that you support. The positional arguments are set to the
        list of binary strings attached to the ArbCmd, and **kwargs is set to
        the JSON object. If you return None, an empty ArbData object will be
        automatically generated for the response.
    """

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def send(self, *args, **kwargs):
        """Sends an ArbData object to the host.

        The arguments to this function are passed to the constructor of
        `ArbData` to produce the object that is to be sent."""
        data = ArbData(*args, **kwargs)._to_raw()
        with data as d:
            self._pc(raw.dqcs_plugin_send, d)

    def recv(self):
        """Receives an ArbData object to the host.

        This blocks until data is received. The data is returned in the form of
        an `ArbData` object."""
        handle = Handle(self._pc(raw.dqcs_plugin_recv))
        return ArbData._from_raw(handle)

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_run(self, state_handle, arb_handle):
        """Routes the run callback to user code."""
        arg = ArbData._from_raw(Handle(arb_handle))
        result = self._cb(state_handle, 'handle_run', *arg._args, **arg._json)
        if result is None:
            result = ArbData()
        if not isinstance(result, ArbData):
            raise TypeError("User implementation of handle_run() should return None or ArbData but returned {}".format(type(result)))
        return result._to_raw().take()

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_FRONT)
        with pdef as pd:
            raw.dqcs_pdef_set_run_cb_pyfun(pd, self._cbent('run'))
        return pdef

Ancestors

• GateStreamSource
• Plugin

Methods

def recv(self)

Receives an ArbData object to the host.

This blocks until data is received. The data is returned in the form of an ArbData object.

Expand source code

def recv(self):
    """Receives an ArbData object to the host.

    This blocks until data is received. The data is returned in the form of
    an `ArbData` object."""
    handle = Handle(self._pc(raw.dqcs_plugin_recv))
    return ArbData._from_raw(handle)

def send(self, *args, **kwargs)

Sends an ArbData object to the host.

The arguments to this function are passed to the constructor of ArbData to produce the object that is to be sent.

Expand source code

def send(self, *args, **kwargs):
    """Sends an ArbData object to the host.

    The arguments to this function are passed to the constructor of
    `ArbData` to produce the object that is to be sent."""
    data = ArbData(*args, **kwargs)._to_raw()
    with data as d:
        self._pc(raw.dqcs_plugin_send, d)

Inherited members

• GateStreamSource:
  • advance
  • allocate
  • arb
  • cnot_gate
  • critical
  • custom_gate
  • debug
  • error
  • fatal
  • fredkin_gate
  • free
  • get_cycle
  • get_cycles_between_measures
  • get_cycles_since_measure
  • get_measurement
  • h_gate
  • i_gate
  • info
  • log
  • measure
  • measure_x
  • measure_y
  • measure_z
  • mx90_gate
  • my90_gate
  • mz90_gate
  • note
  • prepare
  • prepare_x
  • prepare_y
  • prepare_z
  • r_gate
  • random_float
  • random_long
  • run
  • rx_gate
  • ry_gate
  • rz_gate
  • s_gate
  • sdag_gate
  • sqswap_gate
  • start
  • swap_gate
  • t_gate
  • tdag_gate
  • toffoli_gate
  • trace
  • unitary
  • warn
  • warning
  • x90_gate
  • x_gate
  • y90_gate
  • y_gate
  • z90_gate
  • z_gate

class GateStreamSource (host_arb_ifaces=None, upstream_arb_ifaces=None)

Adds gatestream source functions.

Creates the plugin object.

Overriding __init__() in your implementation is fine, but you must call super().__init__() if you do this.

Among other things, this function auto-detects which arb interfaces are supported by your plugin by scanning for handle_host_*() and handle_upstream_*() implementations. However, this auto-detection will fail if there are underscores in any of the supported interface or operation IDs. In this case, you MUST override __init__() and call super().__init__(host_arb_ifaces, upstream_arb_ifaces), where the two arguments are lists of strings representing the supported interface IDs.

Expand source code

class GateStreamSource(Plugin):
    """Adds gatestream source functions."""

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def allocate(self, num_qubits=None, *cmds):
        """Instructs the downstream plugin to allocate one or more qubits.

        If `num_qubits` is specified, this function returns a list of qubit
        references that you can use to refer to the qubits in later function
        calls. These are just integers. If `num_qubits` is not specified or
        `None`, a single qubit is allocated and returned without being wrapped
        in a list.

        Optionally, you can pass (a list of) ArbCmd objects to associate with
        the qubits."""
        with ArbCmdQueue._to_raw(*cmds) as cmds:
            qubits = QubitSet._from_raw(Handle(self._pc(
                raw.dqcs_plugin_allocate,
                1 if num_qubits is None else num_qubits,
                cmds
            ))) #@
        if num_qubits is None:
            return qubits[0]
        else:
            return qubits

    def free(self, *qubits):
        """Instructs the downstream plugin to free the given qubits."""
        with QubitSet._to_raw(*qubits) as qubits:
            self._pc(raw.dqcs_plugin_free, qubits)

    def unitary(self, targets, matrix, controls=[], arb=None):
        """Instructs the downstream plugin to execute a unitary quantum gate.

        `targets` must be a non-empty iterable of qubits or a single qubit,
        representing the qubit(s) targeted by the gate. `matrix` must be a
        unitary matrix appropriately sized for the number of target qubits,
        specified as a row-major one-dimensional list of Python complex
        numbers. `controls` optionally allows additional control qubits to be
        specified to make controlled gates; these qubits should NOT be
        reflected in the gate matrix. The matrix will automatically be extended
        by the downstream plugin, instead. The `targets` and `controls` sets
        must not intersect.
        """
        with QubitSet._to_raw(targets) as targets:
            with QubitSet._to_raw(controls) as controls:
                with Handle(raw.dqcs_mat_new(matrix)) as mat:
                    gate = Handle(raw.dqcs_gate_new_unitary(targets, controls, mat))
                    if arb is not None:
                        if not isinstance(arb, ArbData):
                            raise TypeError('arb must be None or an instance of ArbData')
                        arb._to_raw(gate)
                    with gate as gate_raw:
                        self._pc(raw.dqcs_plugin_gate, gate_raw)

    def i_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute an I gate.

        `target` is the targetted qubit."""
        self.unitary(target, [1.0, 0.0, 0.0, 1.0], arb=arb)

    def rx_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary X rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = math.cos(0.5 * theta)
        b = -1.0j * math.sin(0.5 * theta)
        self.unitary(target, [a, b, b, a], arb=arb)

    def ry_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary Y rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = math.cos(0.5 * theta)
        b = math.sin(0.5 * theta)
        self.unitary(target, [a, -b, b, a], arb=arb)

    def rz_gate(self, target, theta, arb=None):
        """Instructs the downstream plugin to perform an arbitrary Z rotation.

        `target` is the targetted qubit. `theta` is the angle in radians."""
        a = cmath.exp(-0.5j * theta)
        b = cmath.exp(0.5j * theta)
        self.unitary(target, [a, 0.0, 0.0, b], arb=arb)

    def r_gate(self, target, theta, phi, lambd, arb=None):
        """Instructs the downstream plugin to perform a number of rotations at
        once.

        `target` is the targetted qubit. `theta`, `phi`, and `lambd` are the
        angles in radians."""
        a = math.cos(0.5 * theta)
        b = math.sin(0.5 * theta)
        self.unitary(target, [
            a,
            -b * cmath.exp(1.0j * lambd),
            b * cmath.exp(1.0j * phi),
            a * cmath.exp(1.0j * (phi + lambd)),
        ], arb=arb)

    def swap_gate(self, a, b, arb=None):
        """Instructs the downstream plugin to execute a swap gate.

        `a` and `b` are the targetted qubits."""
        self.unitary([a, b], [
            1.0, 0.0, 0.0, 0.0,
            0.0, 0.0, 1.0, 0.0,
            0.0, 1.0, 0.0, 0.0,
            0.0, 0.0, 0.0, 1.0,
        ], arb=arb) #@

    def sqswap_gate(self, a, b, arb=None):
        """Instructs the downstream plugin to execute a square-root-of-swap
        gate.

        `a` and `b` are the targetted qubits."""
        self.unitary([a, b], [
            1.0, 0.0,      0.0,      0.0,
            0.0, 0.5+0.5j, 0.5-0.5j, 0.0,
            0.0, 0.5-0.5j, 0.5+0.5j, 0.0,
            0.0, 0.0,      0.0,      1.0,
        ], arb=arb) #@

    def x_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute an X gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, math.pi, arb=arb)

    def x90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree X gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, 0.5 * math.pi, arb=arb)

    def mx90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree X
        gate.

        `target` is the targetted qubit."""
        self.rx_gate(target, -0.5 * math.pi, arb=arb)

    def y_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Y gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, math.pi, arb=arb)

    def y90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree Y gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, 0.5 * math.pi, arb=arb)

    def my90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree Y
        gate.

        `target` is the targetted qubit."""
        self.ry_gate(target, -0.5 * math.pi, arb=arb)

    def z_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Z gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, math.pi, arb=arb)

    def z90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a 90-degree Z gate, also
        known as an S gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, 0.5 * math.pi, arb=arb)

    def mz90_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a negative 90-degree Z
        gate, also known as an S-dagger gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, -0.5 * math.pi, arb=arb)

    s_gate = z90_gate
    sdag_gate = mz90_gate

    def t_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a T gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, 0.25 * math.pi, arb=arb)

    def tdag_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a T-dagger gate.

        `target` is the targetted qubit."""
        self.rz_gate(target, -0.25 * math.pi, arb=arb)

    def h_gate(self, target, arb=None):
        """Instructs the downstream plugin to execute a Hadamard gate.

        `target` is the targetted qubit."""
        x = 1 / math.sqrt(2.0)
        self.unitary([target], [x, x, x, -x], arb=arb)

    def cnot_gate(self, control, target, arb=None):
        """Instructs the downstream plugin to execute a CNOT gate."""
        self.unitary([target], [
            0.0, 1.0,
            1.0, 0.0,
        ], controls=[control], arb=arb) #@

    def toffoli_gate(self, c1, c2, target, arb=None):
        """Instructs the downstream plugin to execute a Toffoli gate."""
        self.unitary([target], [
            0.0, 1.0,
            1.0, 0.0,
        ], controls=[c1, c2], arb=arb) #@

    def fredkin_gate(self, control, a, b, arb=None):
        """Instructs the downstream plugin to execute a Fredkin gate."""
        self.unitary([a, b], [
            1.0, 0.0, 0.0, 0.0,
            0.0, 0.0, 1.0, 0.0,
            0.0, 1.0, 0.0, 0.0,
            0.0, 0.0, 0.0, 1.0,
        ], controls=[control], arb=arb) #@

    def measure(self, *qubits, basis='Z', arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        given basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only positional
        argument. The basis is always a keyword argument.

        The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
        numbers representing a 2x2 matrix with the following semantics:

         - the qubits are rotated by the inverse of the matrix
         - a Z-axis measurement is performed on each qubit
         - the qubits are rotated by the matrix
        """
        if isinstance(basis, str):
            basis = Handle(raw.dqcs_mat_basis({
                'X': raw.DQCS_BASIS_X,
                'Y': raw.DQCS_BASIS_Y,
                'Z': raw.DQCS_BASIS_Z,
            }[basis]))
        else:
            basis = Handle(raw.dqcs_mat_new(basis))
        if len(qubits) == 1 and not isinstance(qubits[0], int):
            qubits = list(qubits[0])
        with QubitSet._to_raw(qubits) as qubits:
            with basis as mat:
                gate = Handle(raw.dqcs_gate_new_measurement(qubits, mat))
                if arb is not None:
                    if not isinstance(arb, ArbData):
                        raise TypeError('arb must be None or an instance of ArbData')
                    arb._to_raw(gate)
                with gate as gate_raw:
                    self._pc(raw.dqcs_plugin_gate, gate_raw)

    def measure_x(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        X basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='X', arb=arb)

    def measure_y(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        Y basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='Y', arb=arb)

    def measure_z(self, *qubits, arb=None):
        """Instructs the downstream plugin to measure the given qubits in the
        Z basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.measure(*qubits, basis='Z', arb=arb)

    def prepare(self, *qubits, basis='Z', arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the given basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only positional
        argument. The basis is always a keyword argument.

        The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
        numbers representing a 2x2 matrix with the following semantics:

         - the qubits are initialized to |0>
         - the qubits are rotated by the matrix
        """
        if isinstance(basis, str):
            basis = Handle(raw.dqcs_mat_basis({
                'X': raw.DQCS_BASIS_X,
                'Y': raw.DQCS_BASIS_Y,
                'Z': raw.DQCS_BASIS_Z,
            }[basis]))
        else:
            basis = Handle(raw.dqcs_mat_new(basis))
        if len(qubits) == 1 and not isinstance(qubits[0], int):
            qubits = list(qubits[0])
        with QubitSet._to_raw(qubits) as qubits:
            with basis as mat:
                gate = Handle(raw.dqcs_gate_new_prep(qubits, mat))
                if arb is not None:
                    if not isinstance(arb, ArbData):
                        raise TypeError('arb must be None or an instance of ArbData')
                    arb._to_raw(gate)
                with gate as gate_raw:
                    self._pc(raw.dqcs_plugin_gate, gate_raw)

    def prepare_x(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the X basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='X', arb=arb)

    def prepare_y(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the Y basis.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='Y', arb=arb)

    def prepare_z(self, *qubits, arb=None):
        """Instructs the downstream plugin to force the given qubits into the
        base state for the Z basis, being |0>.

        This function takes either one or more qubits as its positional
        arguments, or an iterable of qubits as its first and only argument.
        """
        self.prepare(*qubits, basis='Z', arb=arb)

    def custom_gate(self, name, targets=[], controls=[], measures=[], matrix=None, *args, **kwargs):
        """Instructs the downstream plugin to execute a custom gate.

        `name` must be a non-empty string identifying the gate to be performed.
        `targets`, `constrols`, and `measures` must be iterables of qubits or
        singular qubits, representing respectively the set of qubits to operate
        on, the set of control qubits for controlled gates, and the set of
        qubits measured by the gate.  The `targets` and `controls` sets must
        not intersect. If specified, `matrix` must be a unitary matrix
        appropriately sized for the number of target qubits, specified as a
        row-major one-dimensional list of Python complex numbers. If no matrix
        is applicable or necessary for the custom gate, `None` can be used
        instead. The remainder of the arguments are passed to the constructor
        for `ArbData`; the resulting `ArbData` object is passed along with the
        gate for custom data.
        """
        if matrix is not None:
            matrix = Handle(raw.dqcs_mat_new(matrix))
        with QubitSet._to_raw(targets) as targets:
            with QubitSet._to_raw(controls) as controls:
                with QubitSet._to_raw(measures) as measures:
                    if matrix is not None:
                        with matrix:
                            gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, int(matrix)))
                    else:
                        gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, 0))
                    ArbData(*args, **kwargs)._to_raw(gate)
                    with gate as gate:
                        self._pc(raw.dqcs_plugin_gate, gate)

    def get_measurement(self, qubit):
        """Returns the `Measurement` representing the latest measurement result
        for the given downstream qubit."""
        return Measurement._from_raw(Handle(self._pc(raw.dqcs_plugin_get_measurement, qubit)))

    def get_cycles_since_measure(self, qubit):
        """Returns the number of cycles that have been advanced since the
        latest measurement of the given downstream qubit."""
        return self._pc(raw.dqcs_plugin_get_cycles_since_measure, qubit)

    def get_cycles_between_measures(self, qubit):
        """Returns the number of cycles that were advanced between the
        latest measurement of the given downstream qubit and the one before."""
        return self._pc(raw.dqcs_plugin_get_cycles_between_measures, qubit)

    def advance(self, cycles):
        """Instructs the downstream plugin to advance the simulation time.

        `cycles` must be a nonnegative integer, representing the number of
        cycles to advance the simulation by. The simulation time after the
        advancement is returned."""
        return self._pc(raw.dqcs_plugin_advance, int(cycles))

    def get_cycle(self):
        """Returns the current simulation time for the downstream plugin."""
        return self._pc(raw.dqcs_plugin_get_cycle)

    def arb(self, *args, **kwargs):
        """Sends an `ArbCmd` to the downstream plugin.

        The arguments passed to this function are forwarded directly to the
        `ArbCmd` constructor. The return value is an `ArbData` object
        representing the value returned by the command."""
        with ArbCmd(*args, **kwargs)._to_raw() as cmd:
            return ArbData._from_raw(Handle(self._pc(raw.dqcs_plugin_arb, cmd)))

Ancestors

• Plugin

Subclasses

• Frontend
• Operator

Methods

def advance(self, cycles)

Instructs the downstream plugin to advance the simulation time.

cycles must be a nonnegative integer, representing the number of cycles to advance the simulation by. The simulation time after the advancement is returned.

Expand source code

def advance(self, cycles):
    """Instructs the downstream plugin to advance the simulation time.

    `cycles` must be a nonnegative integer, representing the number of
    cycles to advance the simulation by. The simulation time after the
    advancement is returned."""
    return self._pc(raw.dqcs_plugin_advance, int(cycles))

def allocate(self, num_qubits=None, *cmds)

Instructs the downstream plugin to allocate one or more qubits.

If num_qubits is specified, this function returns a list of qubit references that you can use to refer to the qubits in later function calls. These are just integers. If num_qubits is not specified or None, a single qubit is allocated and returned without being wrapped in a list.

Optionally, you can pass (a list of) ArbCmd objects to associate with the qubits.

Expand source code

def allocate(self, num_qubits=None, *cmds):
    """Instructs the downstream plugin to allocate one or more qubits.

    If `num_qubits` is specified, this function returns a list of qubit
    references that you can use to refer to the qubits in later function
    calls. These are just integers. If `num_qubits` is not specified or
    `None`, a single qubit is allocated and returned without being wrapped
    in a list.

    Optionally, you can pass (a list of) ArbCmd objects to associate with
    the qubits."""
    with ArbCmdQueue._to_raw(*cmds) as cmds:
        qubits = QubitSet._from_raw(Handle(self._pc(
            raw.dqcs_plugin_allocate,
            1 if num_qubits is None else num_qubits,
            cmds
        ))) #@
    if num_qubits is None:
        return qubits[0]
    else:
        return qubits

def arb(self, *args, **kwargs)

Sends an ArbCmd to the downstream plugin.

The arguments passed to this function are forwarded directly to the ArbCmd constructor. The return value is an ArbData object representing the value returned by the command.

Expand source code

def arb(self, *args, **kwargs):
    """Sends an `ArbCmd` to the downstream plugin.

    The arguments passed to this function are forwarded directly to the
    `ArbCmd` constructor. The return value is an `ArbData` object
    representing the value returned by the command."""
    with ArbCmd(*args, **kwargs)._to_raw() as cmd:
        return ArbData._from_raw(Handle(self._pc(raw.dqcs_plugin_arb, cmd)))

def cnot_gate(self, control, target, arb=None)

Instructs the downstream plugin to execute a CNOT gate.

Expand source code

def cnot_gate(self, control, target, arb=None):
    """Instructs the downstream plugin to execute a CNOT gate."""
    self.unitary([target], [
        0.0, 1.0,
        1.0, 0.0,
    ], controls=[control], arb=arb) #@

def custom_gate(self, name, targets=[], controls=[], measures=[], matrix=None, *args, **kwargs)

Instructs the downstream plugin to execute a custom gate.

name must be a non-empty string identifying the gate to be performed. targets, constrols, and measures must be iterables of qubits or singular qubits, representing respectively the set of qubits to operate on, the set of control qubits for controlled gates, and the set of qubits measured by the gate. The targets and controls sets must not intersect. If specified, matrix must be a unitary matrix appropriately sized for the number of target qubits, specified as a row-major one-dimensional list of Python complex numbers. If no matrix is applicable or necessary for the custom gate, None can be used instead. The remainder of the arguments are passed to the constructor for ArbData; the resulting ArbData object is passed along with the gate for custom data.

Expand source code

def custom_gate(self, name, targets=[], controls=[], measures=[], matrix=None, *args, **kwargs):
    """Instructs the downstream plugin to execute a custom gate.

    `name` must be a non-empty string identifying the gate to be performed.
    `targets`, `constrols`, and `measures` must be iterables of qubits or
    singular qubits, representing respectively the set of qubits to operate
    on, the set of control qubits for controlled gates, and the set of
    qubits measured by the gate.  The `targets` and `controls` sets must
    not intersect. If specified, `matrix` must be a unitary matrix
    appropriately sized for the number of target qubits, specified as a
    row-major one-dimensional list of Python complex numbers. If no matrix
    is applicable or necessary for the custom gate, `None` can be used
    instead. The remainder of the arguments are passed to the constructor
    for `ArbData`; the resulting `ArbData` object is passed along with the
    gate for custom data.
    """
    if matrix is not None:
        matrix = Handle(raw.dqcs_mat_new(matrix))
    with QubitSet._to_raw(targets) as targets:
        with QubitSet._to_raw(controls) as controls:
            with QubitSet._to_raw(measures) as measures:
                if matrix is not None:
                    with matrix:
                        gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, int(matrix)))
                else:
                    gate = Handle(raw.dqcs_gate_new_custom(name, targets, controls, measures, 0))
                ArbData(*args, **kwargs)._to_raw(gate)
                with gate as gate:
                    self._pc(raw.dqcs_plugin_gate, gate)

def fredkin_gate(self, control, a, b, arb=None)

Instructs the downstream plugin to execute a Fredkin gate.

Expand source code

def fredkin_gate(self, control, a, b, arb=None):
    """Instructs the downstream plugin to execute a Fredkin gate."""
    self.unitary([a, b], [
        1.0, 0.0, 0.0, 0.0,
        0.0, 0.0, 1.0, 0.0,
        0.0, 1.0, 0.0, 0.0,
        0.0, 0.0, 0.0, 1.0,
    ], controls=[control], arb=arb) #@

def free(self, *qubits)

Instructs the downstream plugin to free the given qubits.

Expand source code

def free(self, *qubits):
    """Instructs the downstream plugin to free the given qubits."""
    with QubitSet._to_raw(*qubits) as qubits:
        self._pc(raw.dqcs_plugin_free, qubits)

def get_cycle(self)

Returns the current simulation time for the downstream plugin.

Expand source code

def get_cycle(self):
    """Returns the current simulation time for the downstream plugin."""
    return self._pc(raw.dqcs_plugin_get_cycle)

def get_cycles_between_measures(self, qubit)

Returns the number of cycles that were advanced between the latest measurement of the given downstream qubit and the one before.

Expand source code

def get_cycles_between_measures(self, qubit):
    """Returns the number of cycles that were advanced between the
    latest measurement of the given downstream qubit and the one before."""
    return self._pc(raw.dqcs_plugin_get_cycles_between_measures, qubit)

def get_cycles_since_measure(self, qubit)

Returns the number of cycles that have been advanced since the latest measurement of the given downstream qubit.

Expand source code

def get_cycles_since_measure(self, qubit):
    """Returns the number of cycles that have been advanced since the
    latest measurement of the given downstream qubit."""
    return self._pc(raw.dqcs_plugin_get_cycles_since_measure, qubit)

def get_measurement(self, qubit)

Returns the Measurement representing the latest measurement result for the given downstream qubit.

Expand source code

def get_measurement(self, qubit):
    """Returns the `Measurement` representing the latest measurement result
    for the given downstream qubit."""
    return Measurement._from_raw(Handle(self._pc(raw.dqcs_plugin_get_measurement, qubit)))

def h_gate(self, target, arb=None)

Instructs the downstream plugin to execute a Hadamard gate.

target is the targetted qubit.

Expand source code

def h_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a Hadamard gate.

    `target` is the targetted qubit."""
    x = 1 / math.sqrt(2.0)
    self.unitary([target], [x, x, x, -x], arb=arb)

def i_gate(self, target, arb=None)

Instructs the downstream plugin to execute an I gate.

target is the targetted qubit.

Expand source code

def i_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute an I gate.

    `target` is the targetted qubit."""
    self.unitary(target, [1.0, 0.0, 0.0, 1.0], arb=arb)

def measure(self, *qubits, basis='Z', arb=None)

Instructs the downstream plugin to measure the given qubits in the given basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only positional argument. The basis is always a keyword argument.

The basis can be 'X', 'Y', 'Z', or a 4-entry list of complex numbers representing a 2x2 matrix with the following semantics:

• the qubits are rotated by the inverse of the matrix
• a Z-axis measurement is performed on each qubit
• the qubits are rotated by the matrix

Expand source code

def measure(self, *qubits, basis='Z', arb=None):
    """Instructs the downstream plugin to measure the given qubits in the
    given basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only positional
    argument. The basis is always a keyword argument.

    The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
    numbers representing a 2x2 matrix with the following semantics:

     - the qubits are rotated by the inverse of the matrix
     - a Z-axis measurement is performed on each qubit
     - the qubits are rotated by the matrix
    """
    if isinstance(basis, str):
        basis = Handle(raw.dqcs_mat_basis({
            'X': raw.DQCS_BASIS_X,
            'Y': raw.DQCS_BASIS_Y,
            'Z': raw.DQCS_BASIS_Z,
        }[basis]))
    else:
        basis = Handle(raw.dqcs_mat_new(basis))
    if len(qubits) == 1 and not isinstance(qubits[0], int):
        qubits = list(qubits[0])
    with QubitSet._to_raw(qubits) as qubits:
        with basis as mat:
            gate = Handle(raw.dqcs_gate_new_measurement(qubits, mat))
            if arb is not None:
                if not isinstance(arb, ArbData):
                    raise TypeError('arb must be None or an instance of ArbData')
                arb._to_raw(gate)
            with gate as gate_raw:
                self._pc(raw.dqcs_plugin_gate, gate_raw)

def measure_x(self, *qubits, arb=None)

Instructs the downstream plugin to measure the given qubits in the X basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def measure_x(self, *qubits, arb=None):
    """Instructs the downstream plugin to measure the given qubits in the
    X basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.measure(*qubits, basis='X', arb=arb)

def measure_y(self, *qubits, arb=None)

Instructs the downstream plugin to measure the given qubits in the Y basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def measure_y(self, *qubits, arb=None):
    """Instructs the downstream plugin to measure the given qubits in the
    Y basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.measure(*qubits, basis='Y', arb=arb)

def measure_z(self, *qubits, arb=None)

Instructs the downstream plugin to measure the given qubits in the Z basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def measure_z(self, *qubits, arb=None):
    """Instructs the downstream plugin to measure the given qubits in the
    Z basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.measure(*qubits, basis='Z', arb=arb)

def mx90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a negative 90-degree X gate.

target is the targetted qubit.

Expand source code

def mx90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a negative 90-degree X
    gate.

    `target` is the targetted qubit."""
    self.rx_gate(target, -0.5 * math.pi, arb=arb)

def my90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a negative 90-degree Y gate.

target is the targetted qubit.

Expand source code

def my90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a negative 90-degree Y
    gate.

    `target` is the targetted qubit."""
    self.ry_gate(target, -0.5 * math.pi, arb=arb)

def mz90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a negative 90-degree Z gate, also known as an S-dagger gate.

target is the targetted qubit.

Expand source code

def mz90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a negative 90-degree Z
    gate, also known as an S-dagger gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, -0.5 * math.pi, arb=arb)

def prepare(self, *qubits, basis='Z', arb=None)

Instructs the downstream plugin to force the given qubits into the base state for the given basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only positional argument. The basis is always a keyword argument.

The basis can be 'X', 'Y', 'Z', or a 4-entry list of complex numbers representing a 2x2 matrix with the following semantics:

• the qubits are initialized to |0>
• the qubits are rotated by the matrix

Expand source code

def prepare(self, *qubits, basis='Z', arb=None):
    """Instructs the downstream plugin to force the given qubits into the
    base state for the given basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only positional
    argument. The basis is always a keyword argument.

    The basis can be `'X'`, `'Y'`, `'Z'`, or a 4-entry list of complex
    numbers representing a 2x2 matrix with the following semantics:

     - the qubits are initialized to |0>
     - the qubits are rotated by the matrix
    """
    if isinstance(basis, str):
        basis = Handle(raw.dqcs_mat_basis({
            'X': raw.DQCS_BASIS_X,
            'Y': raw.DQCS_BASIS_Y,
            'Z': raw.DQCS_BASIS_Z,
        }[basis]))
    else:
        basis = Handle(raw.dqcs_mat_new(basis))
    if len(qubits) == 1 and not isinstance(qubits[0], int):
        qubits = list(qubits[0])
    with QubitSet._to_raw(qubits) as qubits:
        with basis as mat:
            gate = Handle(raw.dqcs_gate_new_prep(qubits, mat))
            if arb is not None:
                if not isinstance(arb, ArbData):
                    raise TypeError('arb must be None or an instance of ArbData')
                arb._to_raw(gate)
            with gate as gate_raw:
                self._pc(raw.dqcs_plugin_gate, gate_raw)

def prepare_x(self, *qubits, arb=None)

Instructs the downstream plugin to force the given qubits into the base state for the X basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def prepare_x(self, *qubits, arb=None):
    """Instructs the downstream plugin to force the given qubits into the
    base state for the X basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.prepare(*qubits, basis='X', arb=arb)

def prepare_y(self, *qubits, arb=None)

Instructs the downstream plugin to force the given qubits into the base state for the Y basis.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def prepare_y(self, *qubits, arb=None):
    """Instructs the downstream plugin to force the given qubits into the
    base state for the Y basis.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.prepare(*qubits, basis='Y', arb=arb)

def prepare_z(self, *qubits, arb=None)

Instructs the downstream plugin to force the given qubits into the base state for the Z basis, being |0>.

This function takes either one or more qubits as its positional arguments, or an iterable of qubits as its first and only argument.

Expand source code

def prepare_z(self, *qubits, arb=None):
    """Instructs the downstream plugin to force the given qubits into the
    base state for the Z basis, being |0>.

    This function takes either one or more qubits as its positional
    arguments, or an iterable of qubits as its first and only argument.
    """
    self.prepare(*qubits, basis='Z', arb=arb)

def r_gate(self, target, theta, phi, lambd, arb=None)

Instructs the downstream plugin to perform a number of rotations at once.

target is the targetted qubit. theta, phi, and lambd are the angles in radians.

Expand source code

def r_gate(self, target, theta, phi, lambd, arb=None):
    """Instructs the downstream plugin to perform a number of rotations at
    once.

    `target` is the targetted qubit. `theta`, `phi`, and `lambd` are the
    angles in radians."""
    a = math.cos(0.5 * theta)
    b = math.sin(0.5 * theta)
    self.unitary(target, [
        a,
        -b * cmath.exp(1.0j * lambd),
        b * cmath.exp(1.0j * phi),
        a * cmath.exp(1.0j * (phi + lambd)),
    ], arb=arb)

def rx_gate(self, target, theta, arb=None)

Instructs the downstream plugin to perform an arbitrary X rotation.

target is the targetted qubit. theta is the angle in radians.

Expand source code

def rx_gate(self, target, theta, arb=None):
    """Instructs the downstream plugin to perform an arbitrary X rotation.

    `target` is the targetted qubit. `theta` is the angle in radians."""
    a = math.cos(0.5 * theta)
    b = -1.0j * math.sin(0.5 * theta)
    self.unitary(target, [a, b, b, a], arb=arb)

def ry_gate(self, target, theta, arb=None)

Instructs the downstream plugin to perform an arbitrary Y rotation.

target is the targetted qubit. theta is the angle in radians.

Expand source code

def ry_gate(self, target, theta, arb=None):
    """Instructs the downstream plugin to perform an arbitrary Y rotation.

    `target` is the targetted qubit. `theta` is the angle in radians."""
    a = math.cos(0.5 * theta)
    b = math.sin(0.5 * theta)
    self.unitary(target, [a, -b, b, a], arb=arb)

def rz_gate(self, target, theta, arb=None)

Instructs the downstream plugin to perform an arbitrary Z rotation.

target is the targetted qubit. theta is the angle in radians.

Expand source code

def rz_gate(self, target, theta, arb=None):
    """Instructs the downstream plugin to perform an arbitrary Z rotation.

    `target` is the targetted qubit. `theta` is the angle in radians."""
    a = cmath.exp(-0.5j * theta)
    b = cmath.exp(0.5j * theta)
    self.unitary(target, [a, 0.0, 0.0, b], arb=arb)

def s_gate(self, target, arb=None)

Instructs the downstream plugin to execute a 90-degree Z gate, also known as an S gate.

target is the targetted qubit.

Expand source code

def z90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a 90-degree Z gate, also
    known as an S gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, 0.5 * math.pi, arb=arb)

def sdag_gate(self, target, arb=None)

Instructs the downstream plugin to execute a negative 90-degree Z gate, also known as an S-dagger gate.

target is the targetted qubit.

Expand source code

def mz90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a negative 90-degree Z
    gate, also known as an S-dagger gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, -0.5 * math.pi, arb=arb)

def sqswap_gate(self, a, b, arb=None)

Instructs the downstream plugin to execute a square-root-of-swap gate.

a and b are the targetted qubits.

Expand source code

def sqswap_gate(self, a, b, arb=None):
    """Instructs the downstream plugin to execute a square-root-of-swap
    gate.

    `a` and `b` are the targetted qubits."""
    self.unitary([a, b], [
        1.0, 0.0,      0.0,      0.0,
        0.0, 0.5+0.5j, 0.5-0.5j, 0.0,
        0.0, 0.5-0.5j, 0.5+0.5j, 0.0,
        0.0, 0.0,      0.0,      1.0,
    ], arb=arb) #@

def swap_gate(self, a, b, arb=None)

Instructs the downstream plugin to execute a swap gate.

a and b are the targetted qubits.

Expand source code

def swap_gate(self, a, b, arb=None):
    """Instructs the downstream plugin to execute a swap gate.

    `a` and `b` are the targetted qubits."""
    self.unitary([a, b], [
        1.0, 0.0, 0.0, 0.0,
        0.0, 0.0, 1.0, 0.0,
        0.0, 1.0, 0.0, 0.0,
        0.0, 0.0, 0.0, 1.0,
    ], arb=arb) #@

def t_gate(self, target, arb=None)

Instructs the downstream plugin to execute a T gate.

target is the targetted qubit.

Expand source code

def t_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a T gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, 0.25 * math.pi, arb=arb)

def tdag_gate(self, target, arb=None)

Instructs the downstream plugin to execute a T-dagger gate.

target is the targetted qubit.

Expand source code

def tdag_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a T-dagger gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, -0.25 * math.pi, arb=arb)

def toffoli_gate(self, c1, c2, target, arb=None)

Instructs the downstream plugin to execute a Toffoli gate.

Expand source code

def toffoli_gate(self, c1, c2, target, arb=None):
    """Instructs the downstream plugin to execute a Toffoli gate."""
    self.unitary([target], [
        0.0, 1.0,
        1.0, 0.0,
    ], controls=[c1, c2], arb=arb) #@

def unitary(self, targets, matrix, controls=[], arb=None)

Instructs the downstream plugin to execute a unitary quantum gate.

targets must be a non-empty iterable of qubits or a single qubit, representing the qubit(s) targeted by the gate. matrix must be a unitary matrix appropriately sized for the number of target qubits, specified as a row-major one-dimensional list of Python complex numbers. controls optionally allows additional control qubits to be specified to make controlled gates; these qubits should NOT be reflected in the gate matrix. The matrix will automatically be extended by the downstream plugin, instead. The targets and controls sets must not intersect.

Expand source code

def unitary(self, targets, matrix, controls=[], arb=None):
    """Instructs the downstream plugin to execute a unitary quantum gate.

    `targets` must be a non-empty iterable of qubits or a single qubit,
    representing the qubit(s) targeted by the gate. `matrix` must be a
    unitary matrix appropriately sized for the number of target qubits,
    specified as a row-major one-dimensional list of Python complex
    numbers. `controls` optionally allows additional control qubits to be
    specified to make controlled gates; these qubits should NOT be
    reflected in the gate matrix. The matrix will automatically be extended
    by the downstream plugin, instead. The `targets` and `controls` sets
    must not intersect.
    """
    with QubitSet._to_raw(targets) as targets:
        with QubitSet._to_raw(controls) as controls:
            with Handle(raw.dqcs_mat_new(matrix)) as mat:
                gate = Handle(raw.dqcs_gate_new_unitary(targets, controls, mat))
                if arb is not None:
                    if not isinstance(arb, ArbData):
                        raise TypeError('arb must be None or an instance of ArbData')
                    arb._to_raw(gate)
                with gate as gate_raw:
                    self._pc(raw.dqcs_plugin_gate, gate_raw)

def x90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a 90-degree X gate.

target is the targetted qubit.

Expand source code

def x90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a 90-degree X gate.

    `target` is the targetted qubit."""
    self.rx_gate(target, 0.5 * math.pi, arb=arb)

def x_gate(self, target, arb=None)

Instructs the downstream plugin to execute an X gate.

target is the targetted qubit.

Expand source code

def x_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute an X gate.

    `target` is the targetted qubit."""
    self.rx_gate(target, math.pi, arb=arb)

def y90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a 90-degree Y gate.

target is the targetted qubit.

Expand source code

def y90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a 90-degree Y gate.

    `target` is the targetted qubit."""
    self.ry_gate(target, 0.5 * math.pi, arb=arb)

def y_gate(self, target, arb=None)

Instructs the downstream plugin to execute a Y gate.

target is the targetted qubit.

Expand source code

def y_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a Y gate.

    `target` is the targetted qubit."""
    self.ry_gate(target, math.pi, arb=arb)

def z90_gate(self, target, arb=None)

Instructs the downstream plugin to execute a 90-degree Z gate, also known as an S gate.

target is the targetted qubit.

Expand source code

def z90_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a 90-degree Z gate, also
    known as an S gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, 0.5 * math.pi, arb=arb)

def z_gate(self, target, arb=None)

Instructs the downstream plugin to execute a Z gate.

target is the targetted qubit.

Expand source code

def z_gate(self, target, arb=None):
    """Instructs the downstream plugin to execute a Z gate.

    `target` is the targetted qubit."""
    self.rz_gate(target, math.pi, arb=arb)

Inherited members

• Plugin:
  • critical
  • debug
  • error
  • fatal
  • info
  • log
  • note
  • random_float
  • random_long
  • run
  • start
  • trace
  • warn
  • warning

class JoinHandle

Returned by Plugin.start() to allow waiting for completion.

Expand source code

class JoinHandle(object):
    """Returned by `Plugin.start()` to allow waiting for completion."""
    def __init__(self, handle):
        super().__init__()
        if raw.dqcs_handle_type(int(handle)) != raw.DQCS_HTYPE_PLUGIN_JOIN:
            raise TypeError("Specified handle is not a JoinHandle")
        self._handle = handle

    def wait(self):
        """Waits for the associated plugin to finish executing."""
        if self._handle:
            raw.dqcs_plugin_wait(int(self._handle))
            self._handle.take()

Methods

def wait(self)

Waits for the associated plugin to finish executing.

Expand source code

def wait(self):
    """Waits for the associated plugin to finish executing."""
    if self._handle:
        raw.dqcs_plugin_wait(int(self._handle))
        self._handle.take()

class Operator (host_arb_ifaces=None, upstream_arb_ifaces=None)

Implements an operator plugin.

Operators sit between frontends and backends, allowing them to observe or modify the quantum gate and measurement streams between them.

The following functions MUST be implemented by the user:

• get_name() -> str

  Must return the name of the plugin implementation.

• get_author() -> str

  Must return the name of the plugin author.

• get_version() -> str

  Must return the plugin's version string.

The following functions MAY be implemented by the user:

• handle_init(cmds: [ArbCmd]) -> None

  Called by the simulator to initialize this plugin. The cmds parameter is passed a list of ArbCmds that the simulator wishes to associate with the plugin. If this function is not implemented, or if it is implemented but does not take an argument, the initialization ArbCmds are treated as regular host arbs (that is, they're passed to handle_host_<iface>_<oper>() if those functions do exist).

• handle_drop() -> None

  Called by the simulator when the simulation terminates.

• handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None

  Called when the upstream plugin needs more qubits. The qubits list specifies the (integer) references that will be used by future calls to refer to the qubits (thus, the length of the list is the number of qubits that are to be allocated). The cmds parameter is passed a list of ArbCmds that the upstream plugin wants to associate with the qubits.

  In almost all cases, this handler must call allocate() to forward the allocation downstream, in some cases modifying the number of qubits or command list. If the handler is not specified, the allocation is forwarded automatically.

• handle_free(qubits: [Qubit]) -> None

  Called when the upstream plugin doesn't need the specified qubits anymore.

  In almost all cases, this handler must call free() to forward the deallocation downstream. If the operator performs some kind of mapping function and/or modifies allocations, it may also need to modify the qubit list. If the handler is not specified, the deallocation is forwarded automatically.

• handle_unitary_gate( targets: [Qubit], matrix: [complex], arb: ArbData ) -> None

  Called when the upstream plugin wants to execute a non-controlled unitary gate. The gate is normally forwarded downstream using unitary(). If this callback is not defined, DQCsim attempts to call handle_controlled_gate() with an empty list for the control qubits. If that callback is also not defined, the gate is forwarded downstream automatically.

• handle_controlled_gate( targets: [Qubit], controls: [Qubit], matrix: [complex], arb: ArbData ) -> None

  Called when the upstream plugin wants to execute a controlled unitary gate, or a non-controlled gate if handle_unitary_gate() is not defined. The gate is normally forwarded downstream using unitary(). If this handler is not defined, the gate is forwarded downstream automatically.

• handle_measurement_gate( meas: [Qubit], basis: [complex], arb: ArbData ) -> [Measurement]

  Called when the upstream plugin wants to execute a basic measurement on the given set of qubits. The gate is normally forwarded downstream using measure(). If this handler is not defined, this is done automatically.

  The basis is a 2x2 matrix. The semantics are as follows:

  • rotate each qubit by the inverse/hermitian/conjugate transpose of the given matrix
  • do a Z measurement for each qubit
  • rotate each qubit by the given matrix

  The result of this handler must be that measurements for exactly those qubits specified in meas are forwarded upstream after all downstream gates issued by it finish executing. Usually this happens naturally by forwarding the measurement gate unchanged. However, operators that perform some kind of qubit remapping will need to do some extra work to perform the inverse mapping when the measurements are returned upstream. The recommended way to do this is by defining handle_measurement(), but it is also possible to return measurement results directly in the same way backends must do it (or to do a combination of the two). The latter might negatively impact simulation speed, though.

• handle_prepare_gate( target: [Qubit], basis: [complex], arb: ArbData ) -> None

  Called when the upstream plugin wants to reset the state for the given qubits. The gate is normally forwarded downstream using prepare(). If this handler is not defined, this is done automatically.

  The basis is a 2x2 matrix. The semantics are as follows:

  • initialize each qubit to |0>
  • rotate each qubit by the given matrix

• handle_<name>_gate( targets: [Qubit], controls: [Qubit], measures: [Qubit], matrix: [complex] or None, *args, **kwargs ) -> {Qubit: value} or None

  Called when a custom (named) gate must be performed. The targets, controls, measures, and matrix share the functionality of handle_controlled_gate() and handle_measurement_gate(), as does the return value for the latter. Custom gates also have an attached ArbData, of which the binary string list is passed to *args, and the JSON object is passed to **kwargs.

  The gate is normally forwarded downstream using custom_gate(). If this handler is not defined, this is done automatically.

• handle_measurement(meas: Measurement) -> [Measurement]

  Called when measurement data is received from the downstream plugin, allowing it to be modified before it is forwarded upstream. Modification includes not passing the measurement through (by returning an empty list), turning it into multiple measurements, changing the qubit reference to support qubit mapping, or just changing the measurement data itself to introduce errors or compensate for an earlier modification of the gatestream. If this handler is not defined, measurements are forwarded without modification.

• handle_advance(cycles: int) -> None

  Called when the upstream plugin wants to advance simulation time. Normally this handler calls advance(cycles) in response, which is the default behavior if it is not defined. The primary use of this handler is to send additional gates downstream to model errors.

• handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None

  Called when an ArbCmd is received from the upstream plugin or from the host with the interface and operation identifiers embedded in the name. That is, you don't have to do interface/operation identifier matching yourself; you just specify the operations that you support. The positional arguments are set to the list of binary strings attached to the ArbCmd, and **kwargs is set to the JSON object. If you return None, an empty ArbData object will be automatically generated for the response.

  If no handlers are available for the requested command interface, the command is forwarded downstream. If there is at least one, it is NOT forwarded downstream, even if the requested operation does not have a handler (an error will be reported instead).

Creates the plugin object.

Overriding __init__() in your implementation is fine, but you must call super().__init__() if you do this.

Among other things, this function auto-detects which arb interfaces are supported by your plugin by scanning for handle_host_*() and handle_upstream_*() implementations. However, this auto-detection will fail if there are underscores in any of the supported interface or operation IDs. In this case, you MUST override __init__() and call super().__init__(host_arb_ifaces, upstream_arb_ifaces), where the two arguments are lists of strings representing the supported interface IDs.

Expand source code

class Operator(GateStreamSource):
    """Implements an operator plugin.

    Operators sit between frontends and backends, allowing them to observe or
    modify the quantum gate and measurement streams between them.

    The following functions MUST be implemented by the user:

     - `get_name() -> str`

        Must return the name of the plugin implementation.

     - `get_author() -> str`

        Must return the name of the plugin author.

     - `get_version() -> str`

        Must return the plugin's version string.

    The following functions MAY be implemented by the user:

     - `handle_init(cmds: [ArbCmd]) -> None`

        Called by the simulator to initialize this plugin. The cmds parameter
        is passed a list of ArbCmds that the simulator wishes to associate with
        the plugin. If this function is not implemented, or if it is implemented
        but does not take an argument, the initialization ArbCmds are treated as
        regular host arbs (that is, they're passed to
        `handle_host_<iface>_<oper>()` if those functions do exist).

     - `handle_drop() -> None`

        Called by the simulator when the simulation terminates.

     - `handle_allocate(qubits: [Qubit], cmds: [ArbCmd]) -> None`

        Called when the upstream plugin needs more qubits. The qubits list
        specifies the (integer) references that will be used by future calls to
        refer to the qubits (thus, the length of the list is the number of
        qubits that are to be allocated). The cmds parameter is passed a list of
        ArbCmds that the upstream plugin wants to associate with the qubits.

        In almost all cases, this handler must call `allocate()` to forward the
        allocation downstream, in some cases modifying the number of qubits or
        command list. If the handler is not specified, the allocation is
        forwarded automatically.

     - `handle_free(qubits: [Qubit]) -> None`

        Called when the upstream plugin doesn't need the specified qubits
        anymore.

        In almost all cases, this handler must call `free()` to forward the
        deallocation downstream. If the operator performs some kind of mapping
        function and/or modifies allocations, it may also need to modify the
        qubit list. If the handler is not specified, the deallocation is
        forwarded automatically.

     - `handle_unitary_gate(
            targets: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to execute a non-controlled
        unitary gate. The gate is normally forwarded downstream using
        `unitary()`. If this callback is not defined, DQCsim attempts to call
        `handle_controlled_gate()` with an empty list for the control qubits.
        If that callback is also not defined, the gate is forwarded downstream
        automatically.

     - `handle_controlled_gate(
            targets: [Qubit],
            controls: [Qubit],
            matrix: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to execute a controlled unitary
        gate, or a non-controlled gate if `handle_unitary_gate()` is not
        defined. The gate is normally forwarded downstream using `unitary()`.
        If this handler is not defined, the gate is forwarded downstream
        automatically.

     - `handle_measurement_gate(
            meas: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> [Measurement]`

        Called when the upstream plugin wants to execute a basic measurement on
        the given set of qubits. The gate is normally forwarded downstream using
        `measure()`. If this handler is not defined, this is done automatically.

        The basis is a 2x2 matrix. The semantics are as follows:

         - rotate each qubit by the inverse/hermitian/conjugate transpose of the
           given matrix
         - do a Z measurement for each qubit
         - rotate each qubit by the given matrix

        The result of this handler must be that measurements for exactly those
        qubits specified in `meas` are forwarded upstream after all downstream
        gates issued by it finish executing. Usually this happens naturally by
        forwarding the measurement gate unchanged. However, operators that
        perform some kind of qubit remapping will need to do some extra work
        to perform the inverse mapping when the measurements are returned
        upstream. The recommended way to do this is by defining
        `handle_measurement()`, but it is also possible to return measurement
        results directly in the same way backends must do it (or to do a
        combination of the two). The latter might negatively impact simulation
        speed, though.

     - `handle_prepare_gate(
            target: [Qubit],
            basis: [complex],
            arb: ArbData
        ) -> None`

        Called when the upstream plugin wants to reset the state for the given
        qubits. The gate is normally forwarded downstream using `prepare()`. If
        this handler is not defined, this is done automatically.

        The basis is a 2x2 matrix. The semantics are as follows:

         - initialize each qubit to |0>
         - rotate each qubit by the given matrix

     -  `handle_<name>_gate(
            targets: [Qubit],
            controls: [Qubit],
            measures: [Qubit],
            matrix: [complex] or None,
            *args, **kwargs
        ) -> {Qubit: value} or None
        `

        Called when a custom (named) gate must be performed. The targets,
        controls, measures, and matrix share the functionality of
        `handle_controlled_gate()` and `handle_measurement_gate()`, as does the
        return value for the latter. Custom gates also have an attached
        `ArbData`, of which the binary string list is passed to `*args`, and
        the JSON object is passed to `**kwargs`.

        The gate is normally forwarded downstream using `custom_gate()`.
        If this handler is not defined, this is done automatically.

     - `handle_measurement(meas: Measurement) -> [Measurement]`

        Called when measurement data is received from the downstream plugin,
        allowing it to be modified before it is forwarded upstream. Modification
        includes not passing the measurement through (by returning an empty
        list), turning it into multiple measurements, changing the qubit
        reference to support qubit mapping, or just changing the measurement
        data itself to introduce errors or compensate for an earlier
        modification of the gatestream. If this handler is not defined,
        measurements are forwarded without modification.

     - `handle_advance(cycles: int) -> None`

        Called when the upstream plugin wants to advance simulation time.
        Normally this handler calls `advance(cycles)` in response, which is the
        default behavior if it is not defined. The primary use of this handler
        is to send additional gates downstream to model errors.

     - `handle_<host|upstream>_<iface>_<oper>(*args, **kwargs) -> ArbData or None`

        Called when an `ArbCmd` is received from the upstream plugin or from
        the host with the interface and operation identifiers embedded in the
        name. That is, you don't have to do interface/operation identifier
        matching yourself; you just specify the operations that you support.
        The positional arguments are set to the list of binary strings attached
        to the `ArbCmd`, and `**kwargs` is set to the JSON object. If you
        return `None`, an empty `ArbData` object will be automatically
        generated for the response.

        If no handlers are available for the requested command interface, the
        command is forwarded downstream. If there is at least one, it is NOT
        forwarded downstream, even if the requested operation does not have a
        handler (an error will be reported instead).
    """

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _route_allocate(self, state_handle, qubits_handle, cmds_handle):
        """Routes the allocate callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        cmds = ArbCmdQueue._from_raw(Handle(cmds_handle))
        self._cb(state_handle, 'handle_allocate', qubits, cmds)

    def _route_free(self, state_handle, qubits_handle):
        """Routes the free callback to user code."""
        qubits = QubitSet._from_raw(Handle(qubits_handle))
        self._cb(state_handle, 'handle_free', qubits)

    def _forward_unitary_gate(self, targets, controls, matrix, arb):
        self.unitary(targets, matrix, controls, arb)

    def _forward_measurement_gate(self, qubits, basis, arb):
        self.measure(qubits, basis=basis, arb=arb)

    def _forward_prepare_gate(self, qubits, basis, arb):
        self.prepare(qubits, basis=basis, arb=arb)

    def _route_gate(self, state_handle, gate_handle):
        """Routes the gate callback to user code."""

        typ = raw.dqcs_gate_type(gate_handle)
        name = None
        if raw.dqcs_gate_has_name(gate_handle):
            name = raw.dqcs_gate_name(gate_handle)

        # Forward gate types that don't have handlers as soon as possible.
        fast_forward = False
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            if raw.dqcs_gate_has_controls(gate_handle):
                fast_forward = not hasattr(self, 'handle_controlled_gate')
            else:
                fast_forward = ( #@
                    not hasattr(self, 'handle_unitary_gate')
                    and not hasattr(self, 'handle_controlled_gate'))
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            fast_forward = fast_forward and not hasattr(self, 'handle_measurement_gate')
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            fast_forward = fast_forward and not hasattr(self, 'handle_prepare_gate')
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            fast_forward = not hasattr(self, 'handle_{}_gate'.format(name))
        if fast_forward:
            raw.dqcs_plugin_gate(state_handle, gate_handle)
            return MeasurementSet._to_raw([]).take()

        # Convert from Rust domain to Python domain.
        targets = QubitSet._from_raw(Handle(raw.dqcs_gate_targets(gate_handle)))
        controls = QubitSet._from_raw(Handle(raw.dqcs_gate_controls(gate_handle)))
        measures = QubitSet._from_raw(Handle(raw.dqcs_gate_measures(gate_handle)))
        if raw.dqcs_gate_has_matrix(gate_handle):
            matrix = raw.dqcs_gate_matrix(gate_handle)
            matrix = raw.dqcs_mat_get(matrix)
        else:
            matrix = None

        # Route to the user's callback functions or execute the default
        # actions.
        data = ArbData._from_raw(Handle(gate_handle))
        measurements = []
        if typ == raw.DQCS_GATE_TYPE_UNITARY:
            try:
                if not controls and hasattr(self, 'handle_unitary_gate'):
                    self._cb(state_handle, 'handle_unitary_gate', targets, matrix, data)
                else:
                    self._cb(state_handle, 'handle_controlled_gate', targets, controls, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_unitary_gate', targets, controls, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_MEASUREMENT:
            try:
                measurements = self._cb(state_handle, 'handle_measurement_gate', measures, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_measurement_gate', measures, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_PREP:
            try:
                self._cb(state_handle, 'handle_prepare_gate', targets, matrix, data)
            except NotImplementedError:
                self._cb(state_handle, '_forward_prepare_gate', targets, matrix, data)
        elif typ == raw.DQCS_GATE_TYPE_CUSTOM:
            # Note that `handle_<name>_gate` must exist at this point,
            # otherwise it would have been forwarded earlier.
            cb_name = 'handle_{}_gate'.format(name)
            assert(hasattr(self, cb_name))
            measurements = self._cb(state_handle,
                cb_name, targets, controls, measures, matrix, *data._args, **data._json)
        else:
            raise NotImplementedError("unknown gate type")

        if measurements is None:
            measurements = []
        return MeasurementSet._to_raw(measurements).take()

    def _route_measurement(self, state_handle, measurement_handle):
        """Routes the measurement callback to user code."""
        measurement = Measurement._from_raw(Handle(measurement_handle))
        measurements = self._cb(state_handle, 'handle_measurement', measurement)
        if measurements is None:
            measurements = []
        elif isinstance(measurements, Measurement):
            measurements = [measurements]
        else:
            measurements = list(measurements)
        return MeasurementSet._to_raw(measurements).take()

    def _route_advance(self, state_handle, cycles):
        """Routes the advance callback to user code."""
        self._cb(state_handle, 'handle_advance', cycles)

    def _forward_upstream_arb(self, cmd):
        """Forwards an `ArbCmd` that originates from the upstream plugin
        further downstream."""
        return self.arb(cmd)

    def _route_upstream_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the upstream plugin."""
        return self._route_arb(state_handle, 'upstream', cmd_handle, '_forward_upstream_arb')

    def _to_pdef(self):
        """Creates a plugin definition handle for this plugin."""
        pdef = self._new_pdef(raw.DQCS_PTYPE_OPER)
        with pdef as pd:
            # Install Python callback handlers only when the user actually has
            # handlers defined for them. When they're not installed, events
            # will be forwarded downstream by Rust code, which is probably
            # significantly faster.
            handlers = set((
                member[0]
                for member in inspect.getmembers(
                    self, predicate=inspect.ismethod)
                if member[0].startswith('handle_')))
            if 'handle_allocate' in handlers:
                raw.dqcs_pdef_set_allocate_cb_pyfun(pd, self._cbent('allocate'))
            if 'handle_free' in handlers:
                raw.dqcs_pdef_set_free_cb_pyfun(pd, self._cbent('free'))
            if any(map(lambda name: name.endswith('_gate'), handlers)):
                raw.dqcs_pdef_set_gate_cb_pyfun(pd, self._cbent('gate'))
            if 'handle_measurement' in handlers:
                raw.dqcs_pdef_set_modify_measurement_cb_pyfun(pd, self._cbent('measurement'))
            if 'handle_advance' in handlers:
                raw.dqcs_pdef_set_advance_cb_pyfun(pd, self._cbent('advance'))
            if any(map(lambda name: name.startswith('handle_upstream_'), handlers)):
                raw.dqcs_pdef_set_upstream_arb_cb_pyfun(pd, self._cbent('upstream_arb'))
        return pdef

Ancestors

• GateStreamSource
• Plugin

Inherited members

• GateStreamSource:
  • advance
  • allocate
  • arb
  • cnot_gate
  • critical
  • custom_gate
  • debug
  • error
  • fatal
  • fredkin_gate
  • free
  • get_cycle
  • get_cycles_between_measures
  • get_cycles_since_measure
  • get_measurement
  • h_gate
  • i_gate
  • info
  • log
  • measure
  • measure_x
  • measure_y
  • measure_z
  • mx90_gate
  • my90_gate
  • mz90_gate
  • note
  • prepare
  • prepare_x
  • prepare_y
  • prepare_z
  • r_gate
  • random_float
  • random_long
  • run
  • rx_gate
  • ry_gate
  • rz_gate
  • s_gate
  • sdag_gate
  • sqswap_gate
  • start
  • swap_gate
  • t_gate
  • tdag_gate
  • toffoli_gate
  • trace
  • unitary
  • warn
  • warning
  • x90_gate
  • x_gate
  • y90_gate
  • y_gate
  • z90_gate
  • z_gate

class Plugin (host_arb_ifaces=None, upstream_arb_ifaces=None)

Represents a plugin implementation. Must be subclassed; use Frontend, Operator, or Backend instead.

Creates the plugin object.

Overriding __init__() in your implementation is fine, but you must call super().__init__() if you do this.

Among other things, this function auto-detects which arb interfaces are supported by your plugin by scanning for handle_host_*() and handle_upstream_*() implementations. However, this auto-detection will fail if there are underscores in any of the supported interface or operation IDs. In this case, you MUST override __init__() and call super().__init__(host_arb_ifaces, upstream_arb_ifaces), where the two arguments are lists of strings representing the supported interface IDs.

Expand source code

class Plugin(object):
    """Represents a plugin implementation. Must be subclassed; use Frontend,
    Operator, or Backend instead."""

    #==========================================================================
    # Launching the plugin
    #==========================================================================
    def __init__(self, host_arb_ifaces=None, upstream_arb_ifaces=None):
        """Creates the plugin object.

        Overriding `__init__()` in your implementation is fine, but you must
        call `super().__init__()` if you do this.

        Among other things, this function auto-detects which arb interfaces are
        supported by your plugin by scanning for `handle_host_*()` and
        `handle_upstream_*()` implementations. However, this auto-detection
        will fail if there are underscores in any of the supported interface or
        operation IDs. In this case, you MUST override `__init__()` and call
        `super().__init__(host_arb_ifaces, upstream_arb_ifaces)`, where the two
        arguments are lists of strings representing the supported interface
        IDs.
        """
        super().__init__()

        # CPython's trace functionality is used by kcov to test Python
        # coverage. That's great and all, but when we call into the C API and
        # the C API calls back into us (from another thread, no less), we're in
        # a new Python context that doesn't have the trace function set. We
        # need to fix this manually when we get such a callback. This is done
        # by the functions generated by `self._cbent()` if this is not `None`.
        # If no trace function is specified, there should be no runtime
        # overhead.
        self._trace_fn = sys.gettrace()

        # This local is used to store whether we're inside a user-defined
        # callback and, if so, what the value of the `dqcs_plugin_type_t`
        # handle is. This allows us to A) check that the user isn't calling
        # `dqcs_plugin_*` functions taking such a state without having a valid
        # one, and B) abstract the requirement of passing the plugin handle
        # around away from the user. Note that there are no thread-safety
        # problems with this, since the C API plugin callbacks are all called
        # from the same thread.
        self._state_handle = None

        # Stores whether this plugin has been instantiated. This can only be
        # done once, otherwise we could get plugin callbacks from multiple
        # plugin instances/threads, which would mess up the "thread-safety"
        # of `_state_handle`.
        self._started = False

        # Configure the supported arb interfaces.
        self._arb_interfaces = {}
        if host_arb_ifaces is not None:
            self._arb_interfaces['host'] = set(host_arb_ifaces)
        if upstream_arb_ifaces is not None:
            self._arb_interfaces['upstream'] = set(upstream_arb_ifaces)
        for source in ['host', 'upstream']:
            if source in self._arb_interfaces:
                continue

            # Try to auto-detect arb interfaces for this source.
            ifaces = set()
            for mem, _ in inspect.getmembers(self, predicate=inspect.ismethod):
                if not mem.startswith('handle_{}_'.format(source)):
                    continue
                s = mem.split('_')
                if len(s) > 4:
                    raise RuntimeError(
                        "cannot auto-detect interface and operation ID for arb "
                        "handler {}(), please pass the '{}_arb_ifaces' argument "
                        "to __init__() to specify the supported interfaces manually"
                        "".format(mem, source))
                elif len(s) < 4:
                    continue
                ifaces.add(s[2])
            self._arb_interfaces[source] = ifaces

    def _check_run(self, simulator):
        """Checks that the plugin is ready to be started and figures out the
        simulator address."""
        if not hasattr(self, '_state_handle'):
            raise RuntimeError("It looks like you've overridden __init__ and forgot to call super().__init__(). Please fix!")
        if self._started:
            raise RuntimeError("Plugin has been started before. Make a new instance!")
        if simulator is None:
            if len(sys.argv) != 2:
                print("Usage: [python3] <script> <simulator-address>", file=sys.stderr)
                print("Note: you should be calling this Python script with DQCsim!", file=sys.stderr)
                sys.exit(1)
            simulator = sys.argv[1]
        return simulator

    def run(self, simulator=None):
        """Instantiates and runs the plugin.

        simulator represents the DQCsim address that the plugin must connect to
        when initializing. It is usually passed as the first argument to the
        plugin process; therefore, if it is not specified, it is taken directly
        from sys.argv."""
        simulator = self._check_run(simulator)
        with self._to_pdef() as pdef:
            self._started = True
            raw.dqcs_plugin_run(pdef, simulator)

    def start(self, simulator=None):
        """Instantiates and starts the plugin.

        This has the same behavior as run(), except the plugin is started in a
        different thread, so it returns immediately. The returned object is a
        JoinHandle, which contains a wait() method that can be used to wait
        until the plugin finishes executing. Alternatively, if this is not
        done, the plugin thread will (try to) survive past even the main
        thread.

        Note that the JoinHandle can NOT be transferred to a different
        thread!"""
        simulator = self._check_run(simulator)
        with self._to_pdef() as pdef:
            handle = Handle(raw.dqcs_plugin_start(pdef, simulator))
        self._started = True
        return JoinHandle(handle)

    #==========================================================================
    # API functions operating on plugin state
    #==========================================================================
    def _pc(self, plugin_fn, *args):
        """Use this to call dqcs_plugin functions that take a plugin state."""
        if self._state_handle is None:
            raise RuntimeError("Cannot call plugin operator outside of a callback")
        return plugin_fn(self._state_handle, *args)

    def random_float(self):
        """Produces a random floating point value between 0 (inclusive) and 1
        (exclusive).

        This function is guaranteed to return the same result every time as
        long as the random seed allocated to us by DQCsim stays the same. This
        allows simulations to be reproduced using a reproduction file. Without
        such a reproduction file or user-set seed, this is of course properly
        (pseudo)randomized."""
        return self._pc(raw.dqcs_plugin_random_f64)

    def random_long(self):
        """Produces a random 64-bit unsigned integer.

        This function is guaranteed to return the same result every time as
        long as the random seed allocated to us by DQCsim stays the same. This
        allows simulations to be reproduced using a reproduction file. Without
        such a reproduction file or user-set seed, this is of course properly
        (pseudo)randomized."""
        return self._pc(raw.dqcs_plugin_random_u64)

    #==========================================================================
    # Logging functions
    #==========================================================================
    def _log(self, level, msg, *args, **kwargs):
        # NOTE: we don't need the state handle technically, but this ensures
        # that we're in the right thread.
        if self._state_handle is None:
            raise RuntimeError("Cannot call plugin operator outside of a callback")
        msg = str(msg)
        if args or kwargs:
            msg = msg.format(*args, **kwargs)
        frame = inspect.currentframe().f_back.f_back
        module = frame.f_globals.get('__name__', '?')
        fname = frame.f_globals.get('__file__', '?')
        lineno = frame.f_lineno
        raw.dqcs_log_raw(level, module, fname, lineno, msg)

    def log(self, level, msg, *args, **kwargs):
        """Logs a message with the specified loglevel to DQCsim.

        If any additional positional or keyword arguments are specified, the
        message is formatted using `str.format()`. Otherwise, `str()` is
        applied to the message."""
        # NOTE: this level of indirection is needed to make function name,
        # filename, and line number metadata correct.
        if not isinstance(level, Loglevel):
            raise TypeError('level must be a Loglevel')
        self._log(level, msg, *args, **kwargs)

    def trace(self, msg, *args, **kwargs):
        """Convenience function for logging trace messages. See `log()`."""
        self._log(Loglevel.TRACE, msg, *args, **kwargs)

    def debug(self, msg, *args, **kwargs):
        """Convenience function for logging debug messages. See `log()`."""
        self._log(Loglevel.DEBUG, msg, *args, **kwargs)

    def info(self, msg, *args, **kwargs):
        """Convenience function for logging info messages. See `log()`."""
        self._log(Loglevel.INFO, msg, *args, **kwargs)

    def note(self, msg, *args, **kwargs):
        """Convenience function for logging note messages. See `log()`."""
        self._log(Loglevel.NOTE, msg, *args, **kwargs)

    def warn(self, msg, *args, **kwargs):
        """Convenience function for logging warning messages. See `log()`."""
        self._log(Loglevel.WARN, msg, *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        """Convenience function for logging error messages. See `log()`."""
        self._log(Loglevel.ERROR, msg, *args, **kwargs)

    def fatal(self, msg, *args, **kwargs):
        """Convenience function for logging fatal messages. See `log()`."""
        self._log(Loglevel.FATAL, msg, *args, **kwargs)

    warning = warn
    critical = fatal

    #==========================================================================
    # Callback helpers
    #==========================================================================
    def _cbent(self, router):
        """All callbacks from the C API world are generated by this function.
        Normally it just returns a "router" function, which determines which
        user callback(s) should be called for the given C API callback. If a
        trace function is set however (this is done by kcov for Python code
        coverage), we need to set this trace function every time we're called
        from the C domain for it to work. In that case, a proxy function is
        returned that first sets the trace function and then calls the actual
        router."""
        router_fn = getattr(self, '_route_' + router)
        if self._trace_fn:
            def traced_router_fn(*args, **kwargs):
                sys.settrace(self._trace_fn) # no_kcoverage
                return router_fn(*args, **kwargs) # no_kcoverage
            return traced_router_fn
        return router_fn # no_kcoverage

    def _cb(self, state_handle, name, *args, **kwargs):
        """This function is used to call into the Python callbacks specified by
        the user. It saves the state handle in a local variable so the user
        code doesn't have to carry it around everywhere (and can't accidentally
        break it), then calls the named function is it exists. It also logs the
        user's stack traces with trace loglevel for debugging; if it wouldn't,
        this information would be lost (only the str() representation of the
        exception is passed on to the C API layer). If the user did not
        implement the callback, a NotImplementedError is returned, which may or
        may not be handled by the router function if there is an alternative
        call or a sane default operation."""
        if hasattr(self, name):
            if self._state_handle is not None:
                raise RuntimeError("Invalid state, recursive callback")
            self._state_handle = state_handle
            try:
                try:
                    return getattr(self, name)(*args, **kwargs)
                except Exception as e:
                    for line in traceback.format_exc().split('\n'):
                        self.trace(line)
                    raise
            finally:
                self._state_handle = None
        raise NotImplementedError("Python plugin doesn't implement {}(), which is a required function!".format(name))

    def _route_initialize(self, state_handle, init_cmds_handle):
        """Routes the initialization callback to the user's implementation, if
        there is any. If there isn't, try to route the initialization `ArbCmd`s
        as if they're normal host arbs."""
        cmds = ArbCmdQueue._from_raw(Handle(init_cmds_handle))
        try:
            self._cb(state_handle, 'handle_init', cmds)
        except NotImplementedError:
            for cmd in cmds:
                self._route_converted_arb(state_handle, 'host', cmd)

    def _route_drop(self, state_handle):
        """Routes the drop callback to the user's implementation, if there is
        any."""
        try:
            self._cb(state_handle, 'handle_drop')
        except NotImplementedError:
            pass

    def _route_converted_arb(self, state_handle, source, cmd, forward_fn=None):
        """Routes an `ArbCmd` originating from the given source, which must be
        'upstream' or 'host'. `cmd` should already have been converted to an
        `ArbCmd` object (instead of being passed as a handle)."""
        if cmd.iface not in self._arb_interfaces.get(source, {}):
            if forward_fn is not None:
                return self._cb(state_handle, forward_fn, cmd)
            return ArbData()
        try:
            result = self._cb(state_handle,
                'handle_{}_{}_{}'.format(source, cmd.iface, cmd.oper),
                *cmd._args, **cmd._json
            ) #@
            if result is None:
                result = ArbData()
            elif not isinstance(result, ArbData):
                raise TypeError("User implementation of host arb should return None or ArbData but returned {}".format(type(result)))
            return result
        except NotImplementedError:
            raise ValueError("Invalid operation ID {} for interface ID {}".format(cmd.oper, cmd.iface))

    def _route_arb(self, state_handle, source, cmd_handle, forward_fn=None):
        """Routes an `ArbCmd` originating from the given source, which must be
        'upstream' or 'host'. Takes an integer handle to an `ArbCmd` and
        returns an integer handle to an `ArbData` (that is, they're not wrapped
        in `Handle` objects)."""
        cmd = ArbCmd._from_raw(Handle(cmd_handle))
        result = self._route_converted_arb(state_handle, source, cmd, forward_fn)
        return result._to_raw().take()

    def _route_host_arb(self, state_handle, cmd_handle):
        """Routes an `ArbCmd` that originates from the host."""
        return self._route_arb(state_handle, 'host', cmd_handle)

    def _new_pdef(self, typ):
        """Constructs a pdef `Handle` configured with appropriate metadata and
        the callbacks common to all plugins."""
        pdef = Handle(raw.dqcs_pdef_new(
            typ,
            self._cb(None, 'get_name'),
            self._cb(None, 'get_author'),
            self._cb(None, 'get_version')
        )) #@
        with pdef as pd:
            raw.dqcs_pdef_set_initialize_cb_pyfun(pd, self._cbent('initialize'))
            raw.dqcs_pdef_set_drop_cb_pyfun(pd, self._cbent('drop'))
            raw.dqcs_pdef_set_host_arb_cb_pyfun(pd, self._cbent('host_arb'))
        return pdef

Subclasses

• Backend
• GateStreamSource

Methods

def critical(self, msg, *args, **kwargs)

Convenience function for logging fatal messages. See log().

Expand source code

def fatal(self, msg, *args, **kwargs):
    """Convenience function for logging fatal messages. See `log()`."""
    self._log(Loglevel.FATAL, msg, *args, **kwargs)

def debug(self, msg, *args, **kwargs)

Convenience function for logging debug messages. See log().

Expand source code

def debug(self, msg, *args, **kwargs):
    """Convenience function for logging debug messages. See `log()`."""
    self._log(Loglevel.DEBUG, msg, *args, **kwargs)

def error(self, msg, *args, **kwargs)

Convenience function for logging error messages. See log().

Expand source code

def error(self, msg, *args, **kwargs):
    """Convenience function for logging error messages. See `log()`."""
    self._log(Loglevel.ERROR, msg, *args, **kwargs)

def fatal(self, msg, *args, **kwargs)

Convenience function for logging fatal messages. See log().

Expand source code

def fatal(self, msg, *args, **kwargs):
    """Convenience function for logging fatal messages. See `log()`."""
    self._log(Loglevel.FATAL, msg, *args, **kwargs)

def info(self, msg, *args, **kwargs)

Convenience function for logging info messages. See log().

Expand source code

def info(self, msg, *args, **kwargs):
    """Convenience function for logging info messages. See `log()`."""
    self._log(Loglevel.INFO, msg, *args, **kwargs)

def log(self, level, msg, *args, **kwargs)

Logs a message with the specified loglevel to DQCsim.

If any additional positional or keyword arguments are specified, the message is formatted using str.format(). Otherwise, str() is applied to the message.

Expand source code

def log(self, level, msg, *args, **kwargs):
    """Logs a message with the specified loglevel to DQCsim.

    If any additional positional or keyword arguments are specified, the
    message is formatted using `str.format()`. Otherwise, `str()` is
    applied to the message."""
    # NOTE: this level of indirection is needed to make function name,
    # filename, and line number metadata correct.
    if not isinstance(level, Loglevel):
        raise TypeError('level must be a Loglevel')
    self._log(level, msg, *args, **kwargs)

def note(self, msg, *args, **kwargs)

Convenience function for logging note messages. See log().

Expand source code

def note(self, msg, *args, **kwargs):
    """Convenience function for logging note messages. See `log()`."""
    self._log(Loglevel.NOTE, msg, *args, **kwargs)

def random_float(self)

Produces a random floating point value between 0 (inclusive) and 1 (exclusive).

This function is guaranteed to return the same result every time as long as the random seed allocated to us by DQCsim stays the same. This allows simulations to be reproduced using a reproduction file. Without such a reproduction file or user-set seed, this is of course properly (pseudo)randomized.

Expand source code

def random_float(self):
    """Produces a random floating point value between 0 (inclusive) and 1
    (exclusive).

    This function is guaranteed to return the same result every time as
    long as the random seed allocated to us by DQCsim stays the same. This
    allows simulations to be reproduced using a reproduction file. Without
    such a reproduction file or user-set seed, this is of course properly
    (pseudo)randomized."""
    return self._pc(raw.dqcs_plugin_random_f64)

def random_long(self)

Produces a random 64-bit unsigned integer.

This function is guaranteed to return the same result every time as long as the random seed allocated to us by DQCsim stays the same. This allows simulations to be reproduced using a reproduction file. Without such a reproduction file or user-set seed, this is of course properly (pseudo)randomized.

Expand source code

def random_long(self):
    """Produces a random 64-bit unsigned integer.

    This function is guaranteed to return the same result every time as
    long as the random seed allocated to us by DQCsim stays the same. This
    allows simulations to be reproduced using a reproduction file. Without
    such a reproduction file or user-set seed, this is of course properly
    (pseudo)randomized."""
    return self._pc(raw.dqcs_plugin_random_u64)

def run(self, simulator=None)

Instantiates and runs the plugin.

simulator represents the DQCsim address that the plugin must connect to when initializing. It is usually passed as the first argument to the plugin process; therefore, if it is not specified, it is taken directly from sys.argv.

Expand source code

def run(self, simulator=None):
    """Instantiates and runs the plugin.

    simulator represents the DQCsim address that the plugin must connect to
    when initializing. It is usually passed as the first argument to the
    plugin process; therefore, if it is not specified, it is taken directly
    from sys.argv."""
    simulator = self._check_run(simulator)
    with self._to_pdef() as pdef:
        self._started = True
        raw.dqcs_plugin_run(pdef, simulator)

def start(self, simulator=None)

Instantiates and starts the plugin.

This has the same behavior as run(), except the plugin is started in a different thread, so it returns immediately. The returned object is a JoinHandle, which contains a wait() method that can be used to wait until the plugin finishes executing. Alternatively, if this is not done, the plugin thread will (try to) survive past even the main thread.

Note that the JoinHandle can NOT be transferred to a different thread!

Expand source code

def start(self, simulator=None):
    """Instantiates and starts the plugin.

    This has the same behavior as run(), except the plugin is started in a
    different thread, so it returns immediately. The returned object is a
    JoinHandle, which contains a wait() method that can be used to wait
    until the plugin finishes executing. Alternatively, if this is not
    done, the plugin thread will (try to) survive past even the main
    thread.

    Note that the JoinHandle can NOT be transferred to a different
    thread!"""
    simulator = self._check_run(simulator)
    with self._to_pdef() as pdef:
        handle = Handle(raw.dqcs_plugin_start(pdef, simulator))
    self._started = True
    return JoinHandle(handle)

def trace(self, msg, *args, **kwargs)

Convenience function for logging trace messages. See log().

Expand source code

def trace(self, msg, *args, **kwargs):
    """Convenience function for logging trace messages. See `log()`."""
    self._log(Loglevel.TRACE, msg, *args, **kwargs)

def warn(self, msg, *args, **kwargs)

Convenience function for logging warning messages. See log().

Expand source code

def warn(self, msg, *args, **kwargs):
    """Convenience function for logging warning messages. See `log()`."""
    self._log(Loglevel.WARN, msg, *args, **kwargs)

def warning(self, msg, *args, **kwargs)

Convenience function for logging warning messages. See log().

Expand source code

def warn(self, msg, *args, **kwargs):
    """Convenience function for logging warning messages. See `log()`."""
    self._log(Loglevel.WARN, msg, *args, **kwargs)

class plugin (name, author, version)

Decorator for Plugin class implementations to take some of the boilerplate code away.

Expand source code

class plugin(object):
    """Decorator for Plugin class implementations to take some of the
    boilerplate code away."""

    def __init__(self, name, author, version):
        super().__init__()
        self._name = name
        self._author = author
        self._version = version

    def __call__(self, cls):
        setattr(cls, "get_name", lambda _: self._name)
        setattr(cls, "get_author", lambda _: self._author)
        setattr(cls, "get_version", lambda _: self._version)
        return cls