qpandalite.task.quafu package

Submodules

qpandalite.task.quafu.task module

BAQIS Quafu (ScQ) quantum cloud platform backend.

All HTTP communication is routed through QuafuAdapter in the adapters layer — the raw requests.post to the Quafu URL has been removed from this module.

Configuration is loaded from environment variables (preferred) or from quafu_online_config.json (deprecated fallback):

QUAFU_API_TOKEN: API token for Quafu

Public API:

• submit_task — Submit circuit(s) for execution on Quafu.

• query_by_taskid — Asynchronously query task status by task ID.

• query_by_taskid_sync — Synchronously query task status (blocking).

• query_task_by_group — Retrieve all tasks in a named group.

• query_all_tasks — Query all locally recorded tasks.

class qpandalite.task.quafu.task.Translation_OriginIR_to_QuafuCircuit

基类：object

Translate OriginIR circuits to Quafu QuantumCircuit objects.

Deprecated: use adapter.translate_circuit(originir) instead.

static reconstruct_qasm(qc, operation, qubit, cbit, parameter)

参数:

• operation (str | None)

• qubit (int | list[int])

• cbit (int | None)

• parameter (float | list[float] | None)

static translate(originir)

参数:

originir (str)

qpandalite.task.quafu.task.query_all_task(savepath=None)

Deprecated — use query_all_tasks() instead.

参数:

savepath (Path | str | None)

返回类型:

None

qpandalite.task.quafu.task.query_all_tasks(savepath=None)

Query all locally recorded Quafu tasks and cache results.

参数:

savepath (Path | str | None)

返回类型:

None

qpandalite.task.quafu.task.query_by_taskid(taskid, savepath=None)

Query task status by task ID (non-blocking).

参数:

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

• savepath (Path | str | None, optional) -- Directory path for saving task results locally. Defaults to ./quafu_online_info/.

返回:

Task status dictionary containing:

• "status": "success", "running", or "failed"

• "result": Task result data when status is "success", or error information when status is "failed"

返回类型:

dict[str, Any]

抛出:

ValueError -- If taskid is empty or has invalid type.

qpandalite.task.quafu.task.query_by_taskid_sync(taskid, interval=2.0, timeout=60.0, retry=5)

Query task status by task ID (blocking) until completion or timeout.

Continuously polls the Quafu platform until the task completes, fails, or the timeout is reached.

参数:

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

• interval (float, optional) -- Polling interval in seconds. Defaults to 2.0.

• timeout (float, optional) -- Maximum wait time in seconds. Defaults to 60.0.

• retry (int, optional) -- Number of retry attempts on query failure. Defaults to 5.

抛出:

• TimeoutError -- If the task does not complete within the specified timeout.

• RuntimeError -- If the task fails or all retry attempts are exhausted.

返回:

List of task result dictionaries when all tasks succeed.

返回类型:

list[dict[str, Any]]

qpandalite.task.quafu.task.query_task_by_group(group_name, history=None, verbose=True, savepath=None)

Retrieve all tasks belonging to a named Quafu group.

Queries the Quafu platform for all tasks associated with the specified group name and caches the results locally.

参数:

• group_name (str) -- The name of the task group to query.

• history (dict[str, list[str]] | None, optional) -- Cache of previously queried group tasks. If None, loads from local savepath.

• verbose (bool, optional) -- Whether to print detailed output. Defaults to True.

• savepath (Path | str | None, optional) -- Directory for saving results. Defaults to current working directory / "quafu_online_info".

抛出:

ValueError -- If group_name is empty or not a string.

返回:

List of task result objects from the Quafu platform.

返回类型:

list

qpandalite.task.quafu.task.query_task_by_group_sync(group_name, verbose=True, savepath=None, interval=2.0, timeout=60.0, retry=5)

Blocking query for all tasks in a named Quafu group.

Polls the Quafu platform until all tasks in the specified group have reached a terminal state (completed or failed), or until timeout.

参数:

• group_name (str) -- The name of the task group to query.

• verbose (bool, optional) -- Whether to print detailed output. Defaults to True.

• savepath (Path | str | None, optional) -- Directory for saving results. Defaults to current working directory / "quafu_online_info".

• interval (float, optional) -- Polling interval in seconds. Defaults to 2.0.

• timeout (float, optional) -- Maximum wait time in seconds. Defaults to 60.0.

• retry (int, optional) -- Number of retry attempts on query failure. Defaults to 5.

抛出:

TimeoutError -- If not all tasks complete within the specified timeout.

返回:

List of task result objects when all tasks are complete.

返回类型:

list

qpandalite.task.quafu.task.submit_task(circuit=None, task_name=None, chip_id=None, shots=10000, auto_mapping=True, savepath=None, group_name=None, **kwargs)

Submit one or more quantum circuits for execution on the Quafu platform.

参数:

• circuit (str | list[str] | None) -- OriginIR circuit string or list of strings.

• task_name (str | None) -- Human-readable task name.

• chip_id (str | None) -- Target chip ID (e.g. 'ScQ-P10').

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

• auto_mapping (bool) -- Whether to enable automatic qubit mapping / compilation.

• savepath (Path | str | None) -- Directory for local task records.

• group_name (str | None) -- Optional group name for batch submissions.

返回:

Task ID(s) for the submitted circuit(s).

返回类型:

str or list[str]

Module contents

BAQIS Quafu (ScQ) quantum cloud platform backend.