qpandalite.task.adapters package

Submodules

qpandalite.task.adapters.base module

Base adapter interface for quantum cloud backends.

Every backend adapter must implement this interface, providing: 1. Translation from OriginIR string to the provider's native circuit type. 2. Task submission via the provider's Python SDK (not raw REST). 3. Task status query and result retrieval.

The adapter layer replaces all direct requests REST calls within the task modules. Each adapter is a stateful object that holds the provider session / client and configuration.

class qpandalite.task.adapters.base.QuantumAdapter

基类：ABC

Abstract base class for quantum cloud backend adapters.

Subclass this for each backend (originq_cloud, quafu, ibm, ...). Each adapter is instantiated once per task module and reused.

is_available()

Return True if the required packages / credentials are configured.

Defaults to False so that subclasses must explicitly opt-in, avoiding the risk of an unconfigured adapter incorrectly reporting availability.

返回类型:

bool

name: str = 'base'

abstractmethod query(taskid)

Query a single task's status and result.

参数:

taskid (str) -- Task identifier.

返回:

• status: 'success' | 'failed' | 'running'

• result: execution result (present when status is 'success' or 'failed')

返回类型:

dict with keys

abstractmethod query_batch(taskids)

Query multiple tasks' status and merge results.

Overall status is the worst case: 'failed' > 'running' > 'success'.

参数:

taskids (list[str]) -- List of task identifiers.

返回:

status, result (list of results).

返回类型:

dict with keys

abstractmethod submit(circuit, *, shots=1000, **kwargs)

Submit a circuit to the backend and return a task ID.

参数:

• circuit (Any) -- Provider-native circuit object (result of translate_circuit).

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional provider-specific parameters (e.g. chip_id, auto_mapping, circuit_optimize).

返回:

Task ID assigned by the backend.

返回类型:

str

abstractmethod submit_batch(circuits, *, shots=1000, **kwargs)

Submit multiple circuits as a single batch.

参数:

• circuits (list[Any]) -- List of provider-native circuit objects.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional provider-specific parameters.

返回:

Task IDs (one per circuit), or a single task ID if the backend returns a group ID.

返回类型:

list[str]

abstractmethod translate_circuit(originir)

Translate an OriginIR circuit string to the provider's native circuit type.

参数:

originir (str) -- Circuit in OriginIR format.

返回:

Provider-specific circuit object.

返回类型:

Any

qpandalite.task.adapters.originq_adapter module

OriginQ Cloud backend adapter.

Submits OriginIR circuits to the OriginQ Cloud service using pyqpanda3.

Installation:

pip install qpandalite[originq]

class qpandalite.task.adapters.originq_adapter.OriginQAdapter

基类：QuantumAdapter

Adapter for OriginQ Cloud (本源量子云) using pyqpanda3.

This adapter uses pyqpanda3's QCloudService API for cloud task submission, which simplifies configuration by only requiring an API key.

备注

The pyqpanda3 package is required for this adapter. Install with: pip install qpandalite[originq]

is_available()

Check if the OriginQ adapter is available (credentials configured).

返回:

True if api_key is configured.

返回类型:

bool

name: str = 'originq'

query(taskid)

Query a single task's status.

参数:

taskid (str) -- Task ID to query.

返回:

taskid, status, result (if completed)

返回类型:

dict with keys

query_batch(taskids)

Query multiple tasks and merge results.

参数:

taskids (list[str]) -- List of task IDs to query.

返回:

Combined result dict with status and merged results.

返回类型:

dict[str, Any]

query_sync(taskid, interval=2.0, timeout=60.0, retry=5)

Poll task status until completion or timeout.

参数:

• taskid (str | list[str]) -- Task ID or list of task IDs.

• interval (float) -- Polling interval in seconds.

• timeout (float) -- Maximum wait time in seconds.

• retry (int) -- Number of retries on query failure.

返回:

List of result dicts.

抛出:

• TimeoutError -- If timeout is reached.

• RuntimeError -- If task fails or retry exhausted.

返回类型:

list[dict[str, Any]]

submit(circuit, *, shots=1000, **kwargs)

Submit a single circuit to OriginQ Cloud.

参数:

• circuit (str) -- OriginIR format circuit string.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional options: - backend_name: Backend name (e.g., 'origin:wuyuan:d5') - circuit_optimize: Enable circuit optimization (default: True) - measurement_amend: Enable measurement amendment (default: False) - auto_mapping: Enable automatic qubit mapping (default: False)

返回:

Task ID string.

返回类型:

str

submit_batch(circuits, *, shots=1000, **kwargs)

Submit circuits as a group.

Note: pyqpanda3 handles batch submission internally. This method submits circuits sequentially if needed for grouping.

参数:

• circuits (list[str]) -- List of OriginIR format circuit strings.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional options (see submit()).

返回:

Single task ID or list of task IDs if split into groups.

返回类型:

str | list[str]

translate_circuit(originir)

Convert OriginIR string to QProg using pyqpanda3.

参数:

originir (str) -- OriginIR format circuit string.

返回:

QProg object for pyqpanda3.

返回类型:

Any

qpandalite.task.adapters.qiskit_adapter module

Qiskit backend adapter.

Translates OriginIR circuits to Qiskit QuantumCircuit objects and submits via the qiskit / qiskit_ibm_provider packages. No raw REST calls.

Installation:

pip install qpandalite[qiskit]

class qpandalite.task.adapters.qiskit_adapter.QiskitAdapter(proxy=None)

基类：QuantumAdapter

Adapter for IBM Quantum backends via Qiskit.

Proxy Configuration:

Proxies can be passed via the proxy parameter: - Dict with 'http' and/or 'https' keys - Or a single proxy URL string for both protocols

抛出:

MissingDependencyError -- If qiskit or qiskit_ibm_provider is not installed.

参数:

proxy (dict[str, str] | str | None)

示例

>>> adapter = QiskitAdapter(proxy={
...     "http": "http://proxy.example.com:8080",
...     "https": "https://proxy.example.com:8080"
... })

is_available()

Check if the Qiskit adapter is available (IBM provider initialized).

返回:

True if the IBM provider was successfully initialized.

返回类型:

bool

name: str = 'ibm'

query(taskid)

Query a single IBM Quantum job's status.

参数:

taskid (str)

返回类型:

dict[str, Any]

query_batch(taskids)

Query multiple IBM Quantum jobs and merge results.

参数:

taskids (list[str])

返回类型:

dict[str, Any]

query_sync(taskid, interval=2.0, timeout=60.0, retry=5)

Poll task status until completion or timeout.

参数:

• taskid (str | list[str])

• interval (float)

• timeout (float)

• retry (int)

返回类型:

list[dict[str, Any]]

submit(circuit, *, shots=1000, **kwargs)

Submit a single circuit to IBM Quantum.

参数:

• circuit (qiskit.QuantumCircuit)

• shots (int)

• kwargs (Any)

返回类型:

str

submit_batch(circuits, *, shots=1000, **kwargs)

Submit multiple circuits as a batch. Returns a single job ID.

参数:

• circuits (list['qiskit.QuantumCircuit'])

• shots (int)

• kwargs (Any)

返回类型:

str

translate_circuit(originir)

Translate an OriginIR string to a Qiskit QuantumCircuit.

The conversion path is OriginIR → QASM string → Qiskit QuantumCircuit. This is the most compatible route given the current API surface of qpandalite.circuit_builder.qcircuit (which exposes QASM export but not a direct Qiskit-native constructor). An optimised direct path can be evaluated in a future iteration if needed.

参数:

originir (str)

返回类型:

qiskit.QuantumCircuit

qpandalite.task.adapters.quafu_adapter module

Quafu backend adapter.

Translates OriginIR circuits to Quafu QuantumCircuit objects and submits via the quafu package (User / Task API). No raw REST calls.

Installation:

pip install qpandalite[quafu]

class qpandalite.task.adapters.quafu_adapter.QuafuAdapter

基类：QuantumAdapter

Adapter for the BAQIS Quafu (ScQ) quantum cloud platform.

抛出:

MissingDependencyError -- If quafu package is not installed.

VALID_CHIP_IDS = frozenset({'Dongling', 'ScQ-P10', 'ScQ-P10C', 'ScQ-P136', 'ScQ-P18'})

property api_token: str

Return the API token used for Quafu authentication.

返回:

The Quafu API token.

返回类型:

str

is_available()

Check if the Quafu adapter is available (quafu package installed).

返回:

True if the quafu package was successfully imported.

返回类型:

bool

name: str = 'quafu'

query(taskid)

Query a single Quafu task's status via SDK Task.retrieve().

Uses the internally maintained history dict so the caller does not need to pass any additional context.

参数:

taskid (str)

返回类型:

dict[str, Any]

query_batch(taskids)

Query multiple Quafu tasks and merge results.

参数:

taskids (list[str])

返回类型:

dict[str, Any]

submit(circuit, *, shots=10000, **kwargs)

Submit a single circuit to Quafu.

参数:

• circuit (QuantumCircuit)

• shots (int)

• kwargs (Any)

返回类型:

str

submit_batch(circuits, *, shots=10000, **kwargs)

Submit multiple circuits as a group to Quafu.

参数:

• circuits (list['QuantumCircuit'])

• shots (int)

• kwargs (Any)

返回类型:

list[str]

translate_circuit(originir)

Translate an OriginIR string to a Quafu QuantumCircuit.

参数:

originir (str)

返回类型:

QuantumCircuit

Module contents

Quantum cloud backend adapters.

Each adapter provides a consistent interface (submit / query / translate) for a specific quantum computing provider, encapsulating all network communication within the adapter layer.

class qpandalite.task.adapters.DummyAdapter(noise_model=None, available_qubits=None, available_topology=None)

基类：QuantumAdapter

Local simulator adapter that mimics cloud backends.

This adapter executes circuits locally using the built-in OriginIR simulator instead of submitting to real quantum hardware. It provides the same interface as cloud adapters, making it a drop-in replacement.

Features: - Immediate result availability (no waiting for queue) - Optional noise simulation - Deterministic task IDs (based on circuit hash) - Same result format as cloud backends

参数:

• noise_model (Optional[Dict[str, Any]])

• available_qubits (Optional[List[int]])

• available_topology (Optional[List[List[int]]])

name

Adapter identifier ('dummy').

Type:

str

noise_model

Optional noise configuration for simulation.

available_qubits

List of qubit indices available for simulation.

available_topology

List of [u, v] edges defining qubit connectivity.

示例

>>> adapter = DummyAdapter()
>>> task_id = adapter.submit("QINIT 2\nH q[0]\nCNOT q[0] q[1]\nMEASURE")
>>> result = adapter.query(task_id)
>>> print(result['status'])
'success'

clear_cache()

Clear the internal result cache.

返回类型:

None

is_available()

Check if the dummy adapter is available.

返回:

True if simulation dependencies are installed.

返回类型:

bool

name: str = 'dummy'

query(taskid)

Retrieve the cached result for a task.

Since dummy tasks are executed immediately on submission, results are always available instantly.

参数:

taskid (str) -- Task identifier.

返回:

Result dict with 'status' and 'result' (or 'error') keys.

返回类型:

Dict[str, Any]

query_batch(taskids)

Query multiple tasks and merge results.

参数:

taskids (List[str]) -- List of task identifiers.

返回:

Combined result dict with overall status and merged results.

返回类型:

Dict[str, Any]

submit(circuit, *, shots=1000, **kwargs)

Simulate a circuit locally and cache the result.

The circuit is executed immediately using the local simulator, and results are cached for later retrieval via query().

参数:

• circuit (str) -- Circuit in OriginIR format (or pre-translated).

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional parameters (ignored for dummy adapter).

返回:

Task ID for result retrieval.

返回类型:

str

submit_batch(circuits, *, shots=1000, **kwargs)

Simulate multiple circuits locally.

参数:

• circuits (List[str]) -- List of circuits in OriginIR format.

• shots (int) -- Number of measurement shots per circuit.

• **kwargs (Any) -- Additional parameters (ignored).

返回:

List of task IDs, one per circuit.

返回类型:

List[str]

translate_circuit(originir)

Return the OriginIR string unchanged.

The dummy adapter accepts OriginIR directly, so no translation needed.

参数:

originir (str) -- Circuit in OriginIR format.

返回:

The same OriginIR string.

返回类型:

str

class qpandalite.task.adapters.OriginQAdapter

基类：QuantumAdapter

Adapter for OriginQ Cloud (本源量子云) using pyqpanda3.

This adapter uses pyqpanda3's QCloudService API for cloud task submission, which simplifies configuration by only requiring an API key.

备注

The pyqpanda3 package is required for this adapter. Install with: pip install qpandalite[originq]

is_available()

Check if the OriginQ adapter is available (credentials configured).

返回:

True if api_key is configured.

返回类型:

bool

name: str = 'originq'

query(taskid)

Query a single task's status.

参数:

taskid (str) -- Task ID to query.

返回:

taskid, status, result (if completed)

返回类型:

dict with keys

query_batch(taskids)

Query multiple tasks and merge results.

参数:

taskids (list[str]) -- List of task IDs to query.

返回:

Combined result dict with status and merged results.

返回类型:

dict[str, Any]

query_sync(taskid, interval=2.0, timeout=60.0, retry=5)

Poll task status until completion or timeout.

参数:

• taskid (str | list[str]) -- Task ID or list of task IDs.

• interval (float) -- Polling interval in seconds.

• timeout (float) -- Maximum wait time in seconds.

• retry (int) -- Number of retries on query failure.

返回:

List of result dicts.

抛出:

• TimeoutError -- If timeout is reached.

• RuntimeError -- If task fails or retry exhausted.

返回类型:

list[dict[str, Any]]

submit(circuit, *, shots=1000, **kwargs)

Submit a single circuit to OriginQ Cloud.

参数:

• circuit (str) -- OriginIR format circuit string.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional options: - backend_name: Backend name (e.g., 'origin:wuyuan:d5') - circuit_optimize: Enable circuit optimization (default: True) - measurement_amend: Enable measurement amendment (default: False) - auto_mapping: Enable automatic qubit mapping (default: False)

返回:

Task ID string.

返回类型:

str

submit_batch(circuits, *, shots=1000, **kwargs)

Submit circuits as a group.

Note: pyqpanda3 handles batch submission internally. This method submits circuits sequentially if needed for grouping.

参数:

• circuits (list[str]) -- List of OriginIR format circuit strings.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional options (see submit()).

返回:

Single task ID or list of task IDs if split into groups.

返回类型:

str | list[str]

translate_circuit(originir)

Convert OriginIR string to QProg using pyqpanda3.

参数:

originir (str) -- OriginIR format circuit string.

返回:

QProg object for pyqpanda3.

返回类型:

Any

class qpandalite.task.adapters.QiskitAdapter(proxy=None)

基类：QuantumAdapter

Adapter for IBM Quantum backends via Qiskit.

Proxy Configuration:

Proxies can be passed via the proxy parameter: - Dict with 'http' and/or 'https' keys - Or a single proxy URL string for both protocols

抛出:

MissingDependencyError -- If qiskit or qiskit_ibm_provider is not installed.

参数:

proxy (dict[str, str] | str | None)

示例

>>> adapter = QiskitAdapter(proxy={
...     "http": "http://proxy.example.com:8080",
...     "https": "https://proxy.example.com:8080"
... })

is_available()

Check if the Qiskit adapter is available (IBM provider initialized).

返回:

True if the IBM provider was successfully initialized.

返回类型:

bool

name: str = 'ibm'

query(taskid)

Query a single IBM Quantum job's status.

参数:

taskid (str)

返回类型:

dict[str, Any]

query_batch(taskids)

Query multiple IBM Quantum jobs and merge results.

参数:

taskids (list[str])

返回类型:

dict[str, Any]

query_sync(taskid, interval=2.0, timeout=60.0, retry=5)

Poll task status until completion or timeout.

参数:

• taskid (str | list[str])

• interval (float)

• timeout (float)

• retry (int)

返回类型:

list[dict[str, Any]]

submit(circuit, *, shots=1000, **kwargs)

Submit a single circuit to IBM Quantum.

参数:

• circuit (qiskit.QuantumCircuit)

• shots (int)

• kwargs (Any)

返回类型:

str

submit_batch(circuits, *, shots=1000, **kwargs)

Submit multiple circuits as a batch. Returns a single job ID.

参数:

• circuits (list['qiskit.QuantumCircuit'])

• shots (int)

• kwargs (Any)

返回类型:

str

translate_circuit(originir)

Translate an OriginIR string to a Qiskit QuantumCircuit.

The conversion path is OriginIR → QASM string → Qiskit QuantumCircuit. This is the most compatible route given the current API surface of qpandalite.circuit_builder.qcircuit (which exposes QASM export but not a direct Qiskit-native constructor). An optimised direct path can be evaluated in a future iteration if needed.

参数:

originir (str)

返回类型:

qiskit.QuantumCircuit

class qpandalite.task.adapters.QuafuAdapter

基类：QuantumAdapter

Adapter for the BAQIS Quafu (ScQ) quantum cloud platform.

抛出:

MissingDependencyError -- If quafu package is not installed.

VALID_CHIP_IDS = frozenset({'Dongling', 'ScQ-P10', 'ScQ-P10C', 'ScQ-P136', 'ScQ-P18'})

property api_token: str

Return the API token used for Quafu authentication.

返回:

The Quafu API token.

返回类型:

str

is_available()

Check if the Quafu adapter is available (quafu package installed).

返回:

True if the quafu package was successfully imported.

返回类型:

bool

name: str = 'quafu'

query(taskid)

Query a single Quafu task's status via SDK Task.retrieve().

Uses the internally maintained history dict so the caller does not need to pass any additional context.

参数:

taskid (str)

返回类型:

dict[str, Any]

query_batch(taskids)

Query multiple Quafu tasks and merge results.

参数:

taskids (list[str])

返回类型:

dict[str, Any]

submit(circuit, *, shots=10000, **kwargs)

Submit a single circuit to Quafu.

参数:

• circuit (QuantumCircuit)

• shots (int)

• kwargs (Any)

返回类型:

str

submit_batch(circuits, *, shots=10000, **kwargs)

Submit multiple circuits as a group to Quafu.

参数:

• circuits (list['QuantumCircuit'])

• shots (int)

• kwargs (Any)

返回类型:

list[str]

translate_circuit(originir)

Translate an OriginIR string to a Quafu QuantumCircuit.

参数:

originir (str)

返回类型:

QuantumCircuit

class qpandalite.task.adapters.QuantumAdapter

基类：ABC

Abstract base class for quantum cloud backend adapters.

Subclass this for each backend (originq_cloud, quafu, ibm, ...). Each adapter is instantiated once per task module and reused.

is_available()

Return True if the required packages / credentials are configured.

Defaults to False so that subclasses must explicitly opt-in, avoiding the risk of an unconfigured adapter incorrectly reporting availability.

返回类型:

bool

name: str = 'base'

abstractmethod query(taskid)

Query a single task's status and result.

参数:

taskid (str) -- Task identifier.

返回:

• status: 'success' | 'failed' | 'running'

• result: execution result (present when status is 'success' or 'failed')

返回类型:

dict with keys

abstractmethod query_batch(taskids)

Query multiple tasks' status and merge results.

Overall status is the worst case: 'failed' > 'running' > 'success'.

参数:

taskids (list[str]) -- List of task identifiers.

返回:

status, result (list of results).

返回类型:

dict with keys

abstractmethod submit(circuit, *, shots=1000, **kwargs)

Submit a circuit to the backend and return a task ID.

参数:

• circuit (Any) -- Provider-native circuit object (result of translate_circuit).

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional provider-specific parameters (e.g. chip_id, auto_mapping, circuit_optimize).

返回:

Task ID assigned by the backend.

返回类型:

str

abstractmethod submit_batch(circuits, *, shots=1000, **kwargs)

Submit multiple circuits as a single batch.

参数:

• circuits (list[Any]) -- List of provider-native circuit objects.

• shots (int) -- Number of measurement shots.

• **kwargs (Any) -- Additional provider-specific parameters.

返回:

Task IDs (one per circuit), or a single task ID if the backend returns a group ID.

返回类型:

list[str]

abstractmethod translate_circuit(originir)

Translate an OriginIR circuit string to the provider's native circuit type.

参数:

originir (str) -- Circuit in OriginIR format.

返回:

Provider-specific circuit object.

返回类型:

Any