diff --git a/src/qutip_qip/ionq/__init__.py b/src/qutip_qip/ionq/__init__.py new file mode 100644 index 00000000..6f954ef0 --- /dev/null +++ b/src/qutip_qip/ionq/__init__.py @@ -0,0 +1,9 @@ +"""Simulation of IonQ circuits in ``qutip_qip``.""" + +from .backend import IonQSimulator, IonQQPU +from .converter import ( + convert_qutip_circuit, + convert_ionq_response_to_circuitresult, +) +from .job import Job +from .provider import IonQProvider diff --git a/src/qutip_qip/ionq/backend.py b/src/qutip_qip/ionq/backend.py new file mode 100644 index 00000000..f898878f --- /dev/null +++ b/src/qutip_qip/ionq/backend.py @@ -0,0 +1,33 @@ +"""Backends for simulating circuits.""" + +from .provider import IonQProvider + + +class IonQBackend: + def __init__(self, provider: IonQProvider, backend: str, gateset: str): + self.provider = provider + self.provider.backend = backend + self.provider.gateset = gateset + + def run(self, circuit: dict, shots: int = 1024): + return self.provider.run(circuit, shots=shots) + + +class IonQSimulator(IonQBackend): + def __init__( + self, + provider: IonQProvider, + gateset: str = "qis", + ): + super().__init__(provider, "simulator", gateset) + + +class IonQQPU(IonQBackend): + def __init__( + self, + provider: IonQProvider, + qpu: str = "harmony", + gateset: str = "qis", + ): + qpu_name = ".".join(("qpu", qpu)).lower() + super().__init__(provider, qpu_name, gateset) diff --git a/src/qutip_qip/ionq/converter.py b/src/qutip_qip/ionq/converter.py new file mode 100644 index 00000000..3549ad1d --- /dev/null +++ b/src/qutip_qip/ionq/converter.py @@ -0,0 +1,135 @@ +from qutip import Qobj +from qutip_qip.circuit import QubitCircuit, CircuitResult +import numpy as np + + +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: + g = {"gate": gate.name} + # Map target(s) and control(s) depending on the number of qubits + for attr, key in (("targets", "target"), ("controls", "control")): + items = getattr(gate, attr, None) + if items: + g[key if len(items) == 1 else key + "s"] = ( + items[0] if len(items) == 1 else items + ) + # Include arg_value as angle, phase, and rotation if it exists + if getattr(gate, "arg_value", None) is not None: + g.update( + { + "angle": gate.arg_value, + "phase": gate.arg_value, + "rotation": gate.arg_value, + } + ) + ionq_circuit.append(g) + return ionq_circuit + + +def convert_ionq_response_to_circuitresult(ionq_response: dict): + """ + Convert an IonQ response to a CircuitResult. + + Parameters + ---------- + ionq_response: dict + The IonQ response {state: probability, ...}. + + Returns + ------- + CircuitResult + The CircuitResult. + """ + # Calculate the number of qubits based on the binary representation of the highest state + num_qubits = max(len(state) for state in ionq_response.keys()) + + # Initialize an empty density matrix for the mixed state + density_matrix = np.zeros((2**num_qubits, 2**num_qubits), dtype=complex) + + # Iterate over the measurement outcomes and their probabilities + for state, probability in ionq_response.items(): + # Ensure state string is correctly padded for single-qubit cases + binary_state = format(int(state, base=10), f"0{num_qubits}b") + index = int(binary_state, 2) # Convert binary string back to integer + + # Update the density matrix to include this measurement outcome + state_vector = np.zeros((2**num_qubits,), dtype=complex) + state_vector[index] = ( + 1.0 # Pure state corresponding to the measurement outcome + ) + density_matrix += probability * np.outer( + state_vector, state_vector.conj() + ) # Add weighted outer product + + # Convert the numpy array to a Qobj density matrix + qobj_density_matrix = Qobj( + density_matrix, dims=[[2] * num_qubits, [2] * num_qubits] + ) + + return CircuitResult( + [qobj_density_matrix], [1.0] + ) # Return the density matrix wrapped in CircuitResult + + +def create_job_body( + circuit: dict, + shots: int, + backend: str, + gateset: 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. + gateset: str + Either native or compiled gates. + format: str + The format of the circuit. + + Returns + ------- + dict + The body of the job request. + """ + return { + "target": backend, + "shots": shots, + "input": { + "format": format, + "gateset": gateset, + "circuit": circuit, + "qubits": len( + { + q + for g in circuit + for q in g.get("targets", []) + + g.get("controls", []) + + [g.get("target")] + + [g.get("control")] + if q is not None + } + ), + }, + } diff --git a/src/qutip_qip/ionq/job.py b/src/qutip_qip/ionq/job.py new file mode 100644 index 00000000..c542d3cb --- /dev/null +++ b/src/qutip_qip/ionq/job.py @@ -0,0 +1,92 @@ +"""Class for a running job.""" + +from .converter import create_job_body, convert_ionq_response_to_circuitresult +from qutip_qip.circuit import CircuitResult +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, + gateset: str, + headers: dict, + url: str, + ) -> None: + self.circuit = circuit + self.shots = shots + self.backend = backend + self.gateset = gateset + 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, + self.gateset, + ) + 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) -> CircuitResult: + """ + 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() + return convert_ionq_response_to_circuitresult(response.json()) diff --git a/src/qutip_qip/ionq/provider.py b/src/qutip_qip/ionq/provider.py new file mode 100644 index 00000000..b8734ee6 --- /dev/null +++ b/src/qutip_qip/ionq/provider.py @@ -0,0 +1,64 @@ +"""Provider for the IonQ backends.""" + +from .converter import convert_qutip_circuit +from .job import Job +from ..version import version as __version__ +from os import getenv + + +class IonQProvider: + """ + 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.gateset, + 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__}", + } diff --git a/tests/test_ionq.py b/tests/test_ionq.py new file mode 100644 index 00000000..5fac17c9 --- /dev/null +++ b/tests/test_ionq.py @@ -0,0 +1,90 @@ +import unittest +from unittest.mock import patch, MagicMock +from qutip_qip.circuit import QubitCircuit, Gate +from qutip_qip.ionq import ( + IonQProvider, + IonQSimulator, + IonQQPU, + convert_qutip_circuit, + convert_ionq_response_to_circuitresult, + 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("H", targets=[0]) + qc.add_gate("CNOT", targets=[0], controls=[1]) + # Convert the qutip_qip circuit to IonQ format + ionq_circuit = convert_qutip_circuit(qc) + expected_output = [ + {"gate": "H", "target": 0}, + {"gate": "CNOT", "target": 0, "control": 1}, + ] + self.assertEqual(ionq_circuit, expected_output) + + +class TestIonQBackend(unittest.TestCase): + def setUp(self): + self.provider = IonQProvider(token="dummy_token") + + @patch("qutip_qip.ionq.IonQProvider") + def test_simulator_initialization(self, mock_provider): + simulator = IonQSimulator(provider=mock_provider) + self.assertEqual(simulator.provider, mock_provider) + self.assertEqual(mock_provider.gateset, "qis") + + @patch("qutip_qip.ionq.IonQProvider") + 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", + gateset="qis", + headers={}, + url="http://dummy_url", + ) + job.submit() + self.assertEqual(job.id, "test_job_id") + + @patch("requests.get") + def test_get_results(self, mock_get): + # Simulate the status check response and the final result response + mock_get.side_effect = [ + MagicMock( + json=lambda: {"status": "completed"} + ), # Simulated status check response + MagicMock( + json=lambda: {"0": 0.5, "1": 0.5} + ), # Simulated final results response + ] + job = Job( + circuit={}, + shots=1024, + backend="simulator", + gateset="qis", + 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.get_final_states(), + convert_ionq_response_to_circuitresult( + {"0": 0.5, "1": 0.5} + ).get_final_states(), + )