add support to execut qutip circuits on ionq

Author: splch

src/qutip_qip/ionq/__init__.py

@@ -0,0 +1,6 @@
+"""Simulation of IonQ circuits in ``qutip_qip``."""
+
+from .backend import IonQSimulator, IonQQPU
+from .converter import convert_qutip_circuit
+from .job import Job
+from .provider import Provider

src/qutip_qip/ionq/backend.py

@@ -0,0 +1,29 @@
+"""Backends for simulating circuits."""
+
+from .provider import Provider
+
+
+class IonQBackend:
+    def __init__(self, provider: Provider, backend: str, gateset: str):
+        self.provider = provider
+        self.provider.backend = backend
+        self.gateset = gateset
+
+    def run(self, circuit: dict, shots: int = 1024):
+        return self.provider.run(circuit, shots=shots)
+
+
+class IonQSimulator(IonQBackend):
+    def __init__(self, provider: Provider, gateset: str = "qis"):
+        super().__init__(provider, "simulator", gateset)
+
+
+class IonQQPU(IonQBackend):
+    def __init__(
+        self,
+        provider: Provider,
+        qpu: str = "harmony",
+        gateset: str = "qis",
+    ):
+        qpu_name = ".".join(("qpu", qpu)).lower()
+        super().__init__(provider, qpu_name, gateset)

src/qutip_qip/ionq/converter.py

@@ -0,0 +1,70 @@
+from qutip.qip.circuit import QubitCircuit, Gate
+
+
+def convert_qutip_circuit(qc: QubitCircuit) -> dict:
+    """
+    Convert a qutip_qip circuit to an IonQ circuit.
+
+    Parameters
+    ----------
+    qc: QubitCircuit
+        The qutip_qip circuit to be converted.
+
+    Returns
+    -------
+    dict
+        The IonQ circuit.
+    """
+    ionq_circuit = []
+    for gate in qc.gates:
+        if isinstance(gate, Gate):
+            ionq_circuit.append(
+                {
+                    "gate": gate.name,
+                    "targets": gate.targets,
+                    "controls": gate.controls or [],
+                }
+            )
+    return ionq_circuit
+
+
+def create_job_body(
+    circuit: dict,
+    shots: int,
+    backend: str,
+    format: str = "ionq.circuit.v0",
+) -> dict:
+    """
+    Create the body of a job request.
+
+    Parameters
+    ----------
+    circuit: dict
+        The IonQ circuit.
+    shots: int
+        The number of shots.
+    backend: str
+        The simulator or QPU backend.
+    format: str
+        The format of the circuit.
+
+    Returns
+    -------
+    dict
+        The body of the job request.
+    """
+    return {
+        "target": backend,
+        "input": {
+            "format": format,
+            "qubits": len(
+                {
+                    q
+                    for g in circuit
+                    for q in g.get("targets", []) + g.get("controls", [])
+                }
+            ),
+            "circuit": circuit,
+        },
+        "shots": shots,
+    }

src/qutip_qip/ionq/job.py

@@ -0,0 +1,91 @@
+"""Class for a running job."""
+
+from .converter import create_job_body
+import requests
+import time
+
+
+class Job:
+    """
+    Class for a running job.
+
+    Attributes
+    ----------
+    body: dict
+        The body of the job request.
+    """
+
+    def __init__(
+        self,
+        circuit: dict,
+        shots: int,
+        backend: str,
+        headers: dict,
+        url: str,
+    ) -> None:
+        self.circuit = circuit
+        self.shots = shots
+        self.backend = backend
+        self.headers = headers
+        self.url = url
+        self.id = None
+        self.results = None
+
+    def submit(self) -> None:
+        """
+        Submit the job.
+        """
+        json = create_job_body(
+            self.circuit,
+            self.shots,
+            self.backend,
+        )
+        response = requests.post(
+            f"{self.url}/jobs",
+            json=json,
+            headers=self.headers,
+        )
+        response.raise_for_status()
+        self.id = response.json()["id"]
+
+    def get_status(self) -> dict:
+        """
+        Get the status of the job.
+
+        Returns
+        -------
+        dict
+            The status of the job.
+        """
+        response = requests.get(
+            f"{self.url}/jobs/{self.id}",
+            headers=self.headers,
+        )
+        response.raise_for_status()
+        self.status = response.json()
+        return self.status
+
+    def get_results(self, polling_rate: int = 1) -> dict:
+        """
+        Get the results of the job.
+
+        Returns
+        -------
+        dict
+            The results of the job.
+        """
+        while self.get_status()["status"] not in (
+            "canceled",
+            "completed",
+            "failed",
+        ):
+            time.sleep(polling_rate)
+        response = requests.get(
+            f"{self.url}/jobs/{self.id}/results",
+            headers=self.headers,
+        )
+        response.raise_for_status()
+        self.results = {
+            k: int(round(v * self.shots)) for k, v in response.json().items()
+        }
+        return self.results

src/qutip_qip/ionq/provider.py

@@ -0,0 +1,63 @@
+"""Provider for the simulator backends."""
+
+from .converter import convert_qutip_circuit
+from .job import Job
+from ..version import version as __version__
+from os import getenv
+
+
+class Provider:
+    """
+    Provides access to qutip_qip based IonQ backends.
+
+    Attributes
+    ----------
+    name: str
+        Name of the provider
+    """
+
+    def __init__(
+        self,
+        token: str = None,
+        url: str = "https://api.ionq.co/v0.3",
+    ):
+        token = token or getenv("IONQ_API_KEY")
+        if not token:
+            raise ValueError("No token provided")
+        self.headers = self.create_headers(token)
+        self.url = url
+        self.backend = None
+
+    def run(self, circuit, shots: int = 1024) -> Job:
+        """
+        Run a circuit.
+
+        Parameters
+        ----------
+        circuit: QubitCircuit
+            The circuit to be run.
+        shots: int
+            The number of shots.
+
+        Returns
+        -------
+        Job
+            The running job.
+        """
+        ionq_circuit = convert_qutip_circuit(circuit)
+        job = Job(
+            ionq_circuit,
+            shots,
+            self.backend,
+            self.headers,
+            self.url,
+        )
+        job.submit()
+        return job
+
+    def create_headers(self, token: str):
+        return {
+            "Authorization": f"apiKey {token}",
+            "Content-Type": "application/json",
+            "User-Agent": f"qutip-qip/{__version__}",
+        }

tests/test_ionq.py

@@ -0,0 +1,75 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from qutip.qip.circuit import QubitCircuit, Gate
+from qutip.ionq import (
+    Provider,
+    IonQSimulator,
+    IonQQPU,
+    convert_qutip_circuit,
+    Job,
+)
+
+
+class TestConverter(unittest.TestCase):
+    def test_convert_qutip_circuit(self):
+        # Create a simple QubitCircuit with one gate for testing
+        qc = QubitCircuit(N=1)
+        qc.add_gate("X", targets=[0])
+        # Convert the qutip_qip circuit to IonQ format
+        ionq_circuit = convert_qutip_circuit(qc)
+        expected_output = [{"gate": "X", "targets": [0], "controls": []}]
+        self.assertEqual(ionq_circuit, expected_output)
+
+
+class TestIonQBackend(unittest.TestCase):
+    def setUp(self):
+        self.provider = Provider(token="dummy_token")
+
+    @patch("qutip.ionq.Provider")
+    def test_simulator_initialization(self, mock_provider):
+        simulator = IonQSimulator(provider=mock_provider)
+        self.assertEqual(simulator.provider, mock_provider)
+        self.assertEqual(simulator.gateset, "qis")
+
+    @patch("qutip.ionq.Provider")
+    def test_qpu_initialization(self, mock_provider):
+        qpu = IonQQPU(provider=mock_provider, qpu="harmony")
+        self.assertEqual(qpu.provider, mock_provider)
+        self.assertTrue("qpu.harmony" in qpu.provider.backend)
+
+
+class TestJob(unittest.TestCase):
+    @patch("requests.post")
+    def test_submit(self, mock_post):
+        mock_post.return_value.json.return_value = {"id": "test_job_id"}
+        mock_post.return_value.status_code = 200
+        job = Job(
+            circuit={},
+            shots=1024,
+            backend="simulator",
+            headers={},
+            url="http://dummy_url",
+        )
+        job.submit()
+        self.assertEqual(job.id, "test_job_id")
+
+    @patch("requests.get")
+    def test_get_results(self, mock_get):
+        mock_get.return_value.json.return_value = {"0": 0.5, "1": 0.5}
+        mock_get.return_value.status_code = 200
+        job = Job(
+            circuit={},
+            shots=1024,
+            backend="simulator",
+            headers={},
+            url="http://dummy_url",
+        )
+        job.id = (
+            "test_job_id"  # Simulate a job that has already been submitted
+        )
+        results = job.get_results(polling_rate=0)
+        self.assertEqual(results, {"0": 512, "1": 512})
+
+
+if __name__ == "__main__":
+    unittest.main()