qmio 0.1.1

Helper python module to interact with the different backends available in Qmio.

Table of Contents

1. Intro
2. Estructure
3. Qmio module
   1. qmio.py
      1. services.py
      2. backends.py
      3. clients.py
4. tests
   1. test_clients
   2. test_backends
   3. test_services
   4. test_qmio

Intro

Proyecto de Integración del computador cuántico del cesga. Ejemplo de uso en qmio.py

Estructure

.
├── config
│   ├── development.py
│   ├── __init__.py
│   └── production.py
├── LICENSE
├── main.org
├── main.pdf
├── main.tex
├── notes.org
├── qmio
│   ├── backends.py
│   ├── clients.py
│   ├── __init__.py
│   ├── qmio.py
│   └── services.py
├── README.md
├── requirements-dev.txt
├── requirements.txt
├── setup.cfg
├── setup.py
└── user_test.py

3 directories, 19 files

Qmio module

qmio.py

from qmio.services import QmioRuntimeService

program = """OPENQASM 2.0;
    include "qelib1.inc";
    qreg q[4];
    creg c[4];
    rx(0.01) q[0];
    measure q[0]->c[0];
    measure q[1]->c[1];
    measure q[2]->c[2];
    measure q[3]->c[3];
    """

# config = '{"$type": "<class \'qat.purr.compiler.config.CompilerConfig\'>", "$data": {"repeats": 1, "repetition_period": null, "results_format": {"$type": "<class \'qat.purr.compiler.config.QuantumResultsFormat\'>", "$data": {"format": {"$type": "<enum \'qat.purr.compiler.config.InlineResultsProcessing\'>", "$value": 1}, "transforms": {"$type": "<enum \'qat.purr.compiler.config.ResultsFormatting\'>", "$value": 3}}}, "metrics": {"$type": "<enum \'qat.purr.compiler.config.MetricsType\'>", "$value": 6}, "active_calibrations": [], "optimizations": null}}'

service = QmioRuntimeService()

shots = 100

# You can user with explicit connect and disconnect
# backend = service.backend(name="qpu")
# backend.connect()
# result = backend.run(program, shots=shots)
# backend.disconnect()

# Recommended usage
with service.backend(name="qpu") as backend:
    result = backend.run(program=program, shots=shots)

print(result)

services.py

from qmio.backends import QPUBackend

class QmioRuntimeService():

    def backend(self, name):
        if name == "qpu":
            return QPUBackend()
        else:
            raise ValueError(f"Backend desconocido: {name}")

backends.py

from qmio.clients import ZMQClient
import json

def _config_build(shots: int, repetition_period=None, optimizations=None):
    config = {
        "$type": "<class \'qat.purr.compiler.config.CompilerConfig\'>",
        "$data": {
            "repeats": shots,
            "repetition_period": repetition_period,
            "results_format": {
                "$type": "<class \'qat.purr.compiler.config.QuantumResultsFormat\'>",
                "$data": {
                    "format": {
                        "$type": "<enum \'qat.purr.compiler.config.InlineResultsProcessing\'>",
                        "$value": 1
                    },
                    "transforms": {
                        "$type": "<enum \'qat.purr.compiler.config.ResultsFormatting\'>",
                        "$value": 3
                    }
                }
            },
            "metrics": {
                "$type": "<enum \'qat.purr.compiler.config.MetricsType\'>",
                "$value": 6
            },
            "active_calibrations": [],
            "optimizations": optimizations
        }
    }
    config_str = json.dumps(config)
    return config_str


class QPUBackend:
    def __init__(self):
        self.client = None

    def __enter__(self):
        self.client = ZMQClient()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if self.client:
            self.client.close()
            self.client = None

    def connect(self):
        self.client = ZMQClient()

    def disconnect(self):
        self.client.close()
        self.client = None

    # def run(self, program, config):
    #     if not self.client:
    #         raise RuntimeError("Not connected to the server")

    #     job = (program, config)
    #     self.client._send(job)
    #     result = self.client._await_results()
    #     return result

    def run(self, program, shots):
        if not self.client:
            raise RuntimeError("Not connected to the server")

        config = _config_build(shots)
        job = (program, config)
        self.client._send(job)
        result = self.client._await_results()
        return result

clients.py

from time import time
from typing import Union
from config.development import ZMQ_SERVER
import zmq

class ZMQBase:
    def __init__(self, socket_type):
        self._context = zmq.Context()
        self._socket = self._context.socket(socket_type)
        self._timeout = 30.0
        self._address = ZMQ_SERVER

    def _check_recieved(self):
        try:
            msg = self._socket.recv_pyobj()
            return msg
        except zmq.ZMQError:
            return None

    def _send(self, message) -> None:
        sent = False
        t0 = time()
        while not sent:
            try:
                self._socket.send_pyobj(message)
                sent = True
            except zmq.ZMQError as e:
                if time() > t0 + self._timeout:
                    raise TimeoutError(
                        "Sending %s on %s timedout" % (message, self._address)
                    )
        return

    def close(self):
        """Disconnect the link to the socket."""
        if self._socket.closed:
            return
        self._socket.close()
        self._context.destroy()

    def __del__(self):
        self.close()


class ZMQClient(ZMQBase):
    def __init__(self):
        super().__init__(zmq.REQ)
        self._socket.setsockopt(zmq.LINGER, 0)
        self._socket.connect(self._address)

    def _await_results(self):
        result = None
        while result is None:
            result = self._check_recieved()
        return result

tests

test_clients

test_backends

test_services

test_qmio