Make your own QPU

In our Framework, a QPU class should inherit from qat.qpus.QPUHandler which defines the following methods:

  • Method submit_job() required, this method takes a Job and result a Result

  • Method get_specs() optional, this method does not take any parameters and returns an qat.core.HardwareSpecs. The output of this function describes that hardware capabilities (e.g. maximal number of qubits, topology, gate set) and could be used by compilers to adapt a job for the current QPU (more information on the qat.core.HardwareSpecs can be found in the in the plugin section). If not implemented, a default qat.core.HardwareSpecs object will be returned

  • Some additional methods can be added to better integrate your QPU within the Qaptiva Appliance

Raising exception in the code

A QPU can raise QPUException. A QPU can be accessed remotely (using Qaptiva Access or by starting the QPU in server mode). This exception can be serialized and re-raised on the client side

A QPUException can be raised using assert_qpu()

from qat.core.assertion import assert_qpu

# If my_condition() returns False:  raises a QPUException
# If my_condition() returns True: do nothing
assert_qpu(my_condition(), "Error message")

Warning

If your QPU implements its own constructor, please ensure the parent constructor is called

from qat.core.qpu import QPUHandler

class MyQPU(QPUHandler):
    def __init__(self, parameter):
        super().__init__()  # Please ensure parents constructor is called
        self._parameter = parameter

    def submit_job(self, job):
        ...

Method submit_job

The submit_job() method is the only required method. This function takes one or two parameters:

  • a required argument of type Job. This argument defines what to execute and what to return. Attributes of this class are defined on the job page

  • an optional argument of type dict[str, str] (optional means that this argument can be removed, i.e. def submit_job(self, job): ... is a valid method), this argument containing the meta-data of the Batch

    A good practice

    Meta-data could be used to override temporarly the parameter used to instantiate a QPU (the name of the QPU corresponds to the key of this dictionary, and the associated value contains the value to override). Multiple components (like plugins) can interact with this QPU by sending Batch. Implementing this “good practise” can improve the interaction between these components and the QPU.

    For instance, to override the parameter argument defined in the example above, the submit function shall look like:

    import json
    from qat.core.qpu import QPUHandler
    
    class MyQPU(QPUHandler):
        def __init__(self, parameter):
            super().__init__()  # Please ensure parents constructor is called
            self._parameter = parameter
            self._default_options = {"parameter": parameter}
    
        def _override(self, options):
            if "parameter" in options:
                self._parameter = options["parameter"]
    
        def submit_job(self, job, meta_data=None):
            # Override "self._parameter"
            if meta_data:
                options = json.loads(meta_data.get("MyQPU", "{}"))
                self._override(options)
    
            ... # Perform execution
    
            self._override(self._default_options)
            return ...  # Return result
    

This method is often splitted in 3 main steps:

  • Step 1: ensures the quantum job can be executed “as it is”. This steps ensure the number of shot is valid, the processing type (e.g. circuit, schedule, etc.) is valid regarding the QPU, the final measurement is correct, etc.

  • Step 2: exceutes the quantum job nbshots times (0 shots corresponds to the maximal number of shots supported by the QPU - this value is valid)

    • if the quantum job contains a circuit, gates are executed one by one. Method iterate_simple() can be used to list all the gates composing the circuit

    • if the quantum jobs contains a scheduler, the quantum simulation should be executed accordingly

  • Step 3: builds and cleans result. Samples are added using the add_sample() method, the average value of the observable is set by updating the attribute value of Result. In sample mode, the result can be clean-up using function aggregate_data()

The card underneath provides a skeleton for a QPU executing quantum circuits in sample mode

A skeleton for a custom QPU
from qat.core import Result
from qat.core.qpu import QPUHandler
from qat.core.assertion import assert_qpu
from qat.core.wrappers.result import aggregate_data

MAX_NB_SHOTS = 1024


class QPUSkeleton(QPUHandler):
    """
    Skeleton of a custom QPU

    This skeleton execute a circuit, by running gates one by one. This skeleton also returns
    a result
    """
    def submit_job(self, job) -> Result:
        """
        Execute a job
        The job should contain a circuit (neither a analog job, nor a annealing job)

        Args:
            job: the job to execute

        Returns:
            Result: result of the computation
        """
        # Check job
        nb_shots = job.nbshots or MAX_NB_SHOTS

        assert_qpu(job.circuit is not None, "This skeleton can only execute a circuit job")
        assert_qpu(0 < nb_shots <= MAX_NB_SHOTS, "Invalid number of shots")
        assert_qpu(job.type == ProcessingType.SAMPLE, "This QPU does not support OBSERVABLE measurement")

        # Initialize result
        result = Result()

        # Measured qubits: qubits which should be measured at the end
        # The "qubits" attribute is either:
        #   - a list of qubits (list of integer)
        #   - None (all qubits should be measured)
        measured_qubits = job.qubits or list(range(job.circuit.nbqbits))

        # Execute the circuit several time
        for shot in range(nb_shots):
            for gate in job.circuit.iterate_simple():
                ... # TODO: execute gate

            state = ... # TODO: measure qubits listed in "measured_qubits"
            result.add_sample(state)

        # Aggregate data
        # If set to True, the output will be compressed. The list of sample will be caster into a shorter
        # list of tuple [state, probability]
        if job.aggregate_data:
            # The "threshold" parameter is used to remove state having a probability lower than this value
            # are removed
            aggregate_data(result, threshold=job.threshold)

        # Return result
        return result

Method get_specs

This method does not take any parameter and returns a description of the hardware. The output of this function is used by compilers to update a quantum jobs, to make it executable by the QPU.

This hardware description defines:

  • the number of qubits composing the QPU

  • the topology of the hardware

  • a gate set

  • if the QPU supports SAMPLE measurements or the OBSERVABLE measurements

  • a description of the hardware

This method is already implemented by QPUHandler but could be overrided. If not implemented, the QPU is assumed to support any size of quantum jobs (in term of qubits), all to all interaction, can execute any gate, support SAMPLE and OBSERVABLE measurements

from qat.core import HardwareSpecs
from qat.core.qpu import QPUHandler

def MyQPU(QPUHandler):
    def get_specs(self):
        """
        Returns a description of the hardware
        """
        return HardwareSpecs(...)

Class HardwareSpecs is detailed on this in the plugin section