Skip to content
8 changes: 5 additions & 3 deletions azure-quantum/azure/quantum/qiskit/backends/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,10 @@
QCIQPUBackend,
)

from azure.quantum.qiskit.backends.generic import (
AzureGenericQirBackend,
)

from .backend import AzureBackendBase

__all__ = [
"AzureBackendBase"
]
__all__ = ["AzureBackendBase"]
72 changes: 31 additions & 41 deletions azure-quantum/azure/quantum/qiskit/backends/backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@

try: # Qiskit 1.x legacy support
from qiskit.providers.models import BackendConfiguration # type: ignore

BackendConfigurationType = BackendConfiguration

from qiskit.qobj import QasmQobj, PulseQobj # type: ignore
Expand Down Expand Up @@ -263,15 +264,11 @@ def from_dict(cls, data: Mapping[str, Any]) -> "AzureBackendConfig":
)

@classmethod
def from_backend_configuration(
cls, configuration: Any
) -> "AzureBackendConfig":
def from_backend_configuration(cls, configuration: Any) -> "AzureBackendConfig":
return cls.from_dict(configuration.to_dict())


def _ensure_backend_config(
configuration: Any
) -> AzureBackendConfig:
def _ensure_backend_config(configuration: Any) -> AzureBackendConfig:
if isinstance(configuration, AzureBackendConfig):
return configuration

Expand All @@ -289,15 +286,12 @@ def _ensure_backend_config(
class AzureBackendBase(Backend, SessionHost):

# Name of the provider's input parameter which specifies number of shots for a submitted job.
# If None, backend will not pass this input parameter.
# If None, backend will not pass this input parameter.
_SHOTS_PARAM_NAME = "shots"

@abstractmethod
def __init__(
self,
configuration: Any,
provider: "AzureQuantumProvider" = None,
**fields
self, configuration: Any, provider: "AzureQuantumProvider" = None, **fields
):
if configuration is None:
raise ValueError("Backend configuration is required for Azure backends")
Expand Down Expand Up @@ -339,19 +333,19 @@ def _build_target(self, configuration: AzureBackendConfig) -> Target:
target.add_instruction(instruction)

return target

@abstractmethod
def run(
self,
run_input: Union[QuantumCircuit, List[QuantumCircuit]] = [],
shots: int = None,
shots: int = None,
**options,
) -> AzureQuantumJob:
"""Run on the backend.

This method returns a
:class:`~azure.quantum.qiskit.job.AzureQuantumJob` object
that runs circuits.
that runs circuits.

Args:
run_input (QuantumCircuit or List[QuantumCircuit]): An individual or a
Expand Down Expand Up @@ -399,6 +393,7 @@ def target(self) -> Target:
@property
def max_circuits(self) -> Optional[int]:
return 1

def retrieve_job(self, job_id) -> AzureQuantumJob:
"""Returns the Job instance associated with the given id."""
return self.provider.get_job(job_id)
Expand All @@ -415,8 +410,10 @@ def _get_output_data_format(self, options: Dict[str, Any] = {}) -> str:
output_data_format = options.pop("output_data_format", azure_defined_override)

return output_data_format

def _get_input_params(self, options: Dict[str, Any], shots: int = None) -> Dict[str, Any]:

def _get_input_params(
self, options: Dict[str, Any], shots: int = None
) -> Dict[str, Any]:
# Backend options are mapped to input_params.
input_params: Dict[str, Any] = vars(self.options).copy()

Expand All @@ -426,7 +423,7 @@ def _get_input_params(self, options: Dict[str, Any], shots: int = None) -> Dict[

final_shots = None
# First we check for the explicitly specified 'shots' parameter, then for a provider-specific
# field in options, then for a backend's default value.
# field in options, then for a backend's default value.

# Warn about options conflict, default to 'shots'.
if shots is not None and options_shots is not None:
Expand All @@ -436,7 +433,7 @@ def _get_input_params(self, options: Dict[str, Any], shots: int = None) -> Dict[
stacklevel=3,
)
final_shots = shots

elif shots is not None:
final_shots = shots
elif options_shots is not None:
Expand All @@ -446,7 +443,7 @@ def _get_input_params(self, options: Dict[str, Any], shots: int = None) -> Dict[
stacklevel=3,
)
final_shots = options_shots

# If nothing is found, try to get from default values.
if final_shots is None:
final_shots = input_params.get(self.__class__._SHOTS_PARAM_NAME)
Expand All @@ -456,7 +453,6 @@ def _get_input_params(self, options: Dict[str, Any], shots: int = None) -> Dict[
_ = options.pop("count", None)

input_params[self.__class__._SHOTS_PARAM_NAME] = final_shots


if "items" in options:
input_params["items"] = options.pop("items")
Expand Down Expand Up @@ -487,10 +483,7 @@ def _run(self, job_name, input_data, input_params, metadata, **options):
# Anything left here is an invalid parameter with the user attempting to use
# deprecated parameters.
targetCapability = input_params.get("targetCapability", None)
if (
targetCapability not in [None, "qasm"]
and input_data_format != "qir.v1"
):
if targetCapability not in [None, "qasm"] and input_data_format != "qir.v1":
message = "The targetCapability parameter has been deprecated and is only supported for QIR backends."
message += os.linesep
message += "To find a QIR capable backend, use the following code:"
Expand All @@ -500,7 +493,6 @@ def _run(self, job_name, input_data, input_params, metadata, **options):
)
raise ValueError(message)


# Update metadata with all remaining options values, then clear options
# JobDetails model will error if unknown keys are passed down which
# can happen with estiamtor and backend wrappers
Expand Down Expand Up @@ -586,14 +578,14 @@ def _azure_config(self) -> Dict[str, str]:
"output_data_format": "microsoft.quantum-results.v2",
"is_default": True,
}

def _basis_gates(self) -> List[str]:
return QIR_BASIS_GATES

def run(
self,
run_input: Union[QuantumCircuit, List[QuantumCircuit]] = [],
shots: int = None,
shots: int = None,
**options,
) -> AzureQuantumJob:
"""Run on the backend.
Expand Down Expand Up @@ -632,7 +624,7 @@ def run(

# config normalization
input_params = self._get_input_params(options, shots=shots)

shots_count = None

if self._can_send_shots_input_param():
Expand All @@ -658,11 +650,10 @@ def _prepare_job_metadata(self, circuit: QuantumCircuit) -> Dict[str, str]:
return {
"qiskit": str(True),
"name": circuit.name,
"num_qubits": circuit.num_qubits,
"num_qubits": str(circuit.num_qubits),
"metadata": json.dumps(circuit.metadata),
}


def _get_qir_str(
self, circuit: QuantumCircuit, target_profile: TargetProfile, **kwargs
) -> str:
Expand All @@ -679,9 +670,8 @@ def _get_qir_str(
)

qir_str = backend.qir(circuit)

return qir_str

return qir_str

def _translate_input(
self, circuit: QuantumCircuit, input_params: Dict[str, Any]
Expand All @@ -703,7 +693,7 @@ def _translate_input(
category=DeprecationWarning,
stacklevel=3,
)

qir_str = self._get_qir_str(
circuit, target_profile, skip_transpilation=skip_transpilation
)
Expand Down Expand Up @@ -763,9 +753,9 @@ def __init__(
def _prepare_job_metadata(self, circuit):
"""Returns the metadata relative to the given circuit that will be attached to the Job"""
return {
"qiskit": True,
"qiskit": str(True),
"name": circuit.name,
"num_qubits": circuit.num_qubits,
"num_qubits": str(circuit.num_qubits),
"metadata": json.dumps(circuit.metadata),
}

Expand All @@ -774,12 +764,12 @@ def _translate_input(self, circuit):
pass

def run(
self,
run_input: Union[QuantumCircuit, List[QuantumCircuit]] = [],
shots: int = None,
**options,
):
"""Submits the given circuit to run on an Azure Quantum backend."""
self,
run_input: Union[QuantumCircuit, List[QuantumCircuit]] = [],
shots: int = None,
**options,
):
"""Submits the given circuit to run on an Azure Quantum backend."""
circuit = self._normalize_run_input_params(run_input, **options)
options.pop("run_input", None)
options.pop("circuit", None)
Expand Down
122 changes: 122 additions & 0 deletions azure-quantum/azure/quantum/qiskit/backends/generic.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
##
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
##

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Dict, Optional

from azure.quantum.version import __version__

try:
from qiskit.providers import Options
from qsharp import TargetProfile
except ImportError as exc:
raise ImportError(
"Missing optional 'qiskit' dependencies. \
To install run: pip install azure-quantum[qiskit]"
) from exc

from .backend import AzureBackendConfig, AzureQirBackend

if TYPE_CHECKING:
from azure.quantum.qiskit import AzureQuantumProvider


_DEFAULT_SHOTS_COUNT = 500


class AzureGenericQirBackend(AzureQirBackend):
"""Fallback QIR backend for arbitrary Azure Quantum workspace targets.

This backend is created dynamically by :class:`~azure.quantum.qiskit.provider.AzureQuantumProvider`
for targets present in the workspace that do not have a dedicated Qiskit backend class.

It submits Qiskit circuits using QIR (`qir.v1`) payloads.
"""

_SHOTS_PARAM_NAME = "shots"

def __init__(
self,
name: str,
provider: "AzureQuantumProvider",
*,
provider_id: str,
target_profile: Optional[TargetProfile | str] = None,
num_qubits: Optional[int] = None,
description: Optional[str] = None,
**kwargs: Any,
):
self._provider_id = provider_id

config = AzureBackendConfig.from_dict(
{
"backend_name": name,
"backend_version": __version__,
"simulator": False,
"local": False,
"coupling_map": None,
"description": description
or f"Azure Quantum target '{name}' (generic QIR backend)",
"basis_gates": self._basis_gates(),
"memory": False,
"n_qubits": num_qubits,
"conditional": False,
"max_shots": None,
"open_pulse": False,
"gates": [{"name": "TODO", "parameters": [], "qasm_def": "TODO"}],
"azure": self._azure_config(),
}
)

super().__init__(config, provider, **kwargs)

# Prefer an instance-specific target profile discovered from the workspace target metadata.
default_target_profile = self._coerce_target_profile(target_profile)
if default_target_profile is not None:
self.set_options(target_profile=default_target_profile)

@staticmethod
def _coerce_target_profile(
value: Optional[TargetProfile | str],
) -> Optional[TargetProfile]:
if value is None:
return None
if isinstance(value, TargetProfile):
return value
if not isinstance(value, str):
return None

raw = value.strip()
if not raw:
return None

# Prefer the qsharp helper when available.
from_str = getattr(TargetProfile, "from_str", None)
if callable(from_str):
try:
parsed = from_str(raw)
if isinstance(parsed, TargetProfile):
return parsed
except Exception:
pass

# Best-effort: try enum attribute lookup.
normalized = raw.replace("-", "_")
return getattr(TargetProfile, normalized, None)

@classmethod
def _default_options(cls) -> Options:
# Default to the most conservative QIR profile; users can override per-run via
# `target_profile=` in backend.run(...).
return Options(
**{cls._SHOTS_PARAM_NAME: _DEFAULT_SHOTS_COUNT},
target_profile=TargetProfile.Base,
)

def _azure_config(self) -> Dict[str, str]:
config = super()._azure_config()
config.update({"provider_id": self._provider_id, "is_default": False})
return config
Loading
Loading